Repository

class alchemiq.Repository(model=None, *, cache=None, cache_ttl=None)

Bases: Generic[M]

Data-access surface for one model.

Instantiate ad-hoc with Repository(Model) or subclass to attach behaviour:

E.g.:

# ad-hoc (no subclass)
users = Repository(User)
user = await users.create(name="Ada", email="ada@x.io")

# subclass with cache
class UserRepository(Repository[User]):
    cache = True
    cache_ttl = 300

repo = UserRepository()
user = await repo.get(1)
await repo.update(1, name="Ada Lovelace")

The repository delegates query building to QuerySet and write operations to SQLAlchemy sessions managed by UnitOfWork (or its own session_scope).

Parameters:

• model (type[M])

• cache (Any)

• cache_ttl (int | None)

filter(*args, **lookups)

Return a QuerySet filtered by the given lookup expressions.

Parameters:

• args (Any)

• lookups (Any)

Return type:

QuerySet

exclude(*args, **lookups)

Return a QuerySet excluding rows matching the given lookup expressions.

Parameters:

• args (Any)

• lookups (Any)

Return type:

QuerySet

order_by(*fields)

Return a QuerySet ordered by fields (prefix - for descending).

Parameters:

fields (str)

Return type:

QuerySet

limit(n)

Return a QuerySet capped at n rows.

Parameters:

n (int)

Return type:

QuerySet

offset(n)

Return a QuerySet starting at row n (0-based).

Parameters:

n (int)

Return type:

QuerySet

distinct()

Return a QuerySet that emits a SELECT DISTINCT.

Return type:

QuerySet

only(*fields)

Return a QuerySet that loads only the named columns (deferred load for others).

Parameters:

fields (str)

Return type:

QuerySet

select_related(*names)

Return a QuerySet that JOIN-loads the named relationships (one SQL query).

Parameters:

names (str)

Return type:

QuerySet

prefetch_related(*names)

Return a QuerySet that SELECT-IN-loads the named relationships (extra query per rel).

Parameters:

names (str)

Return type:

QuerySet

with_deleted()

Return a QuerySet that includes soft-deleted rows alongside live ones.

Return type:

QuerySet

only_deleted()

Return a QuerySet scoped to soft-deleted (tombstoned) rows only.

Return type:

QuerySet

async get(*args, **lookups)

Fetch exactly one row matching lookups.

E.g.:

user = await users.get(id=1)
user = await users.get(email="ada@x.io")

Parameters:

• lookups (Any) – column-equality filters; arbitrary SQLAlchemy expressions accepted as positional args.

• args (Any)

Returns:

the matched model instance.

Raises:

• NotFoundError – if no row matches.

• MultipleResultsFound – if more than one row matches.

Return type:

M

async get_or_none(*args, **lookups)

Fetch the row matching lookups, or None if absent.

E.g.:

user = await users.get_or_none(id=42)
if user is None:
    ...  # not found

Parameters:

• lookups (Any) – column-equality filters; arbitrary SQLAlchemy expressions accepted as positional args.

• args (Any)

Returns:

the matched model instance, or None.

Raises:

MultipleResultsFound – if more than one row matches.

Return type:

M | None

async first()

Return the first row in the current ordering (LIMIT 1), or None if empty.

With no order_by chained, no ORDER BY is emitted, so which row is “first” is database-defined; chain order_by for a deterministic result.

Note

There is no default ORDER BY. For a stable “first” result, always chain order_by before calling this method.

Return type:

M | None

async last()

Return the last row in the current ordering, or None if empty.

Reverses each chained order_by direction; with no ordering, falls back to descending primary key.

Return type:

M | None

async all()

Return all rows as a list.

Return type:

list[M]

async exists(*args, **lookups)

Return True if at least one row matches the optional lookups.

Parameters:

• args (Any)

• lookups (Any)

Return type:

bool

async count(*args, **lookups)

Return the number of rows matching the optional lookups.

Parameters:

• args (Any)

• lookups (Any)

Return type:

int

async aggregate(**exprs)

Compute aggregate expressions (Count, Sum, Avg, Min, Max) over the table.

Parameters:

exprs (Aggregate)

Return type:

dict[str, Any]

async explain(*, analyze=False, format='text')

Return the query plan for the current QuerySet (PostgreSQL only).

Parameters:

• analyze (bool)

• format (Literal['text', 'json'])

Return type:

str | list[Any]

async paginate(page=1, size=20)

Return a Page for the given 1-based page number and size.

Parameters:

• page (int)

• size (int)

Return type:

Page[M]

async cursor_paginate(*, size=20, after=None, before=None)

Return a CursorPage navigating forward (after) or backward (before).

Parameters:

• size (int)

• after (str | None)

• before (str | None)

Return type:

CursorPage[M]

async create(**values)

Instantiate the model from values, persist it, and return the new instance.

