QueryBuilder

Fluent API for constructing and executing database queries. Instances are created by SqliterDB.select() -- you do not instantiate QueryBuilder directly.

results = db.select(User).filter(age__gt=18).order("name").fetch_all()

Source: sqliter/query/query.py

See also: Guide -- Filtering, Guide -- Ordering, Guide -- Field Control, Guide -- Caching

Type Parameters

T = TypeVar("T", bound=BaseDBModel)

class QueryBuilder(Generic[T]):
    ...

QueryBuilder is generic over T, the model class. All fetch methods return instances of T.

Type Aliases

FilterValue

The allowed types for filter values:

FilterValue = Union[
    str, int, float, bool, None,
    list[Union[str, int, float, bool]],
]

Constructor

def __init__(
    self,
    db: SqliterDB,
    model_class: type[T],
    fields: list[str] | None = None,
) -> None:

Parameters:

Filter Methods

filter()

Apply filter conditions to the query. Supports operator suffixes on field names and relationship traversal.

def filter(
    self,
    **conditions: FilterValue,
) -> Self:

Parameters:

Returns: Self for method chaining.

Raises:

• InvalidFilterError -- If a field does not exist on the model.
• InvalidRelationshipError -- If a relationship traversal path is invalid.
• TypeError -- If a list is passed for a scalar operator, or a non-string for a string operator.

Example:

# Simple equality (default __eq)
db.select(User).filter(name="Alice")

# Comparison operators
db.select(User).filter(age__gt=18, age__lt=65)

# Multiple chained calls (AND logic)
db.select(User).filter(active=True).filter(age__gte=21)

# Relationship traversal (ORM mode)
db.select(Book).filter(author__name="Alice")

Filter Operators

Field Selection

fields()

Specify which fields to include in the query results.

def fields(
    self,
    fields: list[str] | None = None,
) -> Self:

Parameters:

Returns: Self for method chaining.

Example:

db.select(User).fields(["name", "email"]).fetch_all()

exclude()

Specify which fields to exclude from the query results.

def exclude(
    self,
    fields: list[str] | None = None,
) -> Self:

Parameters:

Returns: Self for method chaining.

Raises:

• ValueError -- If pk is excluded, if invalid fields are specified, or if exclusion leaves no fields.

Example:

db.select(User).exclude(["password_hash"]).fetch_all()

only()

Select a single field (plus pk).

def only(
    self,
    field: str,
) -> Self:

Parameters:

Returns: Self for method chaining.

Raises:

• ValueError -- If the field does not exist.

Example:

db.select(User).only("email").fetch_all()

Relationships

select_related()

Specify FK relationships to eager load via SQL JOINs. Reduces the N+1 query problem by fetching related objects in a single query.

def select_related(
    self,
    *paths: str,
) -> Self:

Parameters:

Returns: Self for method chaining.

Raises:

• InvalidRelationshipError -- If a path contains invalid fields.

Example:

# Single level
books = db.select(Book).select_related("author").fetch_all()
# book.author.name  -- no additional query needed

# Nested
comments = db.select(Comment).select_related(
    "post__author"
).fetch_all()

# Multiple paths
books = db.select(Book).select_related(
    "author", "publisher"
).fetch_all()

prefetch_related()

Specify reverse FK and M2M relationships to eager load via a second query. Reduces the N+1 query problem for reverse relationships.

def prefetch_related(
    self,
    *paths: str,
) -> Self:

Parameters:

Returns: Self for method chaining.

Raises:

• InvalidPrefetchError -- If a path is not a valid reverse FK or M2M relationship. Forward FK paths (which should use select_related()) also raise this error.

Example:

# Reverse FK
authors = db.select(Author).prefetch_related("books").fetch_all()
# author.books.fetch_all()  -- no additional query

# M2M (forward)
articles = db.select(Article).prefetch_related("tags").fetch_all()

# M2M (reverse)
tags = db.select(Tag).prefetch_related("articles").fetch_all()

# Multiple paths
authors = db.select(Author).prefetch_related(
    "books", "reviews"
).fetch_all()

# Nested paths
authors = db.select(Author).prefetch_related(
    "books__reviews"
).fetch_all()

# Combined with select_related
books = db.select(Book).select_related("author").prefetch_related().fetch_all()

Pagination

limit()

Limit the number of results returned.

def limit(
    self,
    limit_value: int,
) -> Self:

Parameters:

Returns: Self for method chaining.

offset()

Skip a number of records before returning results.

def offset(
    self,
    offset_value: int,
) -> Self:

Parameters:

Returns: Self for method chaining.

Raises:

• InvalidOffsetError -- If the offset value is negative.

Example:

# Pagination: page 2, 10 items per page
db.select(User).limit(10).offset(10).fetch_all()

Ordering

order()

Order the query results by a field.

def order(
    self,
    order_by_field: str | None = None,
    direction: str | None = None,
    *,
    reverse: bool = False,
) -> Self:

Parameters:

Returns: Self for method chaining.

Raises:

• InvalidOrderError -- If the field does not exist, or if both direction and reverse are specified.

Warns:

• DeprecationWarning -- If direction is used.

Example:

# Ascending (default)
db.select(User).order("name").fetch_all()

# Descending
db.select(User).order("created_at", reverse=True).fetch_all()

Cache Control

bypass_cache()

Force this query to skip the cache and hit the database directly.

def bypass_cache(self) -> Self:

Returns: Self for method chaining.

Example:

fresh = db.select(User).bypass_cache().fetch_all()

cache_ttl()

Set a custom time-to-live for this query's cached result, overriding the global cache_ttl.

def cache_ttl(
    self,
    ttl: int,
) -> Self:

Parameters:

Returns: Self for method chaining.

Raises:

• ValueError -- If ttl is negative.

Example:

# Cache this query for 60 seconds
db.select(User).cache_ttl(60).fetch_all()

Execution Methods

fetch_all()

Execute the query and return all matching records.

def fetch_all(self) -> list[T]:

Returns: list[T] -- List of model instances.

Example:

users = db.select(User).filter(active=True).fetch_all()

fetch_one()

Execute the query and return a single record.

def fetch_one(self) -> T | None:

Returns: T | None -- A model instance, or None if no match.

fetch_first()

Fetch the first record (sets LIMIT 1).

def fetch_first(self) -> T | None:

Returns: T | None -- The first model instance, or None.

fetch_last()

Fetch the last record (orders by rowid DESC, sets LIMIT 1).

def fetch_last(self) -> T | None:

Returns: T | None -- The last model instance, or None.

count()

Count the number of matching records.

def count(self) -> int:

Returns: int -- The count of matching records.

Example:

total = db.select(User).filter(active=True).count()

exists()

Check if any matching records exist.

def exists(self) -> bool:

Returns: bool -- True if at least one record matches.

Example:

if db.select(User).filter(email="[email protected]").exists():
    print("User exists")

delete()

Delete all records matching the current query conditions.

def delete(self) -> int:

Returns: int -- The number of records deleted.

Raises:

• RecordDeletionError -- If there is an error deleting the records.

Example:

deleted = db.select(User).filter(active=False).delete()
print(f"Deleted {deleted} inactive users")

Supporting Types

JoinInfo

Dataclass holding metadata for a JOIN clause. Used internally by select_related() and relationship filter traversal.

@dataclass
class JoinInfo:
    alias: str
    table_name: str
    model_class: type[BaseDBModel]
    fk_field: str
    parent_alias: str
    fk_column: str
    join_type: str
    path: str
    is_nullable: bool

Fields: