Switch from StrictYAML to Pydantic

RELEASE-NOTES.md

@@ -10,6 +10,8 @@
 ## Fixes and improvements
 * Adding `--image-name` option for image name argument in `spcs image-repository list-tags` for consistency with other commands.
 * Fixed errors during `spcs image-registry login` not being formatted correctly.
+* Project definition no longer accept extra fields. Any extra field will cause an error.
+* Project definition no longer accept extra fields. Any extra field will cause an error.
 
 # v2.1.0
 

pyproject.toml

@@ -25,6 +25,7 @@ dependencies = [
   "typer==0.9.0",
   "urllib3>=1.21.1,<2.3",
   "GitPython==3.1.42",
+  "pydantic==2.6.3"
 ]
 classifiers = [
   "Development Status :: 3 - Alpha",

src/snowflake/cli/api/commands/flags.py

@@ -374,7 +374,7 @@ def project_definition_option(project_name: str):
 
     def _callback(project_path: Optional[str]):
         dm = DefinitionManager(project_path)
-        project_definition = dm.project_definition.get(project_name)
+        project_definition = getattr(dm.project_definition, project_name, None)
         project_root = dm.project_root
 
         if not project_definition:

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

@@ -1,45 +1,36 @@
 from pathlib import Path
-from typing import Dict, List, Union
+from typing import Dict, List
 
+import yaml.loader
 from snowflake.cli.api.cli_global_context import cli_context
 from snowflake.cli.api.constants import DEFAULT_SIZE_LIMIT_MB
-from snowflake.cli.api.project.schemas.project_definition import (
-    project_override_schema,
-    project_schema,
-)
+from snowflake.cli.api.project.schemas.project_definition import ProjectDefinition
 from snowflake.cli.api.project.util import (
     append_to_identifier,
     clean_identifier,
     get_env_username,
     to_identifier,
 )
 from snowflake.cli.api.secure_path import SecurePath
-from strictyaml import (
-    YAML,
-    as_document,
-    load,
-)
+from yaml import load
 
 DEFAULT_USERNAME = "unknown_user"
 
 
-def merge_left(target: Union[Dict, YAML], source: Union[Dict, YAML]) -> None:
+def merge_left(target: Dict, source: Dict) -> None:
     """
     Recursively merges key/value pairs from source into target.
     Modifies the original dict-like "target".
     """
     for k, v in source.items():
-        if k in target and (
-            isinstance(v, dict) or (isinstance(v, YAML) and v.is_mapping())
-        ):
+        if k in target and isinstance(target[k], dict):
             # assumption: all inputs have been validated.
-            assert isinstance(target[k], dict) or isinstance(target[k], YAML)
             merge_left(target[k], v)
         else:
             target[k] = v
 
 
-def load_project_definition(paths: List[Path]) -> dict:
+def load_project_definition(paths: List[Path]) -> ProjectDefinition:
     """
     Loads project definition, optionally overriding values. Definition values
     are merged in left-to-right order (increasing precedence).
@@ -49,22 +40,23 @@ def load_project_definition(paths: List[Path]) -> dict:
         raise ValueError("Need at least one definition file.")
 
     with spaths[0].open("r", read_file_limit_mb=DEFAULT_SIZE_LIMIT_MB) as base_yml:
-        definition = load(base_yml.read(), project_schema)
+        definition = load(base_yml.read(), Loader=yaml.loader.BaseLoader)
 
     for override_path in spaths[1:]:
         with override_path.open(
             "r", read_file_limit_mb=DEFAULT_SIZE_LIMIT_MB
         ) as override_yml:
-            overrides = load(override_yml.read(), project_override_schema)
+            overrides = load(override_yml.read(), Loader=yaml.loader.BaseLoader)
             merge_left(definition, overrides)
 
         # TODO: how to show good error messages here?
-        definition.revalidate(project_schema)
 
-    return definition.data
+    return ProjectDefinition(**definition)
 
 
-def generate_local_override_yml(project: Union[Dict, YAML]) -> YAML:
+def generate_local_override_yml(
+    project: ProjectDefinition,
+) -> ProjectDefinition:
     """
     Generates defaults for optional keys in the same YAML structure as the project
     schema. The returned YAML object can be saved directly to a file, if desired.
@@ -76,8 +68,8 @@ def generate_local_override_yml(project: Union[Dict, YAML]) -> YAML:
     warehouse = conn.warehouse
 
     local: dict = {}
-    if "native_app" in project:
-        name = clean_identifier(project["native_app"]["name"])
+    if project.native_app:
+        name = clean_identifier(project.native_app.name)
         app_identifier = to_identifier(name)
         user_app_identifier = append_to_identifier(app_identifier, f"_{user}")
         package_identifier = append_to_identifier(app_identifier, f"_pkg_{user}")
@@ -90,8 +82,12 @@ def generate_local_override_yml(project: Union[Dict, YAML]) -> YAML:
             },
             "package": {"name": package_identifier, "role": role},
         }
