Querying

All query methods are async. Always await them.

Getting started

Importing query tools

Everything you need for queries is re-exported from buraq.models, so a single import covers models, fields and query expressions:

from buraq import models

await Post.objects.filter(models.Q(published=True) | models.Q(pinned=True))
await Post.objects.update(views=models.F("views") + 1)
await Post.objects.aggregate(total=models.Count("id"))

This mirrors Django, where the same names live on django.db.models. The specific modules also remain importable if you prefer them:

from buraq.orm.query import F, Q
from buraq.orm.aggregates import Count
from buraq.orm.expressions import Case, When, Value
from buraq.orm.window import Rank, Window

Both forms return the same objects.

Basic operations

# All records
posts = await Post.objects.all()

# Filter
posts = await Post.objects.filter(is_published=True)

# Exclude
posts = await Post.objects.exclude(is_published=False)

# Get single record (raises DoesNotExist if not found)
post = await Post.objects.get(id=1)
post = await Post.objects.get(slug="hello-world")

# Get or None
post = await Post.objects.get_or_none(slug="hello-world")

# Count
n = await Post.objects.count()
n = await Post.objects.filter(is_published=True).count()

# Check existence
exists = await Post.objects.filter(slug="hello").exists()

# Create
post = await Post.objects.create(title="Hello", slug="hello", content="...")

# Update
await Post.objects.filter(id=1).update(is_published=True)

# Delete
await Post.objects.filter(is_published=False).delete()

# Save an instance
post = await Post.objects.get(id=1)
post.title = "Updated title"
await post.save()

# Delete an instance
await post.delete()

Filtering

Lookup expressions

Post.objects.filter(title__contains="Django")
Post.objects.filter(title__icontains="django")   # case-insensitive
Post.objects.filter(title__startswith="Hello")
Post.objects.filter(title__istartswith="hello")
Post.objects.filter(title__endswith="World")
Post.objects.filter(created_at__gt=some_date)    # greater than
Post.objects.filter(created_at__gte=some_date)   # greater than or equal
Post.objects.filter(views__lt=100)               # less than
Post.objects.filter(views__lte=100)              # less than or equal
Post.objects.filter(title__in=["A", "B", "C"])
Post.objects.filter(category_id__isnull=True)
Post.objects.filter(title__iexact="hello world")          # case-insensitive exact
Post.objects.filter(views__range=(100, 500))              # BETWEEN 100 AND 500
Post.objects.filter(created_at__year=2024)                # extract year
Post.objects.filter(created_at__month=6)                  # extract month
Post.objects.filter(created_at__day=15)                   # extract day

Full lookup reference

Lookup	SQL equivalent	Notes
exact	= value	Default when no lookup given
iexact	ILIKE value	Case-insensitive exact
contains	LIKE %value%
icontains	ILIKE %value%	Case-insensitive
startswith	LIKE value%
istartswith	ILIKE value%	Case-insensitive
endswith	LIKE %value
iendswith	ILIKE %value	Case-insensitive
gt	> value
gte	>= value
lt	< value
lte	<= value
in	IN (...)	Pass a list
isnull	IS NULL / IS NOT NULL	Pass True or False
range	BETWEEN v1 AND v2	Pass a 2-tuple
year	EXTRACT(year ...)	DateTimeField only
month	EXTRACT(month ...)	DateTimeField only
day	EXTRACT(day ...)	DateTimeField only
iso_year	EXTRACT(isoyear ...)	ISO 8601 year (differs from year around new year)
iso_week_day	EXTRACT(isodow ...)	ISO weekday: 1=Monday … 7=Sunday
contained_by	col <@ value	PostgreSQL JSON/array: column is subset of value
has_key	col ? key	PostgreSQL JSONB: top-level key exists
has_keys	col ?& keys	PostgreSQL JSONB: all listed keys exist
has_any_keys	col ?| keys	PostgreSQL JSONB: any listed key exists
overlap	col && value	PostgreSQL array/range: shares at least one element

Date lookup examples

# ISO year — useful for filtering around week boundaries (e.g. Dec 31 → ISO year +1)
Post.objects.filter(published__iso_year=2025)

