loom.core.repository.sqlalchemy

class loom.core.repository.sqlalchemy.AuditActorMixin

Bases: object

Optional actor tracking for auth-enabled deployments.

class loom.core.repository.sqlalchemy.AuditableModel(*args, **kwargs)

Bases: Base, IdentityMixin, TimestampMixin, AuditActorMixin

Optional model base that also tracks actor fields.

Parameters:

• args (Any)

• kwargs (Any)

Return type:

Any

class loom.core.repository.sqlalchemy.Base(*args, **kwargs)

Bases: DeclarativeBase

Base SQLAlchemy declarative class.

Parameters:

• args (Any)

• kwargs (Any)

Return type:

Any

class loom.core.repository.sqlalchemy.BaseModel(*args, **kwargs)

Bases: Base, IdentityMixin, TimestampMixin

Default project model base (identity + timestamps).

Parameters:

• args (Any)

• kwargs (Any)

Return type:

Any

class loom.core.repository.sqlalchemy.RepositoryRegistration(model, repository_type, contract=None)

Bases: object

Repository registration metadata for a model-bound SQLAlchemy repository.

Parameters:

• model (type[BaseModel])

• repository_type (type[RepositorySQLAlchemy[Any, Any]])

• contract (object | None)

class loom.core.repository.sqlalchemy.IdentityMixin

Bases: object

Integer primary key for entities with standard identity.

class loom.core.repository.sqlalchemy.Projection(*, loader, profiles=('default',), depends_on=(), default=None)

Bases: Generic[T]

Descriptor for derived fields with cache dependency metadata.

Parameters:

• loader (Any)

• profiles (tuple[str, ...])

• depends_on (tuple[str, ...])

• default (T | None)

has_value(obj)

Check whether the projection value has been populated on the given object.

Parameters:

obj (Any) – Model instance to inspect.

Returns:

True if the projection attribute has been set.

Return type:

bool

class loom.core.repository.sqlalchemy.RepositoryToken(model)

Bases: object

Internal DI token used for model-centric repository resolution.

Parameters:

model (type[BaseModel])

class loom.core.repository.sqlalchemy.RepositorySQLAlchemy(session_manager, model)

Bases: SQLAlchemyCreateMixin[OutputT, IdT], SQLAlchemyReadMixin[OutputT, IdT], SQLAlchemyUpdateMixin[OutputT, IdT], SQLAlchemyDeleteMixin[OutputT, IdT], Generic[OutputT, IdT]

Base SQLAlchemy repository with context-aware session management.

Pass model (a Struct-based BaseModel) to __init__; the repository uses the compiled SA class for queries and returns the Struct directly.

Parameters:

• session_manager (SessionManager)

• model (type[Any])

async on_transaction_committed(events)

Handle post-commit mutation events (cache invalidation hook).

Parameters:

events (tuple[MutationEvent, ...])

Return type:

None

class loom.core.repository.sqlalchemy.SessionManager(url, *, echo=False, pool_pre_ping=True, pool_size=10, max_overflow=20, pool_timeout=30, pool_recycle=1800, connect_args=None, inject_trace_id=True, **engine_kwargs)

Bases: object

Async SQLAlchemy session manager with pooling support.

Parameters:

• url (str)

• echo (bool)

• pool_pre_ping (bool)

• pool_size (int | None)

• max_overflow (int | None)

• pool_timeout (int | None)

• pool_recycle (int | None)

• connect_args (dict[str, object] | None)

• inject_trace_id (bool)

• engine_kwargs (object)

session()

Yield a scoped async session that is automatically closed on exit.

Yields:

An AsyncSession bound to the managed engine.

Return type:

AsyncIterator[sqlalchemy.ext.asyncio.AsyncSession]

async dispose()

Dispose of the engine and release all pooled connections.

Return type:

None

property engine: sqlalchemy.ext.asyncio.AsyncEngine

The underlying async SQLAlchemy engine.

property session_factory: sqlalchemy.ext.asyncio.async_sessionmaker.sqlalchemy.ext.asyncio.AsyncSession

The configured async session factory bound to the engine.

class loom.core.repository.sqlalchemy.SupportsPostCommit(*args, **kwargs)

Bases: Protocol

Protocol for objects that react to committed transactions.

class loom.core.repository.sqlalchemy.TimestampMixin

Bases: object

Created/updated timestamps for all persistent entities.

loom.core.repository.sqlalchemy.build_repository_registration_module(session_manager, models)

Build a DI module that registers model repositories and optional contracts.

Custom repositories declared with repository_for() are preferred. Models without a custom registration use RepositorySQLAlchemy.

Parameters:

• session_manager (SessionManager)

• models (Sequence[type[BaseModel]])

Return type:

Callable[[LoomContainer], None]

loom.core.repository.sqlalchemy.get_repository_registration(model)

Return the custom repository registration for model when present.

Parameters:

model (type[BaseModel])

Return type:

RepositoryRegistration | None

loom.core.repository.sqlalchemy.repository_for(model, *, contract=None)

Register a SQLAlchemy repository implementation for model.

Parameters:

• model (type[BaseModel]) – Struct model handled by the repository.

• contract (object | None) – Optional DI contract bound to the same repository instance.

Raises:

RuntimeError – If another repository is already registered for the same model or contract.

Return type:

Callable[[type[RepositorySQLAlchemy[Any, Any]]], type[RepositorySQLAlchemy[Any, Any]]]

loom.core.repository.sqlalchemy.transactional(method)

Create a single transaction boundary for service/orchestrator use cases.

Parameters:

method (Callable[[Concatenate[Any, ~P]], Awaitable[T]])

Return type:

Callable[[Concatenate[Any, ~P]], Awaitable[T]]

loom.core.repository.sqlalchemy.with_session_scope(method)

Inject repository-managed session into custom repository methods.

Parameters:

method (Callable[[...], Awaitable[R]])

Return type:

Callable[[…], Awaitable[R]]