sqlatypemodel package¶

Subpackages¶

Submodules¶

sqlatypemodel.exceptions module¶

Custom exceptions for sqlatypemodel serialization operations.

This module defines domain-specific exceptions for handling serialization and deserialization errors when working with Pydantic models in SQLAlchemy.

Example

>>> from sqlatypemodel.exceptions import SerializationError
>>> try:
...     # serialization operation
...     pass
... except SerializationError as e:
...     print(f"Failed to serialize: {e}")

exception sqlatypemodel.exceptions.DeserializationError(model_name: str, data: Any = None, original_error: Exception | None = None)[source]¶

Bases: SQLATypeModelError

Raised when model deserialization from database format fails.

This exception is raised when database JSON data cannot be converted back into a Pydantic model instance.

model_name¶

Name of the model class that failed to deserialize.

Type:: str

data¶

The raw database data that could not be deserialized.

Type:: dict[str, Any] | None

original_error¶

The underlying exception that caused the failure.

Type:: Exception | None

Example

>>> raise DeserializationError(
...     "MyModel",
...     {"invalid": "data"},
...     ValidationError("Missing required field")
... )

__init__(model_name: str, data: Any = None, original_error: Exception | None = None) → None[source]¶

Initialize DeserializationError.

Parameters:

model_name – Name of the model class that failed to deserialize.
data – The raw database data that could not be deserialized. Defaults to None.
original_error – The underlying exception that caused the failure. Defaults to None.

exception sqlatypemodel.exceptions.SQLATypeModelError[source]¶

Bases: Exception

Base exception for all sqlatypemodel errors.

All custom exceptions in this library inherit from this class, allowing users to catch all library-specific errors with a single except clause.

Example

>>> try:
...     # any sqlatypemodel operation
...     pass
... except SQLATypeModelError:
...     print("A sqlatypemodel error occurred")

exception sqlatypemodel.exceptions.SerializationError(model_name: str, original_error: Exception | None = None)[source]¶

Bases: SQLATypeModelError

Raised when model serialization to database format fails.

This exception is raised when a Pydantic model instance cannot be converted to a JSON-compatible dictionary for database storage.

model_name¶

Name of the model class that failed to serialize.

Type:: str

original_error¶

The underlying exception that caused the failure.

Type:: Exception | None

Example

>>> raise SerializationError(
...     "MyModel",
...     ValueError("Invalid field value")
... )

__init__(model_name: str, original_error: Exception | None = None) → None[source]¶

Initialize SerializationError.

Parameters:

model_name – Name of the model class that failed to serialize.
original_error – The underlying exception that caused the failure. Defaults to None.

Module contents¶

sqlatypemodel - SQLAlchemy TypeDecorator for Pydantic models.

This package provides tools for storing Pydantic models in SQLAlchemy JSON columns with automatic serialization, deserialization, and mutation tracking.

Example

>>> from pydantic import BaseModel
>>> from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
>>> from sqlatypemodel import ModelType, MutableMixin
>>>
>>> class Base(DeclarativeBase):
...     pass
>>>
>>> class UserSettings(MutableMixin, BaseModel):
...     theme: str = "light"
...     notifications: bool = True
...     tags: list[str] = []
>>>
>>> class User(Base):
...     __tablename__ = "users"
...     id: Mapped[int] = mapped_column(primary_key=True)
...
...     # MutableMixin automatically registers with ModelType
...     # via __init_subclass__
...     settings: Mapped[UserSettings] = mapped_column(
...         ModelType(UserSettings)
...     )
>>>
>>> # Usage
>>> user = User(settings=UserSettings())
>>> user.settings.theme = "dark"  # Automatically tracked!
>>> user.settings.tags.append("premium")  # Also tracked!

exception sqlatypemodel.DeserializationError(model_name: str, data: Any = None, original_error: Exception | None = None)[source]

Bases: SQLATypeModelError

Raised when model deserialization from database format fails.

This exception is raised when database JSON data cannot be converted back into a Pydantic model instance.

model_name

Name of the model class that failed to deserialize.

Type:: str

data

The raw database data that could not be deserialized.

Type:: dict[str, Any] | None

original_error

The underlying exception that caused the failure.

Type:: Exception | None

Example

>>> raise DeserializationError(
...     "MyModel",
...     {"invalid": "data"},
...     ValidationError("Missing required field")
... )

__init__(model_name: str, data: Any = None, original_error: Exception | None = None) → None[source]