# ISO weekday
Post.objects.filter(published__iso_week_day=1)   # Monday only

PostgreSQL JSON / array lookup examples

# JSONB: find rows where "metadata" JSONB column contains the key "color"
Product.objects.filter(metadata__has_key="color")

# JSONB: find rows where all of these keys exist
Product.objects.filter(metadata__has_keys=["color", "size"])

# JSONB: find rows where any of these keys exist
Product.objects.filter(metadata__has_any_keys=["color", "material"])

# JSONB / array: column value is contained in the given set
Product.objects.filter(tags__contained_by=["python", "django", "buraq"])

# Array: find rows whose tags array overlaps with the given list
Product.objects.filter(tags__overlap=["python", "async"])

Q objects — complex filters

from buraq.orm.query import Q

# OR
posts = await Post.objects.filter(
    Q(title__contains="Django") | Q(title__contains="FastAPI")
)

# AND
posts = await Post.objects.filter(
    Q(is_published=True) & Q(views__gt=100)
)

# NOT
posts = await Post.objects.filter(~Q(is_published=False))

# Nested
posts = await Post.objects.filter(
    Q(is_published=True) & (Q(title__contains="async") | Q(views__gt=500))
)

# XOR — exactly one condition must be true
posts = await Post.objects.filter(
    Q(is_featured=True) ^ Q(is_editor_pick=True)
)

XOR is emulated as (A OR B) AND NOT (A AND B) for full compatibility across SQLite, PostgreSQL, and MySQL.

F expressions — field references

from buraq.orm.query import F

# Increment a counter without a read-modify-write
await Post.objects.filter(id=1).update(views=F("views") + 1)

# Compare two fields
posts = await Post.objects.filter(updated_at__gt=F("created_at"))

none() — empty queryset

Return a queryset that always yields zero results — useful for conditional query building:

qs = Post.objects.none()
results = await qs.all()   # → []
count = await qs.count()   # → 0

distinct()

Remove duplicate rows from results:

# Unique category IDs
category_ids = await Post.objects.values_list("category_id", flat=True).distinct().all()

Ordering, slicing and paging

Ordering

# Ascending
posts = await Post.objects.all().order_by("created_at")

# Descending
posts = await Post.objects.all().order_by("-created_at")

# Multiple fields
posts = await Post.objects.all().order_by("-is_published", "title")

Limiting

posts = await Post.objects.all().limit(10)
posts = await Post.objects.all().limit(10).offset(20)

Pagination

from buraq.paginator import Paginator

paginator = Paginator(Post.objects.filter(is_published=True), per_page=10)
page      = await paginator.page(request.query_params.get("page", 1))

# page.object_list — items on this page
# page.has_next() / page.has_previous()
# page.next_page_number() / page.previous_page_number()
# paginator.num_pages

Earliest and latest

# First record by created_at
oldest = await Post.objects.earliest("created_at")

# Most recent record by created_at
newest = await Post.objects.latest("created_at")

# Defaults to primary key if no field is specified
first  = await Post.objects.earliest()
last   = await Post.objects.latest()

last()

Return the last object by primary key, or None:

latest_post = await Post.objects.last()
latest_published = await Post.objects.filter(is_published=True).last()

Choosing what comes back

values() and values_list()

Return dicts or tuples instead of model instances — useful for serialization and aggregation:

# List of dicts
posts = await Post.objects.values("id", "title", "views").all()
# → [{"id": 1, "title": "Hello", "views": 42}, ...]

# List of tuples
posts = await Post.objects.values_list("id", "title").all()
# → [(1, "Hello"), (2, "World"), ...]

# Single-column flat list
ids = await Post.objects.values_list("id", flat=True).all()
# → [1, 2, 3, ...]

# Combine with filters and ordering
slugs = await (
    Post.objects
    .filter(is_published=True)
    .order_by("-created_at")
    .values_list("slug", flat=True)
    .all()
)

annotate_expr()

Add arbitrary SQL expression columns to each result row. Accepts aggregates, window functions, ORM expressions, or raw SQLAlchemy constructs:

