loom.core.repository.abc¶

class loom.core.repository.abc.Cursor(backend, keys, tie_breaker)[source]¶

Bases: LoomFrozenStruct

Decoded cursor position.

Parameters:

backend (str) – Name of the backend that issued the token.
keys (tuple[object, ...]) – Sort key values of the last row on the previous page, in sort order.
tie_breaker (object) – Primary key of that row.

class loom.core.repository.abc.CursorResult(*, items, next_cursor, has_next)[source]¶

Bases: LoomFrozenStruct, Generic[OutputT]

Result of a cursor-paginated query.

Parameters:

items (tuple[OutputT, ...])
next_cursor (str | None)
has_next (bool)

items¶

Entities for the current page.

Type:: tuple[loom.core.repository.abc.query.OutputT, …]

next_cursor¶

Opaque token for the next page, or None if this is the last page.

Type:: str | None

has_next¶

True if more items follow.

Type:: bool

class loom.core.repository.abc.FilterGroup(filters, op='AND')[source]¶

Bases: LoomFrozenStruct

A group of filter conditions combined with AND or OR logic.

Parameters:

filters (tuple[FilterSpec, ...]) – Filter conditions to combine.
op (Literal['AND', 'OR']) – Logical operator: "AND" (default) or "OR".

Example:

FilterGroup(
    filters=(
        FilterSpec("price", FilterOp.GTE, 10.0),
        FilterSpec("price", FilterOp.LTE, 100.0),
    ),
    op="AND",
)

class loom.core.repository.abc.FilterOp(value)[source]¶

Bases: StrEnum

Filter operator applied to a single field.

EQ¶: Exact equality.

NE¶: Inequality.

GT¶: Greater than.

GTE¶: Greater than or equal.

LT¶: Less than.

LTE¶: Less than or equal.

IN¶: Value is in a collection.

LIKE¶: Text pattern match (case-sensitive).

ILIKE¶: Text pattern match (case-insensitive).

IS_NULL¶: Field has no value.

EXISTS¶: Related collection is non-empty.

NOT_EXISTS¶: Related collection is empty.

class loom.core.repository.abc.FilterParams(*, filters=<factory>)[source]¶

Bases: LoomFrozenStruct

Generic filter container for list queries.

Parameters:: filters (dict[str, Any])

filters¶

Flat mapping of field name to expected value; each entry is an equality match combined with AND. The backend translates it into its own query mechanism.

Type:: dict[str, Any]

class loom.core.repository.abc.FilterSpec(field, op, value=None)[source]¶

Bases: LoomFrozenStruct

A single field filter condition.

Parameters:

field (str) – Dot-separated field path (e.g. "price" or "category.name").
op (FilterOp) – Comparison operator.
value (Any) – Value to compare against. Ignored for IS_NULL, EXISTS, and NOT_EXISTS.

Example:

FilterSpec(field="price", op=FilterOp.GTE, value=10.0)

class loom.core.repository.abc.PageParams(*, page=1, limit=50)[source]¶

Bases: LoomFrozenStruct

Pagination parameters for list queries.

Parameters:

page (int)
limit (int)

page¶

1-based page number.

Type:: int

limit¶

Maximum number of items per page (1-1000).

Type:: int

property offset: int¶: Calculate the zero-based row offset for the current page.

class loom.core.repository.abc.PageResult(*, items, total_count, page, limit, has_next)[source]¶

Bases: LoomFrozenStruct, Generic[OutputT]

Paginated result set returned by list queries.

Parameters:

items (tuple[OutputT, ...])
total_count (int)
page (int)
limit (int)
has_next (bool)

items¶

Tuple of entity output structs for the current page.

Type:: tuple[loom.core.repository.abc.query.OutputT, …]

total_count¶

Total number of matching entities across all pages.

Type:: int

page¶

Current page number (1-based).

Type:: int

limit¶

Maximum items per page.

Type:: int

has_next¶

True if more pages follow.

Type:: bool

class loom.core.repository.abc.PaginationMode(value)[source]¶

Bases: StrEnum

Pagination strategy for list and query operations.

OFFSET¶: Classic page+limit pagination with a total count. Convenient but optional: computing total_count and skipping to arbitrary offsets is not efficiently supported by every backend, so a backend may decline this mode.

