Extract shared from_dict validation helpers for model classes (#70)

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",

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)

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"],

models/from_dict_validation.py

@@ -0,0 +1,96 @@
+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."""
+    value = raw.get(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:
+        value = raw.get(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 (absent or empty)."""
+    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

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)