Merge pull request #294 from zanebclark/fix/kw_only_get_alphanum_key_…

…fixes

Fix: Remove kw_only and gracefully handle missing max version number

schemachange/config/BaseConfig.py

@@ -18,7 +18,7 @@
 T = TypeVar("T", bound="BaseConfig")
 
 
-@dataclasses.dataclass(frozen=True, kw_only=True)
+@dataclasses.dataclass(frozen=True)
 class BaseConfig(ABC):
     default_config_file_name: ClassVar[str] = "schemachange-config.yml"
 

schemachange/config/DeployConfig.py

@@ -9,7 +9,7 @@
 from schemachange.config.utils import get_snowflake_identifier_string
 
 
-@dataclasses.dataclass(frozen=True, kw_only=True)
+@dataclasses.dataclass(frozen=True)
 class DeployConfig(BaseConfig):
     subcommand: Literal["deploy"] = "deploy"
     snowflake_account: str | None = None

schemachange/config/RenderConfig.py

@@ -8,10 +8,10 @@
 from schemachange.config.utils import validate_file_path
 
 
-@dataclasses.dataclass(frozen=True, kw_only=True)
+@dataclasses.dataclass(frozen=True)
 class RenderConfig(BaseConfig):
+    script_path: Path | None = None
     subcommand: Literal["render"] = "render"
-    script_path: Path
 
     @classmethod
     def factory(
@@ -31,3 +31,9 @@ def factory(
             script_path=validate_file_path(file_path=script_path),
             **kwargs,
         )
+
+    def __post_init__(self):
+        if self.script_path is None:
+            raise TypeError(
+                "RenderConfig is missing 1 required argument: 'script_path'"
+            )

schemachange/deploy.py

@@ -23,7 +23,9 @@ def alphanum_convert(text: str):
 # Each number is converted to and integer and string parts are left as strings
 # This will enable correct sorting in python when the lists are compared
 # e.g. get_alphanum_key('1.2.2') results in ['', 1, '.', 2, '.', 2, '']
-def get_alphanum_key(key):
+def get_alphanum_key(key: str | int | None) -> list:
+    if key == "" or key is None:
+        return []
     alphanum_key = [alphanum_convert(c) for c in re.split("([0-9]+)", key)]
     return alphanum_key
 
@@ -100,7 +102,7 @@ def deploy(config: DeployConfig, session: SnowflakeSession):
             script_metadata = versioned_scripts.get(script.name)
 
             if (
-                max_published_version != ""
+                max_published_version is not None
                 and get_alphanum_key(script.version) <= max_published_version
             ):
                 if script_metadata is None:
@@ -113,7 +115,7 @@ def deploy(config: DeployConfig, session: SnowflakeSession):
                 else:
                     script_log.debug(
                         "Script has already been applied",
-                        max_published_version=str(max_published_version),
+                        max_published_version=max_published_version,
                     )
                     if script_metadata["checksum"] != checksum_current:
                         script_log.info("Script checksum has drifted since application")

schemachange/session/Credential.py

@@ -2,7 +2,6 @@
 
 import dataclasses
 import os
-from abc import ABC
 from typing import Literal, Union
 
 import structlog
@@ -14,37 +13,32 @@
 )
 
 
-@dataclasses.dataclass(kw_only=True, frozen=True)
-class Credential(ABC):
-    authenticator: str
-
-
-@dataclasses.dataclass(kw_only=True, frozen=True)
-class OauthCredential(Credential):
-    authenticator: Literal["oauth"] = "oauth"
+@dataclasses.dataclass(frozen=True)
+class OauthCredential:
     token: str
+    authenticator: Literal["oauth"] = "oauth"
 
 
-@dataclasses.dataclass(kw_only=True, frozen=True)
-class PasswordCredential(Credential):
-    authenticator: Literal["snowflake"] = "snowflake"
+@dataclasses.dataclass(frozen=True)
+class PasswordCredential:
     password: str
