refactor: move safety functions to safety.py

t4_devkit/sanity/context.py

@@ -5,20 +5,21 @@
 from attrs import define
 from returns.maybe import Maybe
 from returns.pipeline import is_successful
-from returns.result import Result, safe
 from typing_extensions import Self
 
-from t4_devkit import DBMetadata, load_metadata
+from t4_devkit import DBMetadata
 from t4_devkit.schema.name import SchemaName
 
+from .safety import load_metadata_safe
+
 
 @define
 class SanityContext:
     metadata: Maybe[DBMetadata]
 
     @classmethod
     def from_path(cls, data_root: str, revision: str | None = None) -> Self:
-        metadata_result = _load_metadata_safe(data_root, revision=revision)
+        metadata_result = load_metadata_safe(data_root, revision=revision)
         metadata = metadata_result.unwrap() if is_successful(metadata_result) else None
         return cls(Maybe.from_optional(metadata))
 
@@ -65,11 +66,3 @@ def status_json(self) -> Maybe[Path]:
     def to_schema_file(self, schema: SchemaName) -> Maybe[Path]:
         """Convert schema name to file path, which is <data_root>/annotation/<schema_name>.json."""
         return self.annotation_dir.map(lambda ann: ann.joinpath(schema.filename))
-
-
-@safe
-def _load_metadata_safe(
-    data_root: str,
-    revision: str | None = None,
-) -> Result[DBMetadata, Exception]:
-    return load_metadata(data_root, revision=revision)

t4_devkit/sanity/format/base.py

@@ -4,13 +4,12 @@
 
 from returns.maybe import Maybe, Nothing, Some
 from returns.pipeline import is_successful
-from returns.result import Result, safe
 
-from t4_devkit.schema import SCHEMAS, SchemaBase, SchemaName
+from t4_devkit.schema import SCHEMAS, SchemaName
 
 from ..checker import Checker
 from ..result import Reason
-from ..safety import load_json_safe
+from ..safety import load_json_safe, load_schema_safe
 
 if TYPE_CHECKING:
     from ..context import SanityContext
@@ -52,12 +51,7 @@ def _build_records(schema: SchemaName, records: list[dict]) -> list[Reason] | No
     module = SCHEMAS.get(schema)
     failures = []
     for record in records:
-        conversion = _safe_from_dict(module, record)
+        conversion = load_schema_safe(module, record)
         if not is_successful(conversion):
             failures.append(Reason(f"[{schema.name}] {record['token']}: {conversion.failure()}"))
     return failures if failures else None
-
-
-@safe
-def _safe_from_dict(module: type[SchemaBase], record: dict) -> Result[SchemaBase, Exception]:
-    return module.from_dict(record)

t4_devkit/sanity/safety.py

@@ -1,9 +1,16 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from returns.result import Result, safe
 
+from t4_devkit import DBMetadata, Tier4, load_metadata
 from t4_devkit.common.io import load_json
-from t4_devkit import Tier4
+
+if TYPE_CHECKING:
+    from t4_devkit.schema import SchemaBase
+
+    from .context import SanityContext
 
 
 @safe
@@ -13,6 +20,24 @@ def load_json_safe(filename: str) -> Result[list[dict], Exception]:
 
 
 @safe
-def init_tier4_safe(data_root: str, revision: str | None = None) -> Result[Tier4, Exception]:
-    """Initialize Tier4 instance safely."""
+def load_schema_safe(module: type[SchemaBase], record: dict) -> Result[SchemaBase, Exception]:
+    """Load schema from dict safely."""
+    return module.from_dict(record)
+
+
+@safe
+def load_metadata_safe(
+    data_root: str,
+    revision: str | None = None,
+) -> Result[DBMetadata, Exception]:
+    """Load DBMetadata safely."""
+    return load_metadata(data_root, revision=revision)
+
+
+@safe
+def load_tier4_safe(context: SanityContext) -> Result[Tier4, Exception]:
+    """Load Tier4 instance safely."""
+    data_root = context.data_root.unwrap()
+    revision = context.version.value_or(None)
+    data_root = data_root.as_posix() if revision is None else data_root.parent.as_posix()
     return Tier4(data_root, revision=revision, verbose=False)

t4_devkit/sanity/tier4/tiv001.py

@@ -4,13 +4,11 @@
 
 from returns.maybe import Maybe, Nothing, Some
 from returns.pipeline import is_successful
-from returns.result import Result, safe
-
-from t4_devkit import Tier4
 
 from ..checker import Checker, RuleID, RuleName, Severity
 from ..registry import CHECKERS
 from ..result import Reason
+from ..safety import load_tier4_safe
 
 if TYPE_CHECKING:
     from ..context import SanityContext
@@ -37,15 +35,7 @@ def can_skip(self, context: SanityContext) -> Maybe[Reason]:
                 return Maybe.from_value(Reason("Data root not found"))
 
     def check(self, context: SanityContext) -> list[Reason] | None:
-        result = _load_tier4_safe(context)
+        result = load_tier4_safe(context)
         return (
             None if is_successful(result) else [Reason(f"Failed to load Tier4: {result.failure()}")]
         )
-
-
-@safe
-def _load_tier4_safe(context: SanityContext) -> Result[Tier4, Exception]:
-    data_root = context.data_root.unwrap()
-    revision = context.version.value_or(None)
-    data_root = data_root.as_posix() if revision is None else data_root.parent.as_posix()
-    return Tier4(data_root, revision=revision, verbose=False)