CURSOR¶: Keyset (cursor) pagination. The portable mode — supported by every backend and performant at scale; requires a stable sort order with a tie-breaker.

class loom.core.repository.abc.QueryCompiler(*args, **kwargs)[source]¶

Bases: Protocol[FilterT_co, SortT_co]

Compiles query parts into a backend’s native expressions.

FilterT_co is the native filter expression type (a SQLAlchemy clause, a Mongo filter document, …) and SortT_co the native sort expression type.

compile_filter(group)[source]¶

Compile a filter group into a native filter expression.

Parameters:: group (FilterGroup) – Flat AND/OR group of field conditions.
Returns:: The backend’s filter expression.
Raises:: UnsupportedQuery – If a field or a FilterOp cannot be served by the backend; the reason names both the operator and the backend.
Return type:: FilterT_co

compile_sort(sort)[source]¶

Compile sort directives into a native sort expression.

Parameters:: sort (tuple[SortSpec, ...]) – Ordered sort directives.
Returns:: The backend’s sort expression.
Raises:: UnsupportedQuery – If a sort field cannot be served by the backend.
Return type:: SortT_co

class loom.core.repository.abc.QuerySpec(filters=None, sort=(), pagination=PaginationMode.OFFSET, limit=50, page=1, cursor=None)[source]¶

Bases: LoomFrozenStruct

Structured query contract for list operations.

Replaces the flat FilterParams dict with an explicit, type-safe representation. The repository implementation compiles this into backend-specific clauses at query time.

Parameters:

filters (FilterGroup | None) – Optional filter group applied to the query.
sort (tuple[SortSpec, ...]) – Ordered tuple of sort directives.
pagination (PaginationMode) – Pagination strategy. Defaults to OFFSET.
limit (int) – Maximum number of items per page (1-1000).
page (int) – 1-based page number (only for OFFSET mode).
cursor (str | None) – Opaque cursor token (only for CURSOR mode).

Example:

QuerySpec(
    filters=FilterGroup(
        filters=(FilterSpec("price", FilterOp.GTE, 10.0),),
    ),
    sort=(SortSpec("name"),),
    pagination=PaginationMode.OFFSET,
    limit=20,
    page=1,
)

class loom.core.repository.abc.BulkCreatable(*args, **kwargs)[source]¶

Bases: Protocol[ModelT]

Repository capability: persist several new entities in one round trip.

async create_many(data)[source]¶

Persist every entity in data and return them in input order.

Parameters:: data (Sequence[Struct])
Return type:: tuple[ModelT, …]

class loom.core.repository.abc.Countable(*args, **kwargs)[source]¶

Bases: Protocol[ModelT]

Repository capability: existence checks and counting.

async exists_by(field, value)[source]¶

Return True if any entity matches field == value.

Parameters:

field (str)
value (Any)

Return type:

bool

async count()[source]¶

Return the total number of entities.

Return type:: int

class loom.core.repository.abc.Creatable(*args, **kwargs)[source]¶

Bases: Protocol[ModelT]

Repository capability: persist new entities.

Example:

class CreateOrderUseCase(UseCase[Order, Order, Creatable[Order]]):
    async def execute(self, cmd: CreateOrderCommand = Input()) -> Order:
        return await self.main_repo.create(cmd)

async create(data)[source]¶

Persist one entity and return the persisted result.

Parameters:: data (Struct)
Return type:: ModelT

class loom.core.repository.abc.Deletable(*args, **kwargs)[source]¶

Bases: Protocol[ModelT]

Repository capability: remove entities.

async delete(obj_id)[source]¶

Delete one entity by primary key.

Returns:: True if the entity existed and was deleted.
Parameters:: obj_id (Any)
Return type:: bool

class loom.core.repository.abc.Listable(*args, **kwargs)[source]¶

Bases: Protocol[ModelT]

Repository capability: paginated and structured listing.

async list_paginated(page_params, filter_params=None, profile='default')[source]¶

Fetch entities with offset pagination.

Parameters:

page_params (PageParams)
filter_params (FilterParams | None)
profile (str)

Return type:

PageResult[ModelT]

async list_with_query(query, profile='default')[source]¶

Fetch entities using a structured query (offset or cursor mode).

Parameters:

query (QuerySpec)
profile (str)

Return type:

PageResult[ModelT] | CursorResult[ModelT]

class loom.core.repository.abc.Readable(*args, **kwargs)[source]¶

Bases: Protocol[ModelT]

Repository capability: fetch individual entities.

Implement this to expose single-entity read operations.

Example:

class IProductRepository(Readable[Product], Protocol):
    async def find_by_sku(self, sku: str) -> Product | None: ...

async get_by_id(obj_id, profile='default')[source]¶

Fetch one entity by primary key.

Parameters:

obj_id (Any)
profile (str)

Return type:

ModelT | None

async get_by(field, value, profile='default')[source]¶

Fetch one entity by arbitrary field.

Parameters:

field (str)
value (Any)
profile (str)

Return type:

ModelT | None

class loom.core.repository.abc.RepoFor(*args, **kwargs)[source]¶

Bases: Readable[ModelT], Creatable[ModelT], Updatable[ModelT], Deletable[ModelT], Listable[ModelT], Countable[ModelT], Protocol[ModelT]

Full-surface repository protocol for standard CRUD use cases.

Composes all capability protocols. Use this as the default RepoT when a use case needs the complete CRUD surface. For use cases that only need a subset of operations, prefer the specific capability protocol (Readable, Creatable, etc.) to respect the Interface Segregation Principle.

Example:

# Full surface — no third param needed, RepoFor[Any] is the default
class CreateProductUseCase(UseCase[Product, Product]):
    async def execute(self, cmd: CreateProductCommand = Input()) -> Product:
        return await self.main_repo.create(cmd)

# Specific capability — only what the use case needs
class GetProductUseCase(UseCase[Product, Product | None, Readable[Product]]):
    async def execute(self, product_id: int) -> Product | None:
        return await self.main_repo.get_by_id(product_id)

class loom.core.repository.abc.Updatable(*args, **kwargs)[source]¶

Bases: Protocol[ModelT]

Repository capability: mutate existing entities.

async update(obj_id, data)[source]¶

Update one entity by primary key.

Parameters:

obj_id (Any)
data (Struct)

Return type:

ModelT | None

class loom.core.repository.abc.Repository(*args, **kwargs)[source]¶

Bases: RepositoryRead[OutputT, IdT], RepositoryWrite[OutputT, CreateT, UpdateT, IdT], Protocol[OutputT, CreateT, UpdateT, IdT]

Combined read-write repository protocol.

class loom.core.repository.abc.RepositoryRead(*args, **kwargs)[source]¶

Bases: Protocol[OutputT, IdT]

Protocol for read-only repository operations (get by id and paginated listing).

async get_by_id(obj_id, profile='default')[source]¶

Fetch a single entity by its primary key.

Parameters:

obj_id (IdT) – Primary key of the entity.
profile (str) – Loading profile name for eager-load options.

Returns:

The entity output struct, or None if not found.

Return type:

OutputT | None

async get_by(field, value, profile='default')[source]¶

Fetch a single entity by an arbitrary field.

Parameters:

field (str) – Entity field name used in the equality lookup.
value (Any) – Value to compare against.
profile (str) – Loading profile name for eager-load options.

Returns:

The entity output struct, or None if not found.

Return type:

OutputT | None

async exists_by(field, value)[source]¶

Check whether any entity exists matching field == value.

Parameters:

field (str)
value (Any)

Return type:

bool

async count()[source]¶

Return the total number of entities in the repository.

Returns:: Total row count as an integer.
Return type:: int

async list_paginated(page_params, filter_params=None, profile='default')[source]¶

Fetch a paginated list of entities.

Parameters:

page_params (PageParams) – Pagination parameters (page and limit).
filter_params (FilterParams | None) – Optional filter criteria.
profile (str) – Loading profile name for eager-load options.

Returns:

A PageResult with the matching items and pagination metadata.

Return type:

PageResult[OutputT]

async list_with_query(query, profile='default')[source]¶

Fetch entities using a structured QuerySpec.

Supports both offset and cursor pagination, structured filters, and explicit sort directives. The concrete return type depends on query.pagination:

PaginationMode.OFFSET → PageResult
PaginationMode.CURSOR → CursorResult

Parameters:

query (QuerySpec) – Structured query specification.
profile (str) – Loading profile name for eager-load options.

Returns:

A PageResult for offset queries or a CursorResult for cursor queries.