Initialize DeserializationError.

Parameters:

model_name – Name of the model class that failed to deserialize.
data – The raw database data that could not be deserialized. Defaults to None.
original_error – The underlying exception that caused the failure. Defaults to None.

class sqlatypemodel.LazyMutableMixin(*args: Any, **kwargs: Any)[source]

Bases: BaseMutableMixin

Lazy Implementation of MutableMixin.

This implementation defers the wrapping of mutable fields until they are first accessed. It is highly optimized for read-heavy workloads where only a subset of fields might be accessed.

__getattribute__(name: str) → Any[source]

Retrieve attribute with Just-In-Time wrapping

Wraps mutable attributes in tracking proxies upon first access.

Parameters:: name – The name of the attribute to retrieve.
Returns:: The attribute value, possibly wrapped in a tracking proxy.

class sqlatypemodel.ModelType(model: type[T], dumper: Callable[[T], dict[str, Any]] | None = None, loader: Callable[[dict[str, Any]], T] | None = None, *args: Any, **kwargs: Any)[source]

Bases: TypeDecorator, Generic[T]

SQLAlchemy TypeDecorator for storing Pydantic models as JSON.

This type handles the automatic serialization and deserialization of Pydantic models (and compatible classes) to and from JSON. It also ensures that mutation tracking is restored when objects are loaded from the database.

impl: The underlying SQLAlchemy type (JSON).

cache_ok: Whether the type is safe to cache (True).

__init__(model: type[T], dumper: Callable[[T], dict[str, Any]] | None = None, loader: Callable[[dict[str, Any]], T] | None = None, *args: Any, **kwargs: Any) → None[source]

Initialize the ModelType.

Parameters:

model – A Pydantic model class (or compatible) to be stored.
dumper – Custom serialization function (Model -> Dict).
loader – Custom deserialization function (Dict -> Model).
*args – Additional arguments for TypeDecorator.
**kwargs – Additional keyword arguments for TypeDecorator.

Raises:

ValueError – If serialization or deserialization functions cannot be resolved.

cache_ok: bool | None = True

Indicate if statements using this ExternalType are “safe to cache”.

The default value None will emit a warning and then not allow caching of a statement which includes this type. Set to False to disable statements using this type from being cached at all without a warning. When set to True, the object’s class and selected elements from its state will be used as part of the cache key. For example, using a TypeDecorator:

class MyType(TypeDecorator):
    impl = String

    cache_ok = True

    def __init__(self, choices):
        self.choices = tuple(choices)
        self.internal_only = True

The cache key for the above type would be equivalent to:

>>> MyType(["a", "b", "c"])._static_cache_key
(<class '__main__.MyType'>, ('choices', ('a', 'b', 'c')))

The caching scheme will extract attributes from the type that correspond to the names of parameters in the __init__() method. Above, the “choices” attribute becomes part of the cache key but “internal_only” does not, because there is no parameter named “internal_only”.

The requirements for cacheable elements is that they are hashable and also that they indicate the same SQL rendered for expressions using this type every time for a given cache value.

To accommodate for datatypes that refer to unhashable structures such as dictionaries, sets and lists, these objects can be made “cacheable” by assigning hashable structures to the attributes whose names correspond with the names of the arguments. For example, a datatype which accepts a dictionary of lookup values may publish this as a sorted series of tuples. Given a previously un-cacheable type as:

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    this is the non-cacheable version, as "self.lookup" is not
    hashable.

    """

    def __init__(self, lookup):
        self.lookup = lookup

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self.lookup" ...

Where “lookup” is a dictionary. The type will not be able to generate a cache key:

>>> type_ = LookupType({"a": 10, "b": 20})
>>> type_._static_cache_key
<stdin>:1: SAWarning: UserDefinedType LookupType({'a': 10, 'b': 20}) will not
produce a cache key because the ``cache_ok`` flag is not set to True.
Set this flag to True if this type object's state is safe to use
in a cache key, or False to disable this warning.
symbol('no_cache')

If we did set up such a cache key, it wouldn’t be usable. We would get a tuple structure that contains a dictionary inside of it, which cannot itself be used as a key in a “cache dictionary” such as SQLAlchemy’s statement cache, since Python dictionaries aren’t hashable:

>>> # set cache_ok = True
>>> type_.cache_ok = True

>>> # this is the cache key it would generate
>>> key = type_._static_cache_key
>>> key
(<class '__main__.LookupType'>, ('lookup', {'a': 10, 'b': 20}))

>>> # however this key is not hashable, will fail when used with
>>> # SQLAlchemy statement cache
>>> some_cache = {key: "some sql value"}
Traceback (most recent call last): File "<stdin>", line 1,
in <module> TypeError: unhashable type: 'dict'

The type may be made cacheable by assigning a sorted tuple of tuples to the “.lookup” attribute:

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    The dictionary is stored both as itself in a private variable,
    and published in a public variable as a sorted tuple of tuples,
    which is hashable and will also return the same value for any
    two equivalent dictionaries.  Note it assumes the keys and
    values of the dictionary are themselves hashable.

    """

    cache_ok = True

    def __init__(self, lookup):
        self._lookup = lookup

        # assume keys/values of "lookup" are hashable; otherwise
        # they would also need to be converted in some way here
        self.lookup = tuple((key, lookup[key]) for key in sorted(lookup))

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self._lookup" ...

