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.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.RepositorySQLAlchemy(session_manager, model)

Bases: SQLAlchemyCreateMixin[OutputT, IdT], SQLAlchemyReadMixin[OutputT, IdT], SQLAlchemyUpdateMixin[OutputT, IdT], SQLAlchemyDeleteMixin[OutputT, IdT], Readable[OutputT], Creatable[OutputT], Updatable[OutputT], Deletable[OutputT], Listable[OutputT], Countable[OutputT], 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.

class loom.core.repository.sqlalchemy.SQLAlchemyDefaultRepositoryBuilder(session_manager)

Bases: object

Default repository builder for SQLAlchemy-backed models.

A frozen dataclass that receives a SessionManager at construction time — injected by the SQLAlchemy DI module. The bootstrap and any other infrastructure layer must not construct this class directly; register it via the DI module so that the SessionManager singleton is shared across all repositories.

Parameters:

session_manager (SessionManager) – Shared SQLAlchemy session manager.

loom.core.repository.sqlalchemy.build_sqlalchemy_repository_registration_module(session_manager, models, *, logical_models=())

Build a DI module that registers model repositories and their capability bindings.

The module self-declares its infrastructure dependencies: it registers both SessionManager and DefaultRepositoryBuilder in the container so that the bootstrap does not need to know about SQLAlchemy internals.

To swap the default builder, register your own DefaultRepositoryBuilder in the container before loading this module — the module will not overwrite an existing registration.

Parameters:

• session_manager (SessionManager)

• models (Sequence[type[BaseModel]])

• logical_models (Sequence[type[Any]])

Return type:

Callable[[LoomContainer], None]

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