from buraq.orm.aggregates import Count
from buraq.orm.window import Window, Rank
from buraq.orm.expressions import Case, When, Value

posts = await Post.objects.annotate_expr(
    rank=Window(Rank(), partition_by="category_id", order_by="-views"),
    label=Case(
        When(is_featured=True, then=Value("featured")),
        default=Value("regular"),
    ),
).all()

Combined with values():

rows = await Post.objects.values("author_id").annotate_expr(
    post_count=Count("id")
).all()
# → [{"author_id": 1, "post_count": 5}, ...]

Annotating with arbitrary expressions

from buraq.orm.window import RowNumber, Window

posts = await Post.objects.annotate_expr(
    row_num=RowNumber(Window(order_by="id")),
).all()

alias()

Create a named subquery alias so the same queryset can be reused in multiple filter() or annotate_expr() calls without repeating SQL:

# Build once
recent_posts = Post.objects.filter(created_at__gte=cutoff).alias("recent")

# Reuse in outer queries
popular = await Post.objects.filter(id__in=recent_posts, views__gte=100).all()
long_read = await Post.objects.filter(id__in=recent_posts, read_time__gte=10).all()

Deferred loading

Load only specific columns; remaining columns are fetched lazily when accessed.

# Load only title and slug — content and other fields are deferred
posts = await Post.objects.only("title", "slug").all()

# Load everything except the large content column
posts = await Post.objects.defer("content").all()

Fetch modes for deferred fields

When a queryset uses defer() or only(), accessing a deferred field normally triggers an extra per-instance query. fetch_mode() lets you control that behaviour explicitly.

from buraq.orm.manager import FETCH_ONE, FETCH_PEERS, FETCH_RAISE

# Default — reload each instance individually on deferred-field access
posts = await Post.objects.only("title").fetch_mode(FETCH_ONE).all()

# Reload all peers in a single batch the first time any deferred field is accessed
posts = await Post.objects.only("title").fetch_mode(FETCH_PEERS).all()

# Raise FieldFetchBlocked immediately on any deferred-field access
posts = await Post.objects.only("title").fetch_mode(FETCH_RAISE).all()

totally_ordered

QuerySet.totally_ordered returns True when the queryset’s ORDER BY includes the primary key (or another unique column). A totally ordered queryset is safe for cursor-based pagination because the order is deterministic.

qs = Post.objects.order_by("-created_at", "id")
qs.totally_ordered  # → True (id is the PK)

qs2 = Post.objects.order_by("-created_at")
qs2.totally_ordered  # → False (created_at is not unique)

in_bulk() with values/values_list

in_bulk() now correctly handles querysets narrowed by values() or values_list():

# Returns {pk: dict}
mapping = await Post.objects.values("id", "title").in_bulk([1, 2, 3])

# Returns {pk: tuple}
mapping = await Post.objects.values_list("id", "title").in_bulk([1, 2, 3])

refresh_from_db()

Reload an instance’s fields from the database — useful after an out-of-band update (e.g. a bulk_update that bypassed the object):

post = await Post.objects.get(id=1)
# … some other code updates the row in the DB …
await post.refresh_from_db()           # reload all fields

# Reload only specific fields (avoids fetching heavy columns)
await post.refresh_from_db(fields=["status", "views"])

Related objects

select_related() / prefetch_related()

Eagerly load related objects to avoid N+1 queries:

# JOIN load (one query) — use for ForeignKey / OneToOneField
posts = await Post.objects.select_related("author").all()

# Subquery load (two queries) — use for ManyToManyField / reverse FK
posts = await Post.objects.prefetch_related("tags").all()

# Chain both
posts = await Post.objects.select_related("author").prefetch_related("tags").all()

For custom filtering on prefetched relations, use a Prefetch object (see Prefetch objects below).

Prefetch objects

Prefetch gives you fine-grained control over the queryset used when calling prefetch_related(). Import it from buraq.models (or buraq.orm.prefetch):

from buraq.models import Prefetch

