Query: Q, QuerySet and aggregates

class alchemiq.Q(*args, **kwargs)

Bases: object

Django-style boolean predicate tree of (field__op, value) leaves.

Combine predicates with & (AND), | (OR), and ~ (NOT). Pass to QuerySet.filter() / Repository.filter() for type-safe filtering.

Supported lookup suffixes: __eq (default), __ne, __lt, __lte, __gt, __gte, __in, __not_in, __isnull, __contains, __icontains, __startswith, __endswith.

E.g.:

>>> q1 = Q(status="active") & Q(age__gte=18)
>>> q2 = ~Q(role__in=["admin", "staff"])

>>> q_or = Q(role="admin") | Q(role="staff")
>>> q_or.connector
'OR'

>>> neg = ~Q(deleted=True)
>>> neg.negated
True

Pass the composed predicate to filter() for execution:

active_adults = await (
    Repository(User)
    .filter(Q(status="active") & Q(age__gte=18))
    .all()
)

Parameters:

• args (Any)

• kwargs (Any)

to_data()

Serialize this Q to a compact, JSON-safe nested list.

Return type:

list[Any]

classmethod from_data(data, model, allow=None, deny=None)

Deserialize a Q from a Q.to_data() payload, validating fields against model.

Parameters:

• data (Any) – nested list produced by Q.to_data().

• model (type) – model class used to validate field names and resolve Python types.

• allow (set[str] | None) – explicit allow-list of field paths; required to permit traversal fields.

• deny (set[str] | None) – optional deny-list of field paths.

Raises:

• DeserializationError – if the payload structure is invalid.

• DisallowedFieldError – if a field is denied or not in the allow-list.

• UnknownOperatorError – if the payload contains an unrecognised lookup operator.

Return type:

Q

to_bytes()

Serialize this Q to compact UTF-8 JSON bytes.

Return type:

bytes

to_base64()

Serialize this Q to a urlsafe base64 string.

Return type:

str

classmethod from_bytes(data, model, allow=None, deny=None)

Deserialize a Q from UTF-8 JSON bytes produced by to_bytes().

Parameters:

• data (bytes)

• model (type)

• allow (set[str] | None)

• deny (set[str] | None)

Return type:

Q

classmethod from_base64(s, model, allow=None, deny=None)

Deserialize a Q from a urlsafe base64 string produced by to_base64().

Parameters:

• s (str)

• model (type)

• allow (set[str] | None)

• deny (set[str] | None)

Return type:

Q

class alchemiq.QuerySet(model, *, where=(), order=(), limit=None, offset=None, distinct=False, projection=None, select_related=(), prefetch_related=(), deleted='exclude', cache=None, cache_ttl=None)

Bases: object

Immutable, lazy query builder. Compiles to a SQLAlchemy Select (no I/O).

Parameters:

• model (type)

• where (tuple[Q, ...])

• order (tuple[str, ...])

• limit (int | None)

• offset (int | None)

• distinct (bool)

• projection (tuple[str, ...] | None)

• select_related (tuple[str, ...])

• prefetch_related (tuple[str, ...])

• deleted (DeletedMode)

• cache (Any)

• cache_ttl (int | None)

select_related(*names)

Join-load the named relationships via a SQL JOIN in the same query.

Uses SQLAlchemy joinedload; the related object is available on the result without triggering a lazy load. For collections, prefer QuerySet.prefetch_related() (a separate SELECT avoids row duplication from the JOIN).

E.g.:

track = await QuerySet(Track).select_related("artist").filter(id=1).first()
print(track.artist.name)  # no RelationNotLoaded

Parameters:

names (str) – relationship attribute names to join-load.

Return type:

QuerySet

See also

QuerySet.prefetch_related() - uses selectinload (separate SELECT).

prefetch_related(*names)

Eager-load the named relationships via a separate SELECT … IN (…) query.

Uses SQLAlchemy selectinload; the related collection is populated without a JOIN (no row duplication). For to-one relationships where you need the foreign object in the same query, prefer QuerySet.select_related().

E.g.:

artist = await QuerySet(Artist).prefetch_related("tracks").filter(id=1).first()
print([t.title for t in artist.tracks])  # no RelationNotLoaded

Parameters:

names (str) – relationship attribute names to selectin-load.

Return type:

QuerySet

See also

QuerySet.select_related() - uses joinedload (same-query JOIN).

with_deleted()

Include soft-deleted (tombstoned) rows in results.

By default, soft-delete models exclude tombstone rows (deleted_at IS NULL). Call this builder to include them as well.

E.g.:

all_docs = await QuerySet(Document).with_deleted().all()

Raises:

ConfigError – if the model does not have Meta.soft_delete = True.

Return type:

QuerySet

See also

QuerySet.only_deleted() - restrict to tombstones only.

only_deleted()

Restrict results to soft-deleted rows only (deleted_at IS NOT NULL).

E.g.:

tombstones = await QuerySet(Document).only_deleted().all()

Raises:

ConfigError – if the model does not have Meta.soft_delete = True.

Return type:

QuerySet

See also

QuerySet.with_deleted() - include both live and deleted rows.

filter(*args, **lookups)

Narrow results by AND-ing the given Q objects and keyword lookups.

Parameters:

• args (Q)

• lookups (Any)

Return type:

QuerySet

exclude(*args, **lookups)

Exclude rows matching the given Q objects or keyword lookups (NOT AND).

Parameters:

• args (Q)

• lookups (Any)

Return type:

QuerySet

order_by(*fields)

Set the ORDER BY clause. Prefix a field name with - for descending.

Parameters:

fields (str)

Return type:

QuerySet

limit(n)

Apply a LIMIT clause.

Parameters:

n (int)

Return type:

QuerySet

offset(n)

Apply an OFFSET clause.

Parameters:

n (int)

Return type:

QuerySet

distinct()

Deduplicate rows with SELECT DISTINCT.

Return type:

QuerySet

only(*fields)

Restrict the SELECT to fields (column projection).

Parameters:

fields (str)

Return type:

QuerySet

compile()

Return the SQLAlchemy Select statement without executing it.

Return type:

Any

async all()

Execute the query and return all matching model instances.

E.g.:

users = await QuerySet(User).filter(status="active").all()

Return type:

list[Any]

async first()

Execute with LIMIT 1 and return the first result, or None.

E.g.:

user = await QuerySet(User).order_by("created_at").first()

Return type:

Any | None

async count()

Return the number of rows matching the current filters.

Strips ORDER BY / LIMIT / OFFSET before wrapping in SELECT count(*) FROM (...) AS subquery.

E.g.:

total = await QuerySet(User).filter(status="active").count()

Return type:

int

async aggregate(**exprs)

Compute reduce-aggregates over the filtered set and return {alias: value}.

Each keyword argument is a Count / Sum / Avg / Min / Max expression. Inherits the current filters, soft-delete predicate, and any traversal joins. No GROUP BY; always one row. Sum / Avg / Min / Max over an empty set return None; Count returns 0.

E.g.:

result = await QuerySet(Order).filter(status="paid").aggregate(
    total=Sum("amount"),
    n=Count(),
)
print(result["total"], result["n"])

Parameters:

exprs (Aggregate) – alias=Aggregate(...) pairs.

Returns:

dict mapping each alias to its computed value.

Raises:

ValueError – if no expressions are given.

Return type:

dict[str, Any]

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

Run EXPLAIN on the compiled SELECT and return the query plan.

analyze=True executes the query for real timings (read-only SELECT inside a rolled-back transaction on PostgreSQL). format="json" returns the parsed JSON plan (a list); format="text" returns the plan as a str. Always hits the database - never cached. PostgreSQL only.

E.g.:

plan = await QuerySet(User).filter(status="active").explain(analyze=True)
print(plan)

Parameters:

• analyze (bool) – when True, run EXPLAIN ANALYZE (real execution).

• format (Literal['text', 'json']) – "text" (default) or "json".

Returns:

plan string (format="text") or parsed list (format="json").

Return type:

str | list[Any]

async exists()

Return True if at least one row matches the current filters.

E.g.:

if await QuerySet(User).filter(email="ada@x.io").exists():
    ...

Return type:

bool

async get(*args, **lookups)

Filter and return the single matching instance.

Fetches at most 2 rows to detect duplicates without a full scan. Accepts the same Q and keyword arguments as QuerySet.filter().

E.g.:

user = await QuerySet(User).get(id=42)

Raises:

• NotFoundError – if no rows match the current filters.

• MultipleResultsFound – if more than one row matches.

Parameters:

• args (Q)

• lookups (Any)

Return type:

Any

async get_or_none(*args, **lookups)

Filter and return the single matching instance, or None if not found.

Raises:

MultipleResultsFound – if more than one row matches.

Parameters:

• args (Q)

• lookups (Any)

Return type:

Any | None

async paginate(page=1, size=20)

Return a Page of results for page/size pagination.

Issues two queries: count() then a windowed all().

E.g.:

page = await QuerySet(User).order_by("id").paginate(page=1, size=20)
print(page.total, page.items)

Parameters:

• page (int) – 1-based page number.

• size (int) – number of items per page.

Raises:

ValueError – if page < 1 or size < 1.