+    # TODO: this is an ugly workaround, because pydantics BaseModel.model_copy(update=) doesn't work properly
+    # After fixing UpdatableModel.update_from_dict it should be used here
+    target_definition = project.model_dump()
+    merge_left(target_definition, local)
 
-    return as_document(local, project_override_schema)
+    return ProjectDefinition(**target_definition)
 
 
 def default_app_package(project_name: str):

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

@@ -7,6 +7,7 @@
 
 from snowflake.cli.api.exceptions import MissingConfiguration
 from snowflake.cli.api.project.definition import load_project_definition
+from snowflake.cli.api.project.schemas.project_definition import ProjectDefinition
 
 
 def _compat_is_mount(path: Path):
@@ -100,5 +101,5 @@ def _user_definition_file_if_available(project_path: Path) -> Optional[Path]:
         )
 
     @functools.cached_property
-    def project_definition(self) -> dict:
+    def project_definition(self) -> ProjectDefinition:
         return load_project_definition(self._project_config_paths)

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

@@ -0,0 +1,24 @@
+from textwrap import dedent
+
+from pydantic import ValidationError
+
+
+class SchemaValidationError(Exception):
+    generic_message = "For field {loc} you provided '{loc}'. This caused: {msg}"
+    message_templates = {
+        "string_type": "{msg} for field '{loc}', you provided '{input}'",
+        "extra_forbidden": "{msg}. You provided field '{loc}' with value '{input}' that is not present in the schema",
+        "missing": "Your project definition is missing following fields: {loc}",
+    }
+
+    def __init__(self, error: ValidationError):
+        errors = error.errors()
+        message = f"During evaluation of {error.title} schema following errors were encoutered:\n"
+        message += "\n".join(
+            [
+                self.message_templates.get(e["type"], self.generic_message).format(**e)
+                for e in errors
+            ]
+        )
+
+        super().__init__(dedent(message))

src/snowflake/cli/api/project/schemas/native_app/application.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from typing import Literal, Optional
+
+from pydantic import Field
+from snowflake.cli.api.project.schemas.updatable_model import (
+    IdentifierField,
+    UpdatableModel,
+)
+
+
+class Application(UpdatableModel):
+    role: Optional[str] = Field(
+        title="Role to use when creating the application instance and consumer-side objects",
+        default=None,
+    )
+    name: Optional[str] = Field(
+        title="Name of the application created when you run the snow app run command",
+        default=None,
+    )
+    warehouse: Optional[str] = IdentifierField(
+        title="Name of the application created when you run the snow app run command",
+        default=None,
+    )
+    debug: Optional[bool] = Field(
+        title="Whether to enable debug mode when using a named stage to create an application",
+        default=True,
+    )
+
+
+DistributionOptions = Literal["internal", "external", "INTERNAL", "EXTERNAL"]

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

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import re
+from typing import List, Optional, Union
+
+from pydantic import Field, field_validator
+from snowflake.cli.api.project.schemas.native_app.application import Application
+from snowflake.cli.api.project.schemas.native_app.package import Package
+from snowflake.cli.api.project.schemas.native_app.path_mapping import PathMapping
+from snowflake.cli.api.project.schemas.updatable_model import UpdatableModel
+from snowflake.cli.api.project.util import (
+    SCHEMA_AND_NAME,
+)
+
+
+class NativeApp(UpdatableModel):
+    name: str = Field(
+        title="Project identifier",
+    )
+    artifacts: List[Union[PathMapping, str]] = Field(
+        title="List of file source and destination pairs to add to the deploy root",
+    )
+    deploy_root: Optional[str] = Field(
+        title="Folder at the root of your project where the build step copies the artifacts.",
+        default="output/deploy/",
+    )
+    source_stage: Optional[str] = Field(
+        title="Identifier of the stage that stores the application artifacts.",
+        default="app_src.stage",
+    )
+    package: Optional[Package] = Field(title="PackageSchema", default=None)
+    application: Optional[Application] = Field(title="Application info", default=None)
+
+    @field_validator("source_stage")
+    @classmethod
+    def validate_source_stage(cls, input_value: str):
+        if not re.match(SCHEMA_AND_NAME, input_value):
+            raise ValueError("Incorrect value for Native Apps source stage value")
+        return input_value

