sqlatypemodel.mixin package

Submodules

sqlatypemodel.mixin.events module

Change notification logic and signal propagation.

sqlatypemodel.mixin.events.batch_change_suppression(instance: Trackable) → Iterator[None]

Context manager to suppress change notifications.

Increments a suppression counter. If modifications occur while suppressed, a single notification is fired upon exiting the outermost context.

Parameters:

instance – The trackable object to suppress notifications for.

Yields:

None

sqlatypemodel.mixin.events.mark_change_or_defer(instance: Trackable) → bool

Check if a change should be emitted or deferred.

Parameters:

instance – The trackable object.

Returns:

True if the change signal should be emitted immediately, False if it was suppressed/deferred.

sqlatypemodel.mixin.events.safe_changed(obj: Trackable, max_failures: int = 10, max_retries: int = 3) → None

Safely notify parent objects about changes (optimized).

Handles race conditions when the _parents dictionary is modified during iteration by using a snapshot-and-retry approach.

Parameters:

• obj – The trackable instance that changed.

• max_failures – Maximum allowed propagation failures before stopping.

• max_retries – Maximum attempts to snapshot parents dictionary.

sqlatypemodel.mixin.inspection module

Introspection and validation logic for object attributes.

sqlatypemodel.mixin.inspection.extract_attrs_to_scan(instance: Any) → dict[str, Any]

Extract attributes from __dict__ and __slots__ for scanning (optimized).

Parameters:

instance – The object instance to extract attributes from.

Returns:

A dictionary mapping attribute names to their values.

sqlatypemodel.mixin.inspection.ignore_attr_name(cls: type[Any], attr_name: str) → bool

Fast check if an attribute should be ignored during scanning.

This function uses caching to improve performance for repeated checks.

Parameters:

• cls – The class of the object being inspected.

• attr_name – The name of the attribute.

Returns:

True if the attribute should be skipped, False otherwise.

sqlatypemodel.mixin.inspection.is_descriptor_property(descriptor: Any) → bool

Check if a descriptor is a property or read-only attribute.

Parameters:

descriptor – The attribute descriptor to check.

Returns:

True if the descriptor is a property or read-only, False otherwise.

sqlatypemodel.mixin.inspection.is_pydantic(obj: Any) → bool

Check if an object is a Pydantic model instance.

Parameters:

obj – The object to inspect.

Returns:

True if the object appears to be a Pydantic model.

sqlatypemodel.mixin.inspection.should_notify_change(old_value: Any, new_value: Any) → bool

Determine if a change notification is necessary (optimized).

Parameters:

• old_value – The previous value of the attribute.

• new_value – The new value being assigned.

Returns:

True if a change notification should be fired, False otherwise.

sqlatypemodel.mixin.mixin module

Main Mixin module providing mutation tracking capabilities.

class sqlatypemodel.mixin.mixin.BaseMutableMixin(*args: Any, **kwargs: Any)

Bases: MutableMethods, Mutable, ABC

Abstract Base Class for Mutable Mixins.

Implements change tracking using State-based parent references. This class serves as the foundation for both Eager and Lazy mutation tracking strategies. It handles the interception of attribute access and assignment to detect changes in nested mutable structures.

__getstate__() → dict[str, Any]

Prepare object state for pickling.

Returns:

A dictionary representing the object state.

__init__(*args: Any, **kwargs: Any) → None

Initialize the mixin with default tracking state.

classmethod __init_subclass__(**kwargs: Any) → None

Register subclass with SQLAlchemy ModelType.

Automatically registers the subclass with the associated SQLAlchemy type unless auto_register=False is passed.

Parameters:

**kwargs –

Keyword arguments passed to __init_subclass__. Includes:

• auto_register (bool): Whether to register with ModelType. Defaults to True.

• associate (type[ModelType]): Specific ModelType to associate with.

__setattr__(name: str, value: Any) → None

Set an attribute with automatic change tracking.

Intercepts attribute assignment to wrap mutable values and notify SQLAlchemy of changes.

Parameters:

• name – The name of the attribute.

• value – The value to set.

__setstate__(state: dict[str, Any]) → None

Restore object state from pickle.

Parameters:

state – The dictionary state to restore.