Parameters:

values (Any)

Return type:

M

async add(obj)

Persist an already-constructed model instance and return it (fires pre/post_create).

Parameters:

obj (M)

Return type:

M

async update(id, *, expected_version=None, **changes)

Apply changes to the row id and return the refreshed instance.

Executes within the ambient UnitOfWork if one is active, otherwise in its own autocommit transaction. Fires the pre_update / post_update signals around the flush.

E.g.:

user = await users.update(3, age=26)

# optimistic concurrency control:
version = alchemiq.version_of(user)
await users.update(3, expected_version=version, name="Ada")

Parameters:

• id (Any) – primary key of the row to update.

• expected_version (int | None) – when given, the row’s _version must equal this value or ConcurrentModificationError is raised and nothing is written. Read it with version_of().

• changes (Any) – column=value pairs to assign; each is validated as on assignment.

Returns:

the refreshed model instance.

Raises:

• NotFoundError – if no row with id exists.

• ConcurrentModificationError – if expected_version did not match, or a concurrent flush detected a stale version.

Return type:

M

See also

Repository.bulk_update() - set-based, no per-row signals.

async delete(id, *, expected_version=None)

Delete the row identified by id.

Soft-delete models stamp deleted_at (no physical row removal); others issue a physical DELETE. Fires the pre_delete / post_delete signals in both cases. Supports optimistic locking via expected_version.

E.g.:

# physical delete (non-soft-delete model)
await posts.delete(5)

# optimistic concurrency control on a soft-delete model
await repo.delete(20, expected_version=1)

Parameters:

• id (Any) – primary key of the row to delete.

• expected_version (int | None) – when given, the row’s _version must equal this value or ConcurrentModificationError is raised and nothing is written.

Raises:

• NotFoundError – if no live row with id exists (for soft-delete models, an already-deleted row also counts as not found).

• ConcurrentModificationError – if expected_version did not match, or a concurrent flush detected a stale version.

Return type:

None

Note

For soft-delete models, delete() stamps deleted_at and leaves the physical row intact. Use Repository.hard_delete() to remove it unconditionally, or Repository.restore() to reverse the deletion.

async restore(id)

Clear deleted_at on a soft-deleted row, returning it to the live set.

Fires the pre_update / post_update signals around the flush.

E.g.:

await repo.restore(1)

Parameters:

id (Any) – primary key of the tombstoned row to restore.

Returns:

the restored model instance.

Raises:

• ConfigError – if the model does not have soft-delete enabled.

• NotFoundError – if no tombstone (soft-deleted row) with id exists.

Return type:

M

async hard_delete(id)

Physically DELETE the row identified by id, regardless of soft-delete status.

Bypasses the soft-delete predicate: removes the physical row even if it has a deleted_at stamp. Fires the pre_delete / post_delete signals.

E.g.:

await repo.hard_delete(31)

Parameters:

id (Any) – primary key of the row to delete (live or tombstoned).

Raises:

NotFoundError – if no row with id exists (live or tombstoned).

Return type:

None

See also

Repository.delete() - respects soft-delete; stamps deleted_at instead of issuing DELETE.

async bulk_create(objs)

Persist multiple instances in a single flush and return the inserted objects.

Fires no per-row signals and writes no outbox entries. Prefer this over repeated Repository.create() calls when inserting many rows at once.

E.g.:

rows = await repo.bulk_create([
    Item(id=10, name="x", age=1),
    Item(id=11, name="y", age=2),
])

Parameters:

objs (Iterable[M]) – iterable of model instances to insert.

Returns:

the list of inserted objects (same order as input); [] if objs is empty.

Return type:

list[M]

See also

Repository.bulk_upsert() - idempotent INSERT ... ON CONFLICT.

async bulk_upsert(objs, *, conflict=None, update_fields=None, ignore_conflicts=False)

Idempotent batch INSERT ... ON CONFLICT (PostgreSQL). Returns the affected rowcount.

conflict defaults to the PK column(s); update_fields defaults to all sent non-conflict columns. ignore_conflicts=True emits DO NOTHING instead of DO UPDATE. Fires no signals and writes no outbox (like all bulk operations). Operates on physical rows - the soft-delete predicate is not applied.

E.g.:

n = await repo.bulk_upsert([User(id=1, email="a@x.c", name="A")])

# override the conflict column:
await repo.bulk_upsert(
    [User(id=1, email="dup@x.c", name="first")],
    conflict=["email"],
)

Parameters:

• objs (Iterable[M]) – iterable of model instances to upsert; empty -> returns 0.

• conflict (Sequence[str] | None) – columns that identify a conflict; defaults to the primary key.

• update_fields (Sequence[str] | None) – columns to overwrite on conflict; defaults to all non-conflict columns present in the batch.

