loom.core.repository.abc.query

Functions

build_page_result(items, total_count, ...)

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

Classes

CursorResult(*, items, next_cursor, has_next)	Result of a cursor-paginated query.
FilterGroup(filters[, op])	A group of filter conditions combined with AND or OR logic.
FilterOp(value)	Filter operator applied to a single field.
FilterParams(*[, filters])	Generic filter container for list queries.
FilterSpec(field, op[, value])	A single field filter condition.
PageParams(*[, page, limit])	Pagination parameters for list queries.
PageResult(*, items, total_count, page, ...)	Paginated result set returned by list queries.
PaginationMode(value)	Pagination strategy for list and query operations.
QuerySpec([filters, sort, pagination, ...])	Structured query contract for list operations.
SortSpec(field[, direction])	A single sort directive applied to a query.

class loom.core.repository.abc.query.PaginationMode(value)

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.query.FilterOp(value)

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.query.FilterSpec(field, op, value=None)

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.query.FilterGroup(filters, op='AND')

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.query.SortSpec(field, direction='ASC')

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

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

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.query.CursorResult(*, items, next_cursor, has_next)

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.query.PageParams(*, page=1, limit=50)

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.query.FilterParams(*, filters=<factory>)

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.query.PageResult(*, items, total_count, page, limit, has_next)

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

loom.core.repository.abc.query.build_page_result(items, total_count, page_params)

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]