batch_changes() → AbstractContextManager[None]

Context manager to batch multiple changes.

Prevents multiple changed() notifications during a block of code. Notifications are coalesced and sent once when the block exits.

Returns:

A context manager that suppresses change notifications.

changed() → None

Notify observers that this object has changed.

This method signals to SQLAlchemy that the object has been modified. It respects the change suppression level to support batched updates.

classmethod coerce(key: str, value: Any) → M | None

Coerce value into the Mixin type.

Used by SQLAlchemy to convert raw values into the mutable type.

Parameters:

• key – The key being coerced.

• value – The value to coerce.

Returns:

The coerced value or None.

class sqlatypemodel.mixin.mixin.LazyMutableMixin(*args: Any, **kwargs: Any)

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

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.mixin.mixin.MutableMixin(*args: Any, **kwargs: Any)

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

Initialize and immediately restore tracking.

sqlatypemodel.mixin.protocols module

Protocols and base implementations for change tracking.

This module defines the protocols that trackable objects must implement, as well as base implementations for common tracking functionality.

class sqlatypemodel.mixin.protocols.MutableMethods

Bases: object

Base implementation of core tracking properties and methods.

Provides the implementations for _parents and _state management, as well as the changed() signal propagation.

changed() → None

Notify parents using the library’s safe propagation logic.

class sqlatypemodel.mixin.protocols.MutableMixinProto(*args, **kwargs)

Bases: Trackable, Protocol

Protocol describing a fully functional MutableMixin instance.

This includes internal state tracking attributes used by the library.

class sqlatypemodel.mixin.protocols.Trackable(*args, **kwargs)

Bases: Protocol

Protocol describing a trackable object.

Objects adhering to this protocol must maintain a reference to their parents and provide a mechanism to signal changes.

changed() → None

Mark the object as changed and propagate the notification.

sqlatypemodel.mixin.serialization module

Pickle state management and serialization helpers.

sqlatypemodel.mixin.serialization.cleanup_pickle_state(state: Any) → Any

Remove unpicklable attributes (like weakrefs) from the state dict.

Parameters:

state – The state object (usually a dict) to be cleaned.

Returns:

The cleaned state object safe for pickling.

sqlatypemodel.mixin.serialization.manual_setstate(instance: Any, state: dict[str, Any]) → None

Manually restore state when parent class lacks __setstate__.

Parameters:

• instance – The object instance to restore state into.

• state – The dictionary containing the state attributes.

sqlatypemodel.mixin.serialization.reset_trackable_state(instance: Any) → None

Reset library-specific tracking attributes to default values.

This is typically called after unpickling to revive the object’s change tracking capabilities.

Parameters:

instance – The object instance to reset.

sqlatypemodel.mixin.state module

class sqlatypemodel.mixin.state.MutableState(obj: T)

Bases: Generic[T]

Immutable wrapper for parent references in the change tracking graph.

This class acts as a hashable token representing the identity of a trackable object. It solves the ‘unhashable parent’ problem by holding a weak reference to the parent object, allowing any object (even unhashable ones like lists or unfreezed dataclasses) to participate in the parent tracking mechanism.

ref

A weak reference to the parent object.

__init__(obj: T) → None

Initialize the mutable state token.

Parameters:

obj – The parent object to create a weak reference for.

link(child: Trackable | Any, key: str | int | None) → None

Establish a tracking connection between this state and a child.

Registers this MutableState instance in the child’s _parents dictionary. This enables the child to bubble up mutation events to the parent.

The operation is thread-safe using a recursive lock.

Parameters:

• child – The child object that should track this parent state.

• key – The attribute name or collection index where the child is stored within the parent.

obj() → T | None

Return the parent object from the weak reference.

This method fulfills the SQLAlchemy Mutable parent protocol, allowing this state token to be used directly in SQLAlchemy’s _parents dictionary.

Returns:

The dereferenced parent object, or None if it has been collected.

ref: ReferenceType[T]

unlink(child: Trackable | Any) → None

Break the tracking connection between this state and a child.

Removes this MutableState instance from the child’s _parents dictionary.

The operation is thread-safe using a recursive lock.

Parameters:

child – The child object to disconnect from this parent state.