Return type:

Page[Any]

Note

count() and all() run in two separate database sessions. A row inserted between the two queries may inflate total without appearing in items, or vice versa.

See also

QuerySet.cursor_paginate() - keyset pagination with no total.

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

Keyset (cursor) pagination with no total-count query.

Adds a PK tiebreaker to the effective order for a deterministic total order. after and before are mutually exclusive opaque tokens from CursorPage.next_cursor / CursorPage.prev_cursor.

E.g.:

p1 = await QuerySet(User).order_by("id").cursor_paginate(size=20)
if p1.has_next:
    p2 = await QuerySet(User).order_by("id").cursor_paginate(
        size=20, after=p1.next_cursor
    )

Parameters:

• size (int) – maximum number of items per page (must be >= 1).

• after (str | None) – opaque forward cursor token; fetch the page after this position.

• before (str | None) – opaque backward cursor token; fetch the page before this position.

Raises:

• ValueError – if size < 1 or both after and before are given.

• InvalidCursorError – if a cursor token is malformed or does not match the current order.

Return type:

CursorPage[Any]

See also

QuerySet.paginate() - offset pagination with a total count.

async update(**changes)

Execute a set-based UPDATE on own-column filters. Returns rowcount.

Raises:

QueryError – if no filter is set; use QuerySet.update_all() for an unguarded full-table update.

Parameters:

changes (Any)

Return type:

int

Note

The automatic soft-delete predicate does not count as a user filter.

async update_all(**changes)

Execute a SET-BASED UPDATE over EVERY matching row (no user filter required).

The explicit, lexically-distinct full-table escape hatch - update() refuses an unfiltered call so a missing .filter() cannot silently rewrite the table. Still honors the current deleted-mode (default EXCLUDE skips tombstones).

Parameters:

changes (Any)

Return type:

int

async delete()

Execute a set-based DELETE on own-column filters. Returns rowcount.

On a soft-delete model this stamps deleted_at (an UPDATE) instead of a physical DELETE.

Raises:

QueryError – if no filter is set; use QuerySet.delete_all() for an unguarded full-table delete.

Return type:

int

async delete_all()

Execute a SET-BASED DELETE over EVERY matching row (no user filter required).

Soft-delete models stamp deleted_at; others physically DELETE. The explicit full-table escape hatch - delete() refuses an unfiltered call. Honors the current deleted-mode (default EXCLUDE skips tombstones).

Return type:

int

async hard_delete()

Execute a physical SET-BASED DELETE on own-column filters. Returns rowcount.

Bypasses soft-delete: the rows are physically removed. Honors the current deleted-mode (default EXCLUDE purges live rows; only_deleted().hard_delete() purges tombstones; with_deleted().hard_delete() purges both).

Return type:

int

async last()

Return the last row according to the current ordering.

If an order_by is set, each field’s direction is reversed (- prefix toggled). If no ordering is set, falls back to descending PK. Delegates to first() on the reversed clone.

Return type:

Any | None

class alchemiq.Count(field='*', *, distinct=False)

Bases: Aggregate

Row count, optionally over a specific column with distinct deduplication.

Count() or Count("*") emits count(*); Count("field") emits count(field); Count("field", distinct=True) emits count(DISTINCT field). Raises QueryError for traversal fields (__ paths).

E.g.:

result = await QuerySet(Order).aggregate(
    n=Count(),
    unique_customers=Count("customer_id", distinct=True),
)

See also

QuerySet.aggregate() - pass aggregate expressions by alias.

Parameters:

• field (str)

• distinct (bool)

resolve(model)

Build the SQLAlchemy count element for model.

Parameters:

model (type)

Return type:

Any

class alchemiq.Sum(field)

Bases: Aggregate

Sum of a numeric column.

Parameters:

field (str)

class alchemiq.Avg(field)

Bases: Aggregate

Arithmetic mean of a numeric column.

Parameters:

field (str)

class alchemiq.Min(field)

Bases: Aggregate

Minimum value of a column.

Parameters:

field (str)

class alchemiq.Max(field)

Bases: Aggregate

Maximum value of a column.

Parameters:

field (str)

alchemiq.version_of(obj)

Return the optimistic-lock _version counter of a versioned model instance.

Read this value before an update to pass as expected_version for optimistic concurrency control.

E.g.:

user = await repo.get(id=1)
version = version_of(user)
await repo.update(1, expected_version=version, name="Ada")

Parameters:

obj (Any) – a model instance whose class has Meta.versioned = True.

Returns:

the current _version integer.

Raises:

ConfigError – if the model was not declared with Meta.versioned = True.

Return type:

int

See also

Repository.update() - accepts expected_version.