loom.core.repository.abc¶

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

Bases: Struct, 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: object

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¶: SQL LIKE pattern (case-sensitive).

ILIKE¶: SQL LIKE pattern (case-insensitive).

IS_NULL¶: Field is NULL.

EXISTS¶: Related collection is non-empty (subquery).

NOT_EXISTS¶: Related collection is empty (subquery).

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

Bases: Struct

Generic filter container for list queries.

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

filters¶

Flat key-value mapping passed as filter_by kwargs to SQLAlchemy.

Type:: dict[str, Any]

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

Bases: object

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: Struct

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: Struct, 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. Compatible with all backends.

CURSOR¶: Keyset (cursor) pagination. Performant at scale; requires a stable sort order with a tie-breaker.

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

Bases: object

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.RepoFor(*args, **kwargs)[source]¶

Bases: Protocol[ModelT]

Model-centric repository protocol used for UseCase dependency injection.

This protocol intentionally focuses on the generic CRUD surface used by most UseCases. Infrastructure bindings may provide a RepositorySQLAlchemy instance or any compatible implementation.

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

Fetch one entity by id.

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

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.

Return type:: int

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

Fetch entities with 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]

async create(data)[source]¶

Persist one entity.

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

async update(obj_id, data)[source]¶

Update one entity.

Parameters:

obj_id (Any)
data (Struct)

Return type:

ModelT | None

async delete(obj_id)[source]¶

Delete one entity.

Parameters:: obj_id (Any)
Return type:: bool

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: object

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")

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]