Return type:

PageResult[OutputT] | CursorResult[OutputT]

class loom.core.repository.abc.RepositoryWrite(*args, **kwargs)[source]¶

Bases: Protocol[OutputT, CreateT, UpdateT, IdT]

Protocol for write repository operations (create, update, delete).

async create(data)[source]¶

Persist a new entity.

Parameters:: data (CreateT) – Creation payload struct.
Returns:: The newly created entity output struct.
Return type:: OutputT

async update(obj_id, data)[source]¶

Apply a partial update to an existing entity.

Parameters:

obj_id (IdT) – Primary key of the entity to update.
data (UpdateT) – Partial update payload struct.

Returns:

The updated entity output struct, or None if not found.

Return type:

OutputT | None

async delete(obj_id)[source]¶

Delete an entity by its primary key.

Parameters:: obj_id (IdT) – Primary key of the entity to delete.
Returns:: True if the entity was deleted, False if not found.
Return type:: bool

class loom.core.repository.abc.SortSpec(field, direction='ASC')[source]¶

Bases: LoomFrozenStruct

A single sort directive applied to a query.

Parameters:

field (str) – Field name to sort by.
direction (Literal['ASC', 'DESC']) – "ASC" (default) or "DESC".

Example:

SortSpec(field="created_at", direction="DESC")

class loom.core.repository.abc.SupportsCallerScopedSession(*args, **kwargs)[source]¶

Bases: Protocol

Optional capability: report whether the caller owns the current session.

A repository that resolves its session from the calling context — a @transactional scope, a unit of work — runs its reads inside a session that is closed when that caller unwinds, and whose uncommitted writes are visible only to it. A wrapper that would otherwise detach a read from its caller must ask first: the cache layer skips its in-process coalescing while this returns True, because a coalesced load outlives the caller that started it and would be shared with another one.

A repository that does not implement this protocol is taken to own the scope of its own reads, which is what a repository opening and closing a session per call does.

This capability is provisional: it exists because each backend publishes its transaction through its own ContextVar. It is expected to be replaced by the neutral transaction scope, and is not a stable extension point to build on.

has_caller_scoped_session()[source]¶

Whether the current context holds a session owned by the caller.

Returns:: True inside a caller-owned transaction, False when the repository would open and close a session of its own.
Return type:: bool

exception loom.core.repository.abc.UnsupportedQuery(backend, model, reason)[source]¶

Bases: DomainError

Raised when a backend cannot serve a query because of its arguments.

A capability the repository class does not declare is absent from DI and from the generated routes; this error covers the gaps that depend on a run-time argument instead, such as a lookup on a field the backend can only reach through a scan.

Parameters:

backend (str) – Name of the persistence backend.
model (str) – Qualified name of the model being queried.
reason (str) – What the backend cannot do with the given arguments.

Return type:

None

loom.core.repository.abc.build_page_result(items, total_count, page_params)[source]¶

Construct a PageResult from a list of items and pagination metadata.

Parameters:

items (list[OutputT]) – Entity output structs for the current page.
total_count (int) – Total number of matching entities across all pages.
page_params (PageParams) – The pagination parameters used for this query.

Returns:

A populated PageResult instance.

Return type:

PageResult[OutputT]

loom.core.repository.abc.decode_cursor(token, backend, model, *, key_count=None)[source]¶

Decode a token issued by encode_cursor() for backend.

Parameters:

token (str) – Opaque token supplied by the client.
backend (str) – Name of the backend decoding the token.
model (str) – Qualified model name, used in the error.
key_count (int | None) – Number of sort keys the query orders by; when given, a token carrying a different number is rejected.

Returns:

The decoded Cursor.

Raises:

UnsupportedQuery – If the token is undecodable, in a legacy format, was issued by another backend or does not match the sort.

Return type:

Cursor

loom.core.repository.abc.encode_cursor(backend, keys, tie_breaker)[source]¶

Encode a cursor position into an opaque URL-safe token.

Parameters:

backend (str) – Name of the issuing backend.
keys (Sequence[object]) – Sort key values of the last row on the page, in sort order.
tie_breaker (object) – Primary key of that row.

Returns:

base64url token that decode_cursor() accepts for backend.

Raises:

TypeError – If a key has a type the token format cannot carry.

Return type:

str