sqlatypemodel.mixin.types module

Custom SQLAlchemy Mutable types with hashing support.

class sqlatypemodel.mixin.types.KeyableMutableDict

Bases: MutableMethods, MutableDict[_KT, _VT]

MutableDict that uses identity hashing and custom change tracking.

class sqlatypemodel.mixin.types.KeyableMutableList(iterable=(), /)

Bases: MutableMethods, MutableList[_T]

MutableList that uses identity hashing and custom change tracking.

class sqlatypemodel.mixin.types.KeyableMutableSet(iterable=(), /)

Bases: MutableMethods, MutableSet[_T]

MutableSet that uses identity hashing and custom change tracking.

sqlatypemodel.mixin.wrapping module

Recursive wrapping logic for mutable structures.

sqlatypemodel.mixin.wrapping.get_or_create_state(parent: Trackable | Any) → MutableState[Any]

Retrieves or creates a MutableState identity token (optimized).

sqlatypemodel.mixin.wrapping.is_mutable_and_untracked(obj: Any) → bool

Check if object needs wrapping OR patching (O(1) fast path).

sqlatypemodel.mixin.wrapping.relink_descendants(parent: Any, _seen: dict[int, Any] | None = None) → None

Recursively re-link already wrapped objects to their current parent.

sqlatypemodel.mixin.wrapping.scan_and_wrap_fields(parent: Any, _seen: dict[int, Any] | None = None) → None

Iterate over object fields and wrap mutable ones (optimized).

sqlatypemodel.mixin.wrapping.wrap_mutable(parent: Trackable | Any, value: Any, _seen: dict[int, Any] | None = None, depth: int = 0, key: str | int | None = None, max_depth_limit: int | None = None) → Any

Recursively wrap collections and trackable objects (optimized).

Module contents

Mixin classes for automatic mutation tracking in database models.

This module provides MutableMixin and LazyMutableMixin for tracking changes to nested Python objects stored in SQLAlchemy JSON columns.

class sqlatypemodel.mixin.BaseMutableMixin(*args: Any, **kwargs: Any)

Bases: MutableMethods, Mutable, ABC

Abstract Base Class for Mutable Mixins.

Implements change tracking using State-based parent references. This class serves as the foundation for both Eager and Lazy mutation tracking strategies. It handles the interception of attribute access and assignment to detect changes in nested mutable structures.

__getstate__() → dict[str, Any]

Prepare object state for pickling.

Returns:

A dictionary representing the object state.

__init__(*args: Any, **kwargs: Any) → None

Initialize the mixin with default tracking state.

classmethod __init_subclass__(**kwargs: Any) → None

Register subclass with SQLAlchemy ModelType.

Automatically registers the subclass with the associated SQLAlchemy type unless auto_register=False is passed.

Parameters:

**kwargs –

Keyword arguments passed to __init_subclass__. Includes:

• auto_register (bool): Whether to register with ModelType. Defaults to True.

• associate (type[ModelType]): Specific ModelType to associate with.

__setattr__(name: str, value: Any) → None

Set an attribute with automatic change tracking.

Intercepts attribute assignment to wrap mutable values and notify SQLAlchemy of changes.

Parameters:

• name – The name of the attribute.

• value – The value to set.

__setstate__(state: dict[str, Any]) → None

Restore object state from pickle.

Parameters:

state – The dictionary state to restore.

batch_changes() → AbstractContextManager[None]

Context manager to batch multiple changes.

Prevents multiple changed() notifications during a block of code. Notifications are coalesced and sent once when the block exits.

Returns:

A context manager that suppresses change notifications.

changed() → None

Notify observers that this object has changed.

This method signals to SQLAlchemy that the object has been modified. It respects the change suppression level to support batched updates.

classmethod coerce(key: str, value: Any) → M | None

Coerce value into the Mixin type.

Used by SQLAlchemy to convert raw values into the mutable type.

Parameters:

• key – The key being coerced.

• value – The value to coerce.

Returns:

The coerced value or None.

class sqlatypemodel.mixin.LazyMutableMixin(*args: Any, **kwargs: Any)

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

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.mixin.MutableMixin(*args: Any, **kwargs: Any)

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

Initialize and immediately restore tracking.