src/snowflake/cli/api/project/schemas/native_app/package.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from typing import List, Optional
+
+from pydantic import Field, field_validator
+from snowflake.cli.api.project.schemas.native_app.application import DistributionOptions
+from snowflake.cli.api.project.schemas.updatable_model import (
+    IdentifierField,
+    UpdatableModel,
+)
+
+
+class Package(UpdatableModel):
+    scripts: Optional[List[str]] = Field(
+        title="List of SQL file paths relative to the project root", default=None
+    )
+    role: Optional[str] = IdentifierField(
+        title="Role to use when creating the application package and provider-side objects",
+        default=None,
+    )
+    name: Optional[str] = IdentifierField(
+        title="Name of the application created when you run the snow app run command",  # TODO: this description seems duplicated, is it ok?
+        default=None,
+    )
+    warehouse: Optional[str] = IdentifierField(
+        title="Warehouse used to run the scripts", default=None
+    )
+    distribution: Optional[DistributionOptions] = Field(
+        title="Distribution of the application package created by the Snowflake CLI",
+        default="internal",
+    )
+
+    @field_validator("scripts")
+    @classmethod
+    def validate_scripts(cls, input_list):
+        if len(input_list) != len(set(input_list)):
+            raise ValueError(
+                "Scripts field should contain unique values. Check the list for duplicates and try again"
+            )
+        return input_list

src/snowflake/cli/api/project/schemas/native_app/path_mapping.py

@@ -0,0 +1,10 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from snowflake.cli.api.project.schemas.updatable_model import UpdatableModel
+
+
+class PathMapping(UpdatableModel):
+    src: str
+    dest: Optional[str] = None

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

@@ -1,23 +1,25 @@
 from __future__ import annotations
 
-from snowflake.cli.api.project.schemas import (
-    native_app,
-    snowpark,
-    streamlit,
-)
-from snowflake.cli.api.project.schemas.relaxed_map import RelaxedMap
-from strictyaml import (
-    Int,
-    Optional,
-)
+from typing import Optional
 
-project_schema = RelaxedMap(
-    {
-        "definition_version": Int(),
-        Optional("native_app"): native_app.native_app_schema,
-        Optional("snowpark"): snowpark.snowpark_schema,
-        Optional("streamlit"): streamlit.streamlit_schema,
-    }
-)
+from pydantic import Field
+from snowflake.cli.api.project.schemas.native_app.native_app import NativeApp
+from snowflake.cli.api.project.schemas.snowpark.snowpark import Snowpark
+from snowflake.cli.api.project.schemas.streamlit.streamlit import Streamlit
+from snowflake.cli.api.project.schemas.updatable_model import UpdatableModel
 
-project_override_schema = project_schema.as_fully_optional()
+
+class ProjectDefinition(UpdatableModel):
+    definition_version: int = Field(
+        title="Version of the project definition schema, which is currently 1"
+    )
+    native_app: Optional[NativeApp] = Field(
+        title="Native app definitions for the project", default=None
+    )
+    snowpark: Optional[Snowpark] = Field(
+        title="Snowpark functions and procedures definitions for the project",
+        default=None,
+    )
+    streamlit: Optional[Streamlit] = Field(
+        title="Streamlit definitions for the project", default=None
+    )