Index

schema

Unified schema management for Marianne's SQLite databases.

Provides serialization utilities for converting Python values to/from SQLite-compatible types. The registry module exists but is not yet wired into production code paths — the state layer uses hand-written SQL.

Functions

deserialize_field

deserialize_field(value, annotation)

Convert a SQLite value back to the expected Python type.

Parameters:

Name	Type	Description	Default
value	Any	The raw value from SQLite.	required
annotation	Any	The type annotation for the target field.	required

Returns:

Type	Description
Any	The reconstructed Python value.

Source code in src/marianne/schema/converters.py

def deserialize_field(value: Any, annotation: Any) -> Any:
    """Convert a SQLite value back to the expected Python type.

    Args:
        value: The raw value from SQLite.
        annotation: The type annotation for the target field.

    Returns:
        The reconstructed Python value.
    """
    if value is None:
        return None

    inner, _ = _unwrap_optional(annotation)

    # datetime
    if inner is datetime:
        if isinstance(value, str):
            try:
                return datetime.fromisoformat(value)
            except ValueError:
                _logger.warning("corrupt_timestamp", value=value)
                return None
        return value

    # bool (stored as INTEGER 0/1)
    if inner is bool:
        return bool(value)

    # Enum
    if isinstance(inner, type) and issubclass(inner, Enum):
        try:
            return inner(value)
        except ValueError:
            _logger.warning("unknown_enum_value", type=inner.__name__, value=value)
            return value

    # Literal — return as-is
    if get_origin(inner) is Literal:
        return value

    # JSON types (list, dict, TypedDict, Pydantic sub-models, dataclasses)
    if _is_json_type(inner):
        if isinstance(value, str):
            try:
                return json.loads(value)
            except json.JSONDecodeError as exc:
                _logger.warning(
                    "json_parse_failed",
                    raw_length=len(value),
                    raw_preview=value[:100],
                    error=str(exc),
                )
                return None
        return value

    # Direct types (str, int, float)
    return value

serialize_field

serialize_field(value, annotation)

Convert a Python value to a SQLite-compatible parameter.

Parameters:

Name	Type	Description	Default
value	Any	The Python value to serialize.	required
annotation	Any	The type annotation for the field.	required

Returns:

Type	Description
SQLParam	A value suitable for sqlite3 execute() binding.

Source code in src/marianne/schema/converters.py

def serialize_field(value: Any, annotation: Any) -> SQLParam:
    """Convert a Python value to a SQLite-compatible parameter.

    Args:
        value: The Python value to serialize.
        annotation: The type annotation for the field.

    Returns:
        A value suitable for sqlite3 execute() binding.
    """
    if value is None:
        return None

    # Unwrap Optional so we operate on the inner type
    inner, _ = _unwrap_optional(annotation)

    # datetime → ISO string (must check before generic str check)
    if isinstance(value, datetime):
        return value.isoformat()

    # bool → int (must check before int, since bool is subclass of int)
    if isinstance(value, bool):
        return 1 if value else 0

    # Enum → .value string
    if isinstance(value, Enum):
        return cast("SQLParam", value.value)

    # Complex types → JSON
    if isinstance(value, (list, dict)):
        return json.dumps(value, default=_json_default)

    # Pydantic model instance → JSON
    if hasattr(value, "model_dump"):
        return json.dumps(value.model_dump(), default=_json_default)

    # Dataclass instance → JSON
    if dataclasses.is_dataclass(value) and not isinstance(value, type):
        return json.dumps(dataclasses.asdict(value), default=_json_default)

    # Direct types (str, int, float, bytes)
    return cast("SQLParam", value)