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], SQLAlchemyBulkCreateMixin[OutputT, IdT], SQLAlchemyReadMixin[OutputT, IdT], SQLAlchemyUpdateMixin[OutputT, IdT], SQLAlchemyDeleteMixin[OutputT, IdT], Readable[OutputT], Creatable[OutputT], BulkCreatable[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])

has_caller_scoped_session()

Whether a read would run inside a session bound to the caller’s context.

True inside a @transactional scope or a unit of work: the session was opened by the caller, dies when the caller unwinds, and holds writes only that caller can see. False when the repository would open and close a session of its own for the call.

Implements SupportsCallerScopedSession. loom.core.transaction.in_atomic_transaction() is the neutral marker this capability used to be a placeholder for, but it answers a different question — whether any atomic transaction is open — not whether this session is scoped to the caller, which is what governs coalescing safely detaching a read into its own task. Keep using this method for that; it is not superseded.

Returns:

True when a caller-scoped session is bound to the context.

Return type:

bool

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)

classmethod from_config(config, *, inject_trace_id=True, **engine_kwargs)

Build a session manager from a resolved SQLAlchemy config mapping.

Parameters:

• config (Mapping[str, Any]) – Resolved config mapping containing a url entry and optional pool tuning keys.

• inject_trace_id (bool) – When True, prefixes SQL statements with the active trace id when available.

• **engine_kwargs (object) – Additional keyword arguments forwarded to the async engine constructor.

Returns:

A configured SessionManager.

Raises:

ValueError – If url is missing or empty.

Return type:

SessionManager

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.

When a session is already active (an outer @transactional call or a unit of work driven by the executor) the method joins it and nothing else happens: the owner of that session runs the post-commit hooks. When the decorator opens the session itself it commits, then runs on_transaction_committed(pending) on the owner and on its SupportsPostCommit attributes through the post-commit channel (loom.core.engine.post_commit) after the session is closed. The decorator owns that channel whenever it owns the session: a channel bound by an outer context is left untouched and restored afterwards, so a committed transaction always drains its own actions.

Parameters:

method (Callable[[Concatenate[Any, ~P]], Awaitable[T]]) – Async method of an object exposing session_manager.

Returns:

The wrapped method.

Raises:

• TypeError – If applied to a repository method or the owner has no session_manager with a session() context manager.

• PostCommitError – If a hook failed after the commit.

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