feat: [SNOW-1942875] initial version of cicd templates in snow init

Author: sfc-gh-jwilkowski

src/snowflake/cli/_plugins/cicd/manager.py

@@ -1,5 +1,5 @@
 import enum
-from typing import Generator, List, Optional, Type
+from typing import List, Optional, Type
 
 from snowflake.cli.api.secure_path import SecurePath
 
@@ -11,7 +11,6 @@ class CIProviderChoices(str, enum.Enum):
 
 class CIProvider:
     name: str
-    files_to_render_directories: list[str] = []
 
     @classmethod
     def cleanup(cls, root: SecurePath) -> None:
@@ -31,18 +30,15 @@ def from_choice(cls, choice: CIProviderChoices) -> "CIProvider":
     def all(cls) -> List[Type["CIProvider"]]:  # noqa: A003
         return [GithubProvider, GitLabProvider]
 
-    def get_files_to_render(self, root_dir: SecurePath) -> Generator[str, None, None]:
-        for directory in self.files_to_render_directories:
-            for path in (root_dir / directory).rglob("*"):
-                yield str(path.relative_to(root_dir.path))
-
     def has_template(self, root_dir: SecurePath) -> bool:
         raise NotImplementedError()
 
+    def copy(self, source: SecurePath, destination: SecurePath):
+        raise NotImplementedError()
+
 
 class GithubProvider(CIProvider):
     name = CIProviderChoices.GITLAB.name
-    files_to_render_directories = [".github/workflows/"]
 
     @classmethod
     def cleanup(cls, root_dir: SecurePath):
@@ -51,6 +47,9 @@ def cleanup(cls, root_dir: SecurePath):
     def has_template(self, root_dir: SecurePath) -> bool:
         return (root_dir / ".github/workflows").exists()
 
+    def copy(self, source: SecurePath, destination: SecurePath) -> None:
+        (source / ".github").copy(destination.path, dirs_exist_ok=True)
+
 
 class GitLabProvider(CIProvider):
     name = CIProviderChoices.GITLAB.name
@@ -59,13 +58,14 @@ class GitLabProvider(CIProvider):
     def cleanup(cls, root_dir: SecurePath):
         (root_dir / ".gitlab-ci.yml").unlink(missing_ok=True)
 
-    def get_files_to_render(self, root_dir: SecurePath):
-        if (root_dir / ".gitlab-ci.yml").exists():
-            return [".gitlab-ci.yml"]
-
     def has_template(self, root_dir: SecurePath) -> bool:
         return (root_dir / ".gitlab-ci.yml").exists()
 
+    def copy(self, source: SecurePath, destination: SecurePath) -> None:
+        if (destination / ".gitlab-ci.yml").exists():
+            (destination / ".gitlab-ci.yml").unlink()
+        (source / ".gitlab-ci.yml").move(destination.path)
+
 
 class CIProviderManager:
     @staticmethod

src/snowflake/cli/_plugins/init/commands.py

@@ -47,7 +47,6 @@
 
 
 DEFAULT_SOURCE = "https://github.com/snowflakedb/snowflake-cli-templates"
-DEFAULT_CI_SOURCE = DEFAULT_SOURCE + "/cicd"
 
 log = logging.getLogger(__name__)
 