# Load only approved comments, ordered by date
posts = await Post.objects.prefetch_related(
    Prefetch(
        "comments",
        queryset=Comment.objects.filter(approved=True).order_by("-created_at"),
    )
).all()

# Access the pre-fetched set on each instance
for post in posts:
    approved = post._prefetched_comments   # list[Comment]

Store the result under a custom attribute with to_attr:

posts = await Post.objects.prefetch_related(
    Prefetch("comments", queryset=Comment.objects.filter(approved=True), to_attr="approved_comments"),
    Prefetch("comments", queryset=Comment.objects.filter(approved=False), to_attr="pending_comments"),
).all()

Creating and updating

get_or_create()

Fetch an object matching kwargs, or create it if it doesn’t exist. Returns (instance, created):

post, created = await Post.objects.get_or_create(
    slug="hello-world",
    defaults={"title": "Hello World", "content": "..."},
)
# created=True  → new object was inserted
# created=False → existing object was returned

defaults are only used when creating — they are not used in the lookup.

Race safety

get_or_create uses a try-create-catch-IntegrityError pattern internally, so it is safe under concurrent requests: if two coroutines race to create the same row, the loser catches the database’s IntegrityError and falls back to fetching the row the winner created — no DoesNotExist is leaked.

post, created = await Post.objects.get_or_create(
    slug="hello-world",
    defaults={"title": "Hello World", "content": "..."},
)

update_or_create is race-safe by the same mechanism.

update_or_create()

Like get_or_create() but updates the existing object with defaults if found:

post, created = await Post.objects.update_or_create(
    slug="hello-world",
    defaults={"title": "Updated Title", "views": 0},
)

Bulk operations

# Bulk create
await Post.objects.bulk_create([
    {"title": "Post 1", "slug": "post-1", "content": "..."},
    {"title": "Post 2", "slug": "post-2", "content": "..."},
])

# With ignore_conflicts (skip duplicates)
await Post.objects.bulk_create(records, ignore_conflicts=True)

# Bulk update — update specific fields on a list of instances
posts = await Post.objects.filter(is_published=False).all()
for post in posts:
    post.status = "archived"
await Post.objects.bulk_update(posts, fields=["status"])

What ignore_conflicts becomes

Every database can skip a row that is already there and no two agree how, so insert_ignoring_duplicates() writes it in the dialect’s own spelling: ON CONFLICT DO NOTHING on PostgreSQL and SQLite, ON DUPLICATE KEY UPDATE on MySQL and MariaDB, which have never had the former.

The same applies to adding to a many-to-many relation, which inserts into the association table the same way.

MySQL’s form updates a key column to the value it already has, which changes nothing. INSERT IGNORE would also skip duplicates, and would swallow a truncated value or a bad foreign key along with them — too much to discard for the convenience.

bulk_update — single round-trip

bulk_update sends a single parameterised UPDATE statement (via sa.bindparam bulk binding) regardless of how many instances are passed — no N-query loop:

posts = await Post.objects.filter(is_published=False).all()
for post in posts:
    post.status = "archived"

# One SQL statement, no matter how many posts
await Post.objects.bulk_update(posts, fields=["status"])

in_bulk

# Fetch a dict keyed by primary key
post_map = await Post.objects.in_bulk([1, 2, 3])
# → {1: <Post id=1>, 2: <Post id=2>, 3: <Post id=3>}

# Keyed by a different field
slug_map = await Post.objects.in_bulk(["hello", "world"], field_name="slug")

Large result sets

Streaming large querysets

async for post in Post.objects.filter(is_published=True).iterator():
    process(post)   # memory-efficient — doesn't load all at once

explain()

Retrieve the database’s query plan for debugging slow queries:

# Basic EXPLAIN
plan = await Post.objects.filter(is_published=True).explain()
print(plan)

# With ANALYZE (actually executes the query — PostgreSQL / SQLite)
plan = await Post.objects.filter(is_published=True).explain(analyze=True)

# With VERBOSE (PostgreSQL)
plan = await Post.objects.filter(is_published=True).explain(analyze=True, verbose=True)

The returned value is a string containing the database’s plan output.

Raw SQL and advanced queries

