diff --git a/models/cli_session.py b/models/cli_session.py index 473531e..a2e5196 100644 --- a/models/cli_session.py +++ b/models/cli_session.py @@ -4,6 +4,7 @@ from typing import Any from models.errors import SchemaError +from models.from_dict_validation import require_dict, require_truthy @dataclass(frozen=True) @@ -16,15 +17,12 @@ class CliSessionMeta: @classmethod def from_dict(cls, raw: dict[str, Any]) -> "CliSessionMeta": - if not isinstance(raw, dict): - raise SchemaError( - "CliSessionMeta", - "meta", - hint=f"expected object, got {type(raw).__name__}", - ) - latest = raw.get("latestRootBlobId") - if not latest: - raise SchemaError("CliSessionMeta", "latestRootBlobId") + raw = require_dict(raw, model="CliSessionMeta", field="meta") + latest = require_truthy( + raw.get("latestRootBlobId"), + model="CliSessionMeta", + field="latestRootBlobId", + ) if not isinstance(latest, str): raise SchemaError( "CliSessionMeta", diff --git a/models/conversation.py b/models/conversation.py index c2a103a..a3d5e2f 100644 --- a/models/conversation.py +++ b/models/conversation.py @@ -4,6 +4,13 @@ from typing import Any from models.errors import SchemaError +from models.from_dict_validation import ( + require_dict, + require_key, + require_non_empty_str, + require_non_empty_str_field, + require_type, +) @dataclass(frozen=True) @@ -20,22 +27,10 @@ class Composer: @classmethod def from_dict(cls, raw: dict[str, Any], *, composer_id: str) -> "Composer": - if not isinstance(raw, dict): - raise SchemaError( - "Composer", - "composerData", - hint=f"expected object, got {type(raw).__name__}", - ) - if not isinstance(composer_id, str) or not composer_id: - raise SchemaError( - "Composer", - "composerId", - hint=f"expected non-empty str, got {type(composer_id).__name__}", - ) - if "fullConversationHeadersOnly" not in raw: - raise SchemaError("Composer", "fullConversationHeadersOnly") - if "createdAt" not in raw: - raise SchemaError("Composer", "createdAt") + raw = require_dict(raw, model="Composer", field="composerData") + require_non_empty_str(composer_id, model="Composer", field="composerId") + require_key(raw, "fullConversationHeadersOnly", model="Composer") + require_key(raw, "createdAt", model="Composer") created_at = raw.get("createdAt") # Numeric-only on purpose: a 2026-05 scan of 17/17 live composers on @@ -49,13 +44,14 @@ def from_dict(cls, raw: dict[str, Any], *, composer_id: str) -> "Composer": hint=f"expected timestamp number, got {type(created_at).__name__}", ) - headers = raw.get("fullConversationHeadersOnly") - if not isinstance(headers, list): - raise SchemaError( - "Composer", - "fullConversationHeadersOnly", - hint=f"expected list, got {type(headers).__name__}", - ) + headers_value = raw.get("fullConversationHeadersOnly") + headers = require_type( + headers_value, + list, + model="Composer", + field="fullConversationHeadersOnly", + hint=f"expected list, got {type(headers_value).__name__}", + ) model_config = raw.get("modelConfig") or {} if not isinstance(model_config, dict): @@ -82,19 +78,10 @@ class WorkspaceLocalComposer: @classmethod def from_dict(cls, raw: dict[str, Any]) -> "WorkspaceLocalComposer": - if not isinstance(raw, dict): - raise SchemaError( - "WorkspaceLocalComposer", - "composer", - hint=f"expected object, got {type(raw).__name__}", - ) - composer_id = raw.get("composerId") - if not isinstance(composer_id, str) or not composer_id: - raise SchemaError( - "WorkspaceLocalComposer", - "composerId", - hint=f"expected non-empty str, got {type(composer_id).__name__}", - ) + raw = require_dict(raw, model="WorkspaceLocalComposer", field="composer") + composer_id = require_non_empty_str_field( + raw, "composerId", model="WorkspaceLocalComposer" + ) return cls( composer_id=composer_id, last_updated_at=raw.get("lastUpdatedAt"), @@ -111,16 +98,6 @@ class Bubble: @classmethod def from_dict(cls, raw: dict[str, Any], *, bubble_id: str) -> "Bubble": - if not isinstance(raw, dict): - raise SchemaError( - "Bubble", - "bubble", - hint=f"expected object, got {type(raw).__name__}", - ) - if not isinstance(bubble_id, str) or not bubble_id: - raise SchemaError( - "Bubble", - "bubbleId", - hint=f"expected non-empty str, got {type(bubble_id).__name__}", - ) + raw = require_dict(raw, model="Bubble", field="bubble") + require_non_empty_str(bubble_id, model="Bubble", field="bubbleId") return cls(bubble_id=bubble_id, raw=raw) diff --git a/models/export.py b/models/export.py index fc5d594..ff08bc9 100644 --- a/models/export.py +++ b/models/export.py @@ -3,7 +3,7 @@ from dataclasses import dataclass, field from typing import Any -from models.errors import SchemaError +from models.from_dict_validation import require_dict, require_non_empty_str_fields @dataclass(frozen=True) @@ -19,20 +19,12 @@ class ExportEntry: @classmethod def from_dict(cls, raw: dict[str, Any]) -> "ExportEntry": - if not isinstance(raw, dict): - raise SchemaError( - "ExportEntry", - "entry", - hint=f"expected object, got {type(raw).__name__}", - ) - for required in ("log_id", "title", "workspace"): - value = raw.get(required) - if not isinstance(value, str) or value == "": - raise SchemaError( - "ExportEntry", - required, - hint=f"expected non-empty str, got {type(value).__name__}", - ) + raw = require_dict(raw, model="ExportEntry", field="entry") + require_non_empty_str_fields( + raw, + ("log_id", "title", "workspace"), + model="ExportEntry", + ) return cls( log_id=raw["log_id"], title=raw["title"], diff --git a/models/from_dict_validation.py b/models/from_dict_validation.py new file mode 100644 index 0000000..5cdbde5 --- /dev/null +++ b/models/from_dict_validation.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +from typing import Any + +from models.errors import SchemaError + + +def require_dict(raw: Any, *, model: str, field: str) -> dict[str, Any]: + """Raise SchemaError when raw is not a dict; return raw for chaining.""" + if not isinstance(raw, dict): + raise SchemaError( + model, + field, + hint=f"expected object, got {type(raw).__name__}", + ) + return raw + + +def require_key(raw: dict[str, Any], key: str, *, model: str) -> None: + """Raise SchemaError when a required key is absent.""" + if key not in raw: + raise SchemaError(model, key) + + +def require_non_empty_str(value: Any, *, model: str, field: str) -> str: + """Validate a caller-supplied id (workspace_id, composer_id, bubble_id).""" + if not isinstance(value, str) or not value: + raise SchemaError( + model, + field, + hint=f"expected non-empty str, got {type(value).__name__}", + ) + return value + + +def require_non_empty_str_field(raw: dict[str, Any], key: str, *, model: str) -> str: + """Validate a non-empty string field read from raw.""" + if key not in raw: + raise SchemaError(model, key) + value = raw[key] + if not isinstance(value, str) or not value: + raise SchemaError( + model, + key, + hint=f"expected non-empty str, got {type(value).__name__}", + ) + return value + + +def require_non_empty_str_fields( + raw: dict[str, Any], + keys: tuple[str, ...], + *, + model: str, +) -> None: + """Validate multiple non-empty string fields in raw (ExportEntry pattern).""" + for key in keys: + if key not in raw: + raise SchemaError(model, key) + value = raw[key] + if not isinstance(value, str) or value == "": + raise SchemaError( + model, + key, + hint=f"expected non-empty str, got {type(value).__name__}", + ) + + +def require_truthy(value: Any, *, model: str, field: str) -> Any: + """Raise missing-field SchemaError when value is falsy. + + Treats None, empty string, 0, and other falsy values as missing (no hint), + matching prior CliSessionMeta ``latestRootBlobId`` behavior via ``raw.get``. + """ + if not value: + raise SchemaError(model, field) + return value + + +def require_type( + value: Any, + expected: type[Any] | tuple[type[Any], ...], + *, + model: str, + field: str, + hint: str | None = None, +) -> Any: + if not isinstance(value, expected): + raise SchemaError( + model, + field, + hint=hint or f"expected {expected!r}, got {type(value).__name__}", + ) + return value + + +def require_optional_str(value: Any, *, model: str, field: str) -> str | None: + if value is not None and not isinstance(value, str): + raise SchemaError( + model, + field, + hint=f"expected str or None, got {type(value).__name__}", + ) + return value diff --git a/models/workspace.py b/models/workspace.py index 75869e3..8d9cdd9 100644 --- a/models/workspace.py +++ b/models/workspace.py @@ -3,7 +3,7 @@ from dataclasses import dataclass, field from typing import Any -from models.errors import SchemaError +from models.from_dict_validation import require_dict, require_non_empty_str, require_optional_str @dataclass(frozen=True) @@ -16,23 +16,7 @@ class Workspace: @classmethod def from_dict(cls, raw: dict[str, Any], *, workspace_id: str) -> "Workspace": - if not isinstance(raw, dict): - raise SchemaError( - "Workspace", - "workspace.json", - hint=f"expected object, got {type(raw).__name__}", - ) - if not isinstance(workspace_id, str) or not workspace_id: - raise SchemaError( - "Workspace", - "workspaceId", - hint=f"expected non-empty str, got {type(workspace_id).__name__}", - ) - folder = raw.get("folder") - if folder is not None and not isinstance(folder, str): - raise SchemaError( - "Workspace", - "folder", - hint=f"expected str or None, got {type(folder).__name__}", - ) + raw = require_dict(raw, model="Workspace", field="workspace.json") + require_non_empty_str(workspace_id, model="Workspace", field="workspaceId") + folder = require_optional_str(raw.get("folder"), model="Workspace", field="folder") return cls(workspace_id=workspace_id, folder=folder, raw=raw) diff --git a/tests/test_from_dict_validation.py b/tests/test_from_dict_validation.py new file mode 100644 index 0000000..1b4c03c --- /dev/null +++ b/tests/test_from_dict_validation.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import os +import sys +import unittest + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if REPO_ROOT not in sys.path: + sys.path.insert(0, REPO_ROOT) + +from models.errors import SchemaError +from models.from_dict_validation import ( + require_non_empty_str_field, + require_non_empty_str_fields, +) + + +class RequireNonEmptyStrFieldMessages(unittest.TestCase): + def test_absent_key_raises_missing_required_field(self) -> None: + with self.assertRaises(SchemaError) as cm: + require_non_empty_str_field({}, "composerId", model="TestModel") + self.assertEqual(cm.exception.field, "composerId") + self.assertIn("missing required field", str(cm.exception)) + self.assertNotIn("invalid field", str(cm.exception)) + + def test_wrong_type_raises_invalid_field(self) -> None: + with self.assertRaises(SchemaError) as cm: + require_non_empty_str_field( + {"composerId": 123}, + "composerId", + model="TestModel", + ) + self.assertEqual(cm.exception.field, "composerId") + self.assertIn("invalid field", str(cm.exception)) + self.assertIn("expected non-empty str, got int", str(cm.exception)) + self.assertNotIn("missing required field", str(cm.exception)) + + +class RequireNonEmptyStrFieldsMessages(unittest.TestCase): + def test_absent_key_raises_missing_required_field(self) -> None: + with self.assertRaises(SchemaError) as cm: + require_non_empty_str_fields( + {"title": "x", "workspace": "w"}, + ("log_id", "title", "workspace"), + model="ExportEntry", + ) + self.assertEqual(cm.exception.field, "log_id") + self.assertIn("missing required field", str(cm.exception)) + self.assertNotIn("invalid field", str(cm.exception)) + + def test_wrong_type_raises_invalid_field(self) -> None: + with self.assertRaises(SchemaError) as cm: + require_non_empty_str_fields( + {"log_id": 1, "title": "x", "workspace": "w"}, + ("log_id", "title", "workspace"), + model="ExportEntry", + ) + self.assertEqual(cm.exception.field, "log_id") + self.assertIn("invalid field", str(cm.exception)) + self.assertIn("expected non-empty str, got int", str(cm.exception)) + self.assertNotIn("missing required field", str(cm.exception)) + + +if __name__ == "__main__": + unittest.main()