Introduce dynamic processor registration

src/snowflake/cli/_plugins/nativeapp/codegen/compiler.py

@@ -33,23 +33,14 @@
 from snowflake.cli._plugins.nativeapp.codegen.templates.templates_processor import (
     TemplatesProcessor,
 )
-from snowflake.cli._plugins.nativeapp.feature_flags import FeatureFlag
 from snowflake.cli.api.cli_global_context import get_cli_context
 from snowflake.cli.api.console import cli_console as cc
 from snowflake.cli.api.metrics import CLICounterField
 from snowflake.cli.api.project.schemas.v1.native_app.path_mapping import (
     ProcessorMapping,
 )
 
-SNOWPARK_PROCESSOR = "snowpark"
-NA_SETUP_PROCESSOR = "native app setup"
-TEMPLATES_PROCESSOR = "templates"
-
-_REGISTERED_PROCESSORS_BY_NAME = {
-    SNOWPARK_PROCESSOR: SnowparkAnnotationProcessor,
-    NA_SETUP_PROCESSOR: NativeAppSetupProcessor,
-    TEMPLATES_PROCESSOR: TemplatesProcessor,
-}
+ProcessorClassType = type[ArtifactProcessor]
 
 
 class NativeAppCompiler:
@@ -66,10 +57,28 @@ def __init__(
         bundle_ctx: BundleContext,
     ):
         self._assert_absolute_paths(bundle_ctx)
+        self._processor_classes_by_name: Dict[str, ProcessorClassType] = {}
         self._bundle_ctx = bundle_ctx
         # dictionary of all processors created and shared between different artifact objects.
         self.cached_processors: Dict[str, ArtifactProcessor] = {}
 
+        self._register(SnowparkAnnotationProcessor)
+        self._register(NativeAppSetupProcessor)
+        self._register(TemplatesProcessor)
+
+    def _register(self, processor_cls: ProcessorClassType):
+        """
+        Registers a processor class to enable.
+        """
+
+        name = getattr(processor_cls, "NAME", None)
+        assert name is not None
+
+        if name in self._processor_classes_by_name:
+            raise ValueError(f"Processor {name} is already registered")
+
+        self._processor_classes_by_name[str(name)] = processor_cls
+
     @staticmethod
     def _assert_absolute_paths(bundle_ctx: BundleContext):
         for name in ["Project", "Deploy", "Bundle", "Generated"]:
@@ -128,8 +137,8 @@ def _try_create_processor(
         if current_processor is not None:
             return current_processor
 
-        processor_factory = _REGISTERED_PROCESSORS_BY_NAME.get(processor_name)
-        if processor_factory is None:
+        processor_cls = self._processor_classes_by_name.get(processor_name)
+        if processor_cls is None:
             # No registered processor with the specified name
             return None
 
@@ -141,7 +150,7 @@ def _try_create_processor(
         processor_ctx.generated_root = (
             self._bundle_ctx.generated_root / processor_subdirectory
         )
-        current_processor = processor_factory(processor_ctx)
+        current_processor = processor_cls(processor_ctx)
         self.cached_processors[processor_name] = current_processor
 
         return current_processor
@@ -154,6 +163,22 @@ def _should_invoke_processors(self):
         return False
 
     def _is_enabled(self, processor: ProcessorMapping) -> bool:
-        if processor.name.lower() == NA_SETUP_PROCESSOR:
-            return FeatureFlag.ENABLE_NATIVE_APP_PYTHON_SETUP.is_enabled()
-        return True
+        """
+        Determines is a process is enabled. All processors are considered enabled
+        unless they are explicitly disabled, typically via a feature flag.
+        """
+        processor_name = processor.name.lower()
+        processor_cls = self._processor_classes_by_name.get(processor_name)
+        if processor_cls is None:
+            # Unknown processor, consider it enabled, even though trying to
+            # invoke it later will raise an exception
+            return True
+
+        # if the processor class defines a static method named "is_enabled", then
+        # call it. Otherwise, it's considered enabled by default.
+        is_enabled_fn = getattr(processor_cls, "is_enabled", None)
+        if is_enabled_fn is None:
+            # No custom static method provided, enabled by default
+            return True
+
+        return is_enabled_fn()

src/snowflake/cli/_plugins/nativeapp/codegen/setup/native_app_setup_processor.py

@@ -36,6 +36,7 @@
     SandboxEnvBuilder,
     execute_script_in_sandbox,
 )
+from snowflake.cli._plugins.nativeapp.feature_flags import FeatureFlag
 from snowflake.cli._plugins.stage.diff import to_stage_path
 from snowflake.cli.api.console import cli_console as cc
 from snowflake.cli.api.project.schemas.v1.native_app.path_mapping import (
@@ -74,9 +75,15 @@ def safe_set(d: dict, *keys: str, **kwargs) -> None:
 
 
 class NativeAppSetupProcessor(ArtifactProcessor):
+    NAME = "native app setup"
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
+    @staticmethod
+    def is_enabled() -> bool:
+        return FeatureFlag.ENABLE_NATIVE_APP_PYTHON_SETUP.is_enabled()
+
     def process(
         self,
         artifact_to_process: PathMapping,

src/snowflake/cli/_plugins/nativeapp/codegen/snowpark/python_processor.py

@@ -164,6 +164,8 @@ class SnowparkAnnotationProcessor(ArtifactProcessor):
     and generate SQL code for creation of extension functions based on those discovered objects.
     """
 
+    NAME = "snowpark"
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 

src/snowflake/cli/_plugins/nativeapp/codegen/templates/templates_processor.py

@@ -49,6 +49,8 @@ class TemplatesProcessor(ArtifactProcessor):
     Processor class to perform template expansion on all relevant artifacts (specified in the project definition file).
     """
 
+    NAME = "templates"
+
     def expand_templates_in_file(
         self, src: Path, dest: Path, template_context: dict[str, Any] | None = None
     ) -> None:

src/snowflake/cli/api/project/definition_conversion.py

@@ -13,7 +13,6 @@
     bundle_artifacts,
 )
 from snowflake.cli._plugins.nativeapp.bundle_context import BundleContext
-from snowflake.cli._plugins.nativeapp.codegen.compiler import TEMPLATES_PROCESSOR
 from snowflake.cli._plugins.nativeapp.codegen.templates.templates_processor import (
     TemplatesProcessor,
 )
@@ -457,7 +456,7 @@ def _convert_templates_in_files(
             artifact
             for artifact in pkg_model.artifacts
             for processor in artifact.processors
-            if processor.name == TEMPLATES_PROCESSOR
+            if processor.name.lower() == TemplatesProcessor.NAME
         ]
         if not in_memory and artifacts_to_template:
             metrics.set_counter(CLICounterField.TEMPLATES_PROCESSOR, 1)

tests_integration/helpers/test_v1_to_v2.py

@@ -2,7 +2,9 @@
 
 import pytest
 
-from snowflake.cli._plugins.nativeapp.codegen.compiler import TEMPLATES_PROCESSOR
+from snowflake.cli._plugins.nativeapp.codegen.templates.templates_processor import (
+    TemplatesProcessor,
+)
 from tests.nativeapp.factories import ProjectV11Factory
 
 
@@ -44,10 +46,10 @@ def test_v1_to_v2_converts_templates_in_files(temp_dir, runner):
         pdf__native_app__package__name="my_pkg",
         pdf__native_app__application__name="my_app",
         pdf__native_app__artifacts=[
-            dict(src="templated.txt", processors=[TEMPLATES_PROCESSOR]),
+            dict(src="templated.txt", processors=[TemplatesProcessor.NAME]),
             dict(src="untemplated.txt"),
-            dict(src="app/*", processors=[TEMPLATES_PROCESSOR]),
-            dict(src="nested/*", processors=[TEMPLATES_PROCESSOR]),
+            dict(src="app/*", processors=[TemplatesProcessor.NAME]),
+            dict(src="nested/*", processors=[TemplatesProcessor.NAME]),
         ],
         files={
             filename: source_contents