Raw SQL

Use when ORM expressions can’t express what you need.

rows = await Post.objects.raw(
    "SELECT id, title FROM posts WHERE views > :min_views",
    {"min_views": 100},
)
# → [{"id": 1, "title": "..."}, ...]

extra() — raw SQL fragments

extra() is a low-level escape hatch for SQL that can’t be expressed with the ORM.

posts = await Post.objects.extra(
    select={"word_count": "length(content)"},
    where=["LENGTH(content) > %s"],
    params=[500],
).all()

Parameter	Purpose
select	Dict of {alias: sql_expression} added to the SELECT list
where	List of raw WHERE clause fragments joined with AND
params	Positional values for %s placeholders in where
tables	Additional table names appended to FROM

Caution

extra() bypasses ORM safety guarantees. Prefer ORM expressions and raw() where possible.

Set operations — union / intersection / difference

Combine multiple querysets using SQL set operations. All querysets must select the same columns.

published = Post.objects.filter(is_published=True)
featured  = Post.objects.filter(is_featured=True)

# UNION — all published or featured posts (duplicates removed)
result = await published.union(featured).all()

# UNION ALL — keep duplicates
result = await published.union(featured, all=True).all()

# INTERSECT — posts that are both published AND featured
result = await published.intersection(featured).all()

# EXCEPT — published but not featured
result = await published.difference(featured).all()

Date and datetime truncation

# Distinct years that have at least one post
years = await Post.objects.dates("created_at", "year")
# → [datetime.date(2023, 1, 1), datetime.date(2024, 1, 1), ...]

# Distinct months
months = await Post.objects.dates("created_at", "month")

# Datetime precision (requires DateTimeField)
hours = await Post.objects.datetimes("created_at", "hour")
# kind: "year" | "month" | "day" | "hour" | "minute" | "second"

Locking rows — select_for_update

# Lock rows for the duration of the current transaction
posts = await Post.objects.filter(is_published=False).select_for_update().all()

# Non-blocking — skip rows that are already locked
posts = await Post.objects.filter(status="pending").select_for_update(skip_locked=True).all()

# Raise immediately if any row is locked
posts = await Post.objects.filter(status="pending").select_for_update(nowait=True).all()

Reading from a replica

Configure the databases and say which of them reads may go to:

config/settings.py

DATABASES = {
    "default": "postgresql+asyncpg://user:pass@primary/db",
    "replica": "postgresql+asyncpg://user:pass@replica/db",
}
DATABASE_READ_REPLICAS = ["replica"]

Nothing in your queries changes. Reads are sent to the replica, writes to the primary:

await Post.objects.filter(published=True)      # replica
await Post.objects.create(title="Hello")       # primary

List more than one replica and reads rotate between them.

What stays on the primary

	goes to
any write	default
any read inside atomic()	default
raw()	default — the SQL may write
everything else	a replica, if one is configured

The second row is the one that matters. A replica is behind the primary by however long replication takes, so a transaction that writes a row and then reads it back would see stale data — or nothing at all:

@atomic
async def publish(post_id):
    await Post.objects.filter(id=post_id).update(published=True)
    # Reads the row it just wrote, so this must not go to a replica.
    return await Post.objects.get(id=post_id)

Buraq routes every read inside an atomic() block to the primary for that reason. You do not have to think about it.

using()

using() names a database outright, and overrides all of the above:

await Post.objects.using("replica").filter(published=True)   # even inside atomic()
await Post.objects.using("default").get(id=1)                # a read you need current

It chains in either position, and survives the calls after it:

Post.objects.using("replica").filter(x=1).order_by("-id")
Post.objects.filter(x=1).using("replica").order_by("-id")    # the same

Reading straight after writing

Outside a transaction, nothing links a write to the read that follows it, so this can return the old row:

await post.save()
await Post.objects.get(id=post.id)     # may be answered by a lagging replica

Wrap the pair in atomic(), or name the database with using("default").

No routers

There is nothing that decides per model or per app which database a query belongs to. Reads go to a replica, writes to the primary, using() overrides both — and migrations only ever run against default.