+    authenticator: Literal["snowflake"] = "snowflake"
 
 
-@dataclasses.dataclass(kw_only=True, frozen=True)
-class PrivateKeyCredential(Credential):
-    authenticator: Literal["snowflake"] = "snowflake"
+@dataclasses.dataclass(frozen=True)
+class PrivateKeyCredential:
     private_key: bytes
+    authenticator: Literal["snowflake"] = "snowflake"
 
 
-@dataclasses.dataclass(kw_only=True, frozen=True)
-class ExternalBrowserCredential(Credential):
-    authenticator: Literal["externalbrowser"] = "externalbrowser"
+@dataclasses.dataclass(frozen=True)
+class ExternalBrowserCredential:
     password: str | None = None
+    authenticator: Literal["externalbrowser"] = "externalbrowser"
 
 
-@dataclasses.dataclass(kw_only=True, frozen=True)
-class OktaCredential(Credential):
+@dataclasses.dataclass(frozen=True)
+class OktaCredential:
     authenticator: str
     password: str
 

schemachange/session/Script.py

@@ -18,7 +18,7 @@
 T = TypeVar("T", bound="Script")
 
 
-@dataclasses.dataclass(kw_only=True, frozen=True)
+@dataclasses.dataclass(frozen=True)
 class Script(ABC):
     pattern: ClassVar[Pattern[str]]
     type: ClassVar[Literal["V", "R", "A"]]
@@ -47,7 +47,7 @@ def from_path(cls, file_path: Path, **kwargs) -> T:
         )
 
 
-@dataclasses.dataclass(kw_only=True, frozen=True)
+@dataclasses.dataclass(frozen=True)
 class VersionedScript(Script):
     pattern: ClassVar[re.Pattern[str]] = re.compile(
         r"^(V)(?P<version>.+?)?__(?P<description>.+?)\.", re.IGNORECASE
@@ -64,15 +64,15 @@ def from_path(cls: T, file_path: Path, **kwargs) -> T:
         )
 
 
-@dataclasses.dataclass(kw_only=True, frozen=True)
+@dataclasses.dataclass(frozen=True)
 class RepeatableScript(Script):
     pattern: ClassVar[re.Pattern[str]] = re.compile(
         r"^(R)__(?P<description>.+?)\.", re.IGNORECASE
     )
     type: ClassVar[Literal["R"]] = "R"
 
 
-@dataclasses.dataclass(kw_only=True, frozen=True)
+@dataclasses.dataclass(frozen=True)
 class AlwaysScript(Script):
     pattern: ClassVar[re.Pattern[str]] = re.compile(
         r"^(A)__(?P<description>.+?)\.", re.IGNORECASE

schemachange/session/SnowflakeSession.py

@@ -209,11 +209,7 @@ def get_script_metadata(
 
         self.logger.info(
             "Max applied change script version %(max_published_version)s"
-            % {
-                "max_published_version": max_published_version
-                if max_published_version != ""
-                else "None"
-            }
+            % {"max_published_version": max_published_version}
         )
         return change_history, r_scripts_checksum, max_published_version
 
@@ -251,10 +247,10 @@ def fetch_versioned_scripts(
 
         # Collect all the results into a list
         versioned_scripts: dict[str, dict[str, str | int]] = defaultdict(dict)
-        versions: list[str | int] = []
+        versions: list[str | int | None] = []
         for cursor in results:
             for version, script, checksum in cursor:
-                versions.append(version)
+                versions.append(version if version != "" else None)
                 versioned_scripts[script] = {
                     "version": version,
                     "script": script,

tests/test_cli_misc.py

@@ -30,7 +30,11 @@ def test_alphanum_convert_given__lowercase():
 
 
 def test_get_alphanum_key_given__empty_string():
-    assert get_alphanum_key("") == [""]
+    assert get_alphanum_key("") == []
+
+
+def test_get_alphanum_key_given__none():
+    assert get_alphanum_key(None) == []
 
 
 def test_get_alphanum_key_given__numbers_only():