Where above, the cache key for LookupType({"a": 10, "b": 20}) will be:

>>> LookupType({"a": 10, "b": 20})._static_cache_key
(<class '__main__.LookupType'>, ('lookup', (('a', 10), ('b', 20))))

Added in version 1.4.14: - added the cache_ok flag to allow some configurability of caching for TypeDecorator classes.

Added in version 1.4.28: - added the ExternalType mixin which generalizes the cache_ok flag to both the TypeDecorator and UserDefinedType classes.

See also

sql_caching

impl: alias of JSON

process_bind_param(value: T | dict[str, Any] | None, dialect: Dialect) → dict[str, Any] | None[source]

Convert Model to Dict for SQLAlchemy’s JSON type.

Parameters:

value – The model instance or dict to bind.
dialect – The database dialect.

Returns:

The serialized dictionary or None.

Raises:

SerializationError – If serialization fails.

process_literal_param(value: T | None, dialect: Dialect) → str[source]

Render value as a literal SQL string (for logs/debugging).

Parameters:

value – The value to render.
dialect – The database dialect.

Returns:

The SQL string representation.

process_result_value(value: dict[str, Any] | str | bytes | None, dialect: Dialect) → T | None[source]

Convert DB value back to Model and restore tracking.

Parameters:

value – The raw value from the database.
dialect – The database dialect.

Returns:

The deserialized model instance or None.

Raises:

DeserializationError – If deserialization fails.

property python_type: type[T]

Return the Python type associated with this type.

Returns:: The Python model class.

classmethod register_mutable(mutable: type[BaseMutableMixin]) → None[source]

Associate a MutableMixin subclass with this type.

Used to automatically register tracking mixins with the SQLAlchemy type system.

Parameters:: mutable – The MutableMixin subclass to register.
Raises:: TypeError – If mutable is not a subclass of BaseMutableMixin.

class sqlatypemodel.MutableMixin(*args: Any, **kwargs: Any)[source]

Bases: BaseMutableMixin

Standard (Eager) Implementation of MutableMixin.

This implementation eagerly scans and wraps all mutable fields upon initialization. It is suitable for write-heavy workloads or when fields are accessed frequently.

__init__(*args: Any, **kwargs: Any) → None[source]: Initialize and immediately restore tracking.

exception sqlatypemodel.SQLATypeModelError[source]

Bases: Exception

Base exception for all sqlatypemodel errors.

All custom exceptions in this library inherit from this class, allowing users to catch all library-specific errors with a single except clause.

Example

>>> try:
...     # any sqlatypemodel operation
...     pass
... except SQLATypeModelError:
...     print("A sqlatypemodel error occurred")

exception sqlatypemodel.SerializationError(model_name: str, original_error: Exception | None = None)[source]

Bases: SQLATypeModelError

Raised when model serialization to database format fails.

This exception is raised when a Pydantic model instance cannot be converted to a JSON-compatible dictionary for database storage.

model_name

Name of the model class that failed to serialize.

Type:: str

original_error

The underlying exception that caused the failure.

Type:: Exception | None

Example

>>> raise SerializationError(
...     "MyModel",
...     ValueError("Invalid field value")
... )

__init__(model_name: str, original_error: Exception | None = None) → None[source]

Initialize SerializationError.

Parameters:

model_name – Name of the model class that failed to serialize.
original_error – The underlying exception that caused the failure. Defaults to None.