• ignore_conflicts (bool) – when True, conflicting rows are silently skipped (DO NOTHING); update_fields is ignored.

Returns:

number of rows affected (inserted + updated).

Return type:

int

See also

Repository.bulk_create() - plain insert without conflict handling.

async get_or_create(defaults=None, **lookups)

Fetch the row matching lookups, or create it from lookups + defaults.

E.g.:

obj, created = await repo.get_or_create(
    id=1, email="a@b.c", defaults={"name": "Ann"}
)

Parameters:

• defaults (dict[str, Any] | None) – extra fields used only when creating (not matched against existing rows).

• lookups (Any) – column-equality filters that identify the row.

Returns:

(obj, created) - True when a new row was inserted.

Return type:

tuple[M, bool]

Warning

NOT atomic outside a UnitOfWork. The lookup and the create run in separate transactions, so a concurrent insert between them can surface an integrity error. Wrap in async with UnitOfWork(): for atomicity.

async update_or_create(defaults=None, **lookups)

Update the row matching lookups (with defaults), or create it.

E.g.:

obj, created = await repo.update_or_create(id=2, defaults={"name": "new"})

Parameters:

• defaults (dict[str, Any] | None) – fields to write when updating or creating.

• lookups (Any) – column-equality filters that identify the row.

Returns:

(obj, created) - True when a new row was inserted.

Return type:

tuple[M, bool]

Warning

Best-effort atomic outside a UnitOfWork. If the matched row is concurrently deleted between the lookup and the update, the NotFoundError is caught and the call falls through to create(). Wrap in async with UnitOfWork(): for true atomicity.

async bulk_update(objs, fields)

Bulk UPDATE by PK for the given fields. Returns len(items), not DB rowcount.

Uses SQLAlchemy’s bulk-update-by-PK path: a single UPDATE per batch rather than one flush per row. Objects must have PK + field attributes accessible (no DetachedInstanceError) - prefer calling inside a UnitOfWork.

E.g.:

n = await repo.bulk_update(rows, fields=["age"])

Parameters:

• objs (Iterable[M]) – iterable of model instances whose PKs identify the rows to update.

• fields (Sequence[str]) – column names to write from each object.

Returns:

number of objects submitted (len(items)); rows absent from the DB are silently skipped by SQLAlchemy’s bulk path - do not rely on this count to detect missing PKs.

Return type:

int

See also

Repository.bulk_upsert() - insert-or-update in one statement.

Warning

Bypasses the optimistic-lock version check and increment (set-based, no per-row ORM flush). Fires no per-row signals.

async update_all(**changes)

Apply changes to every row without requiring a preceding filter().

Explicit full-table escape hatch: filter().update() refuses an unfiltered call. That filtered variant limits lookups to own columns; relationship traversal raises QueryError. Bypasses the optimistic-lock version check and increment (set-based, no per-row ORM flush). Fires no per-row signals.

Parameters:

changes (Any) – column=value pairs to assign across all rows.

Returns:

number of rows updated.

Return type:

int

Warning

This method updates every row in the table. There is no WHERE clause unless you call filter().update() instead. Double-check that a full-table update is intentional before using this method.

async delete_all()

Delete every row without requiring a preceding filter().

Explicit full-table escape hatch: filter().delete() refuses an unfiltered call. That filtered variant limits lookups to own columns; relationship traversal raises QueryError. Soft-delete models stamp deleted_at; others issue a physical DELETE. Bypasses the optimistic-lock version check and increment (set-based, no per-row ORM flush). Fires no per-row signals.

E.g.:

n = await repo.delete_all()

Returns:

number of rows deleted (or soft-deleted).

Return type:

int

Warning

This method deletes every row in the table. There is no WHERE clause unless you call filter().delete() instead. Double-check that a full-table delete is intentional before using this method.

async cache_clear()

Invalidate the entire model cache (version bump + flush precise object keys).

Return type:

None

async cache_evict(pk)

Invalidate one object (drop its precise key + bump version).

Parameters:

pk (Any)

Return type:

None

class alchemiq.Page(items, total, page, size, pages, has_next, has_prev)

Bases: Generic[M]

Offset-based pagination result with total-row metadata.

Parameters:

• items (list[M])

• total (int)

• page (int)

• size (int)

• pages (int)

• has_next (bool)

• has_prev (bool)

classmethod build(items, total, page, size)

Construct a Page from raw query results, computing pages and nav flags.

Parameters:

• items (list[M])

• total (int)

• page (int)

• size (int)

Return type:

Page[M]

class alchemiq.CursorPage(items, next_cursor, prev_cursor, has_next, has_prev)

Bases: Generic[M]

Keyset/cursor pagination result with opaque forward and backward cursors.

Parameters:

• items (list[M])

• next_cursor (str | None)

• prev_cursor (str | None)

• has_next (bool)

• has_prev (bool)