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
 
@@ -10,39 +10,32 @@ class CIProviderChoices(str, enum.Enum):
 
 
 class CIProvider:
-    name: str
-    files_to_render_directories: list[str] = []
+    NAME: str
 
     @classmethod
     def cleanup(cls, root: SecurePath) -> None:
         raise NotImplementedError()
 
-    def __eq__(self, other) -> bool:
-        return self.name == other.name
-
     @classmethod
     def from_choice(cls, choice: CIProviderChoices) -> "CIProvider":
         return {
-            CIProviderChoices.GITHUB: GithubProvider,
-            CIProviderChoices.GITLAB: GitLabProvider,
-        }[choice]()
+            GithubProvider.NAME: GithubProvider,
+            GitLabProvider.NAME: GitLabProvider,
+        }[choice.name]()
 
     @classmethod
     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/"]
+    NAME = CIProviderChoices.GITHUB.name
 
     @classmethod
     def cleanup(cls, root_dir: SecurePath):
@@ -51,21 +44,25 @@ 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
+    NAME = CIProviderChoices.GITLAB.name
 
     @classmethod
     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,52 +219,30 @@ 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,
-            )
-
-        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.copy(template_root.path)
-
-            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)
+        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)
+            clone(
+                ci_provider_instance,
+                ci_template_source,
+                template_metadata,
+                template_root,
+            )
+        else:
+            ci_provider_instance = None
+
         variable_values = _determine_variable_values(
             variables_metadata=template_metadata.variables,
             variables_from_flags=variables_from_flags,
@@ -274,13 +251,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 +267,51 @@ def init(
     return MessageResult(f"Initialized the new project in {path}")
 
 
+def clone(
+    ci_provider_instance: CIProvider,
+    ci_template_source: Optional[str],
+    template_metadata: Template,
+    template_root: SecurePath,
+):
+    if ci_template_source is not None:
+        with SecurePath.temporary_directory() as cicd_tmpdir:
+            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.",
+            )
+            template_metadata.merge(ci_template_metadata)
+
+    elif ci_provider_instance.has_template(template_root):
+        pass  # template has ci files
+    else:
+        raise ClickException(
+            f"Template for {ci_provider_instance.NAME} not provided and not configured on selected template."
+        )
+
+
+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 merge(self, other: Template):
+        if not isinstance(other, Template):
+            raise ClickException(f"Can not merge template with {type(other)}")
+
+        errors = []
+        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.