@@ -220,51 +219,39 @@ def init(
     variables_from_flags = {
         v.key: v.value for v in parse_key_value_variables(variables)
     }
-    is_remote = any(
-        template_source.startswith(prefix) for prefix in ["git@", "http://", "https://"]  # type: ignore
-    )
     args_error_msg = f"Check whether {TemplateOption.param_decls[0]} and {SourceOption.param_decls[0]} arguments are correct."
 
     # copy/download template into tmpdir, so it is going to be removed in case command ends with an error
     with SecurePath.temporary_directory() as tmpdir:
-        if is_remote:
-            template_root = _fetch_remote_template(
-                url=template_source, path=template, destination=tmpdir  # type: ignore
-            )
-        else:
-            template_root = _fetch_local_template(
-                template_source=SecurePath(template_source),
-                path=template,
-                destination=tmpdir,
-            )
+        assert isinstance(template_source, str)
+        template_root = _fetch_template(template_source, template, tmpdir)
+
+        template_metadata = _read_template_metadata(
+            template_root, args_error_msg=args_error_msg
+        )
+        if template_metadata.minimum_cli_version:
+            _validate_cli_version(template_metadata.minimum_cli_version)
 
         if ci_provider:
             ci_provider_instance = CIProvider.from_choice(ci_provider)
             if ci_template_source is not None:
                 with SecurePath.temporary_directory() as cicd_tmpdir:
-                    cicd_template_root = _fetch_remote_template(
-                        url=ci_template_source,
-                        path=None,
-                        destination=cicd_tmpdir
-                        # type: ignore
+                    cicd_template_root = _fetch_template(
+                        ci_template_source, None, cicd_tmpdir
+                    )
+                    ci_provider_instance.copy(cicd_template_root, template_root)
+                    ci_template_metadata = _read_template_metadata(
+                        cicd_template_root,
+                        args_error_msg="template.yml is required for --ci-template-source.",
                     )
-                    cicd_template_root.copy(template_root.path)
+                    template_metadata += ci_template_metadata
 
             elif ci_provider_instance.has_template(template_root):
                 pass  # template has ci files
             else:
-                # use generic ci/cd template
-                with SecurePath.temporary_directory() as cicd_tmpdir:
-                    cicd_template_root = _fetch_remote_template(
-                        url=DEFAULT_SOURCE, path=f"cicd/{ci_provider_instance.name.lower()}", destination=cicd_tmpdir  # type: ignore
-                    )
-                    cicd_template_root.copy(template_root.path)
-
-        template_metadata = _read_template_metadata(
-            template_root, args_error_msg=args_error_msg
-        )
-        if template_metadata.minimum_cli_version:
-            _validate_cli_version(template_metadata.minimum_cli_version)
+                raise ClickException(
+                    f"Template for {ci_provider_instance.name} not provided and not configured on selected template."
+                )
 
         variable_values = _determine_variable_values(
             variables_metadata=template_metadata.variables,
@@ -274,13 +261,12 @@ def init(
             "project_dir_name": SecurePath(path).name,
             "snowflake_cli_version": VERSION,
         }
-        files_to_render = template_metadata.files_to_render + list(
-            ci_provider_instance.get_files_to_render(template_root)
+        log.debug(
+            "Rendering template files: %s", ", ".join(template_metadata.files_to_render)
         )
-        log.debug("Rendering template files: %s", ", ".join(files_to_render))
         render_template_files(
             template_root=template_root,
-            files_to_render=files_to_render,
+            files_to_render=template_metadata.files_to_render,
             data=variable_values,
         )
         _remove_template_metadata_file(template_root)
@@ -291,5 +277,27 @@ def init(
     return MessageResult(f"Initialized the new project in {path}")
 
 
+def _fetch_template(
+    template_source: str, template: Optional[str], tmpdir: SecurePath
+) -> SecurePath:
+    if _is_remote_source(template_source):
+        template_root = _fetch_remote_template(
+            url=template_source, path=template, destination=tmpdir  # type: ignore
+        )
+    else:
+        template_root = _fetch_local_template(
+            template_source=SecurePath(template_source),
+            path=template,
+            destination=tmpdir,
+        )
+    return template_root
+
+
+def _is_remote_source(template_source: str) -> bool:
+    return any(
+        template_source.startswith(prefix) for prefix in ["git@", "http://", "https://"]  # type: ignore
+    )
+
+
 def post_generate(template_root: SecurePath, ci_provider: Optional[CIProvider]):
     CIProviderManager.project_post_gen_cleanup(ci_provider, template_root)

src/snowflake/cli/api/project/schemas/template.py

@@ -18,12 +18,14 @@
 
 import typer
 from click import ClickException
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 from snowflake.cli.api.exceptions import InvalidTemplate
 from snowflake.cli.api.secure_path import SecurePath
 
 
 class TemplateVariable(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
     name: str = Field(..., title="Variable identifier")
     type: Optional[Literal["string", "float", "int"]] = Field(  # noqa: A003
         title="Type of the variable", default=None
@@ -64,6 +66,32 @@ def __init__(self, template_root: SecurePath, **kwargs):
         super().__init__(**kwargs)
         self._validate_files_exist(template_root)
 
+    def __add__(self, other: Template):
+        errors = []
+        if not isinstance(other, Template):
+            raise ClickException(f"Can not merge template with {type(other)}")
+
+        if self.minimum_cli_version != other.minimum_cli_version:
+            errors.append(
+                f"minimum_cli_versions do not match: {self.minimum_cli_version} != {other.minimum_cli_version}"
+            )
+        variable_map = {variable.name: variable for variable in self.variables}
+        for other_variable in other.variables:
+            if self_variable := variable_map.get(other_variable.name):
+                for attr in ["type", "prompt", "default"]:
+                    if getattr(self_variable, attr) != getattr(other_variable, attr):
+                        errors.append(
+                            f"Conflicting variable definitions: '{self_variable.name}' has different values for attribute '{attr}': '{getattr(self_variable, attr)}' != '{getattr(other_variable, attr)}'"
+                        )
+        if errors:
+            error_str = "\n\t" + "\n\t".join(error for error in errors)
+            raise ClickException(
+                f"Could not merge templates. Following errors found:{error_str}"
+            )
+        self.files_to_render = list(set(self.files_to_render + other.files_to_render))
+        self.variables = list(set(self.variables + other.variables))
+        return self
+
     def _validate_files_exist(self, template_root: SecurePath) -> None:
         for path_in_template in self.files_to_render:
             full_path = template_root / path_in_template

src/snowflake/cli/api/secure_path.py

@@ -116,14 +116,6 @@ def glob(self, pattern: str):
         """
         return self._path.glob(pattern)
 
-    def rglob(self, pattern: str) -> Generator[Path, None, None]:
-        """
-        Recursively yield all existing files (of any kind, including
-        directories) matching the given relative pattern, anywhere in
-        this subtree.
-        """
-        return self._path.rglob(pattern)
-
     def as_posix(self) -> str:
         """
         Return the string representation of the path with forward slashes (/) as the path separator.