Build custom lookups, transforms, Func and Aggregate classes, correlated subqueries, QuerySet APIs, check and exclusion constraints, and generated fields on Django 5.2, then test query counts and SQL.
The Django ORM is a compiler with a public extension surface. A custom lookup, a Func subclass, a correlated subquery, a database constraint, and a generated column all end up as SQL fragments the planner has to execute. This guide builds each one against a small order-management schema on Django 5.2 LTS and Python 3.12+, targeting PostgreSQL with SQLite fallbacks where they are cheap, and shows where each technique breaks.
Every example uses three models. Later sections add managers, constraints, and generated columns to them.
# myapp/models.py
from django.db import models
class Customer(models.Model):
class Flag(models.IntegerChoices):
BETA = 1
INVOICING = 2
PRIORITY = 4
email = models.EmailField()
flags = models.PositiveIntegerField(default=0)
class Order(models.Model):
class Status(models.TextChoices):
DRAFT = "draft"
OPEN = "open"
PAID = "paid"
SHIPPED = "shipped"
CANCELLED = "cancelled"
customer = models.ForeignKey(
Customer, on_delete=models.PROTECT, related_name="orders"
)
status = models.CharField(max_length=10, choices=Status, default=Status.DRAFT)
reference = models.CharField(max_length=30, blank=True)
placed_at = models.DateTimeField(null=True, blank=True)
shipped_at = models.DateTimeField(null=True, blank=True)
class OrderLine(models.Model):
order = models.ForeignKey(Order, on_delete=models.CASCADE, related_name="lines")
sku = models.CharField(max_length=40)
quantity = models.PositiveIntegerField()
unit_cents = models.PositiveBigIntegerField()Three ideas explain nearly every extension point.
F, Value, Func, Subquery, Q, lookups, and transforms all resolve against a Query and then compile to a (sql, params) pair.compiler.compile(node) looks for as_postgresql, as_sqlite, as_mysql, or as_oracle on the node before falling back to as_sql. Vendor differences belong in those methods, not in if connection.vendor branches around your queries.params, and the driver binds them. Your SQL fragment must contain exactly as many placeholders as there are parameters, in order. Most bugs in custom expressions are placeholder bugs.A lookup is the final part of a filter, such as __gte, and produces a boolean condition. A transform is a middle part, such as __year, and produces a value that further lookups can act on.
The flags column packs several booleans into an integer. Testing whether all requested bits are set should read as a filter, not as a raw extra() call.
# myapp/lookups.py
from django.db import models
from django.db.models import Lookup, Transform
@models.IntegerField.register_lookup
class HasAllBits(Lookup):
lookup_name = "hasbits"
def as_sql(self, compiler, connection):
lhs, lhs_params = self.process_lhs(compiler, connection)
rhs, rhs_params = self.process_rhs(compiler, connection)
# rhs appears twice in the SQL, so its params must appear twice too.
params = (*lhs_params, *rhs_params, *rhs_params)
return f"({lhs} & {rhs}) = {rhs}", paramsprocess_lhs compiles the column or expression on the left. The right-hand value already went through the field's get_prep_value when the lookup was built, so a non-integer raises before it reaches the database. Using the lookup is now ordinary filter syntax:
wanted = Customer.Flag.INVOICING | Customer.Flag.PRIORITY
Customer.objects.filter(flags__hasbits=wanted)
# WHERE ("myapp_customer"."flags" & %s) = %s params: (6, 6)The trade-off is indexing. No ordinary B-tree index helps this predicate, so it suits a secondary filter on an already narrow result. If one flag dominates your queries, a boolean column with a partial index is the better schema.
Filtering customers by email domain is a value derivation, so it is a transform. Transform subclasses Func, which means template and vendor methods work the same way.
@models.EmailField.register_lookup
class EmailDomain(Transform):
lookup_name = "domain"
output_field = models.CharField()
template = "LOWER(SPLIT_PART(%(expressions)s, '@', 2))"
def as_sqlite(self, compiler, connection, **extra_context):
lhs, params = compiler.compile(self.lhs)
sql = f"LOWER(SUBSTR({lhs}, INSTR({lhs}, '@') + 1))"
return sql, (*params, *params)Because output_field is a CharField, every text lookup chains after the transform, and it works across relations:
Customer.objects.filter(email__domain="example.com")
Customer.objects.filter(email__domain__startswith="mail.")
Order.objects.filter(customer__email__domain="example.com")The SQLite branch shows the failure mode from the previous section. {lhs} appears twice, so params must be doubled. For a plain column params is empty and a mistake stays hidden. It surfaces months later, when someone applies the transform to a parameterized expression and gets a binding-count error.
To make the transform efficient, index the same expression. PostgreSQL only uses an expression index when the query expression matches the indexed one, and reusing the class guarantees that:
class Customer(models.Model):
...
class Meta:
indexes = [
models.Index(EmailDomain("email"), name="customer_email_domain_idx"),
]Set bilateral = True on a transform when it should also apply to the right-hand side, as a case-folding transform would.
Register on the narrowest field class that makes sense. EmailField keeps __domain away from unrelated text columns, and subclasses inherit the registration, so PositiveIntegerField gets hasbits through IntegerField.
The module has to be imported before queries run. Importing it from models.py happens naturally if a model references the class. Otherwise, do it explicitly:
# myapp/apps.py
from django.apps import AppConfig
class MyappConfig(AppConfig):
name = "myapp"
def ready(self):
from . import lookups # noqa: F401A missing import produces a FieldError beginning Unsupported lookup 'domain' for EmailField. Note also that migrations record the import path of any class used in an index, constraint, or generated field. Moving EmailDomain to another module later breaks old migrations unless the old path keeps working, so leave a re-export in place.
Any scalar SQL function can be expressed inline with Func(F("x"), function="..."), but a named class gives you a declared output type, arity checking, and a place for vendor logic.
This expression returns the elapsed seconds between two timestamps. It uses arg_joiner so that the compiled arguments become a subtraction.
# myapp/expressions.py
from django.db import models
from django.db.models import Aggregate, Func
class SecondsBetween(Func):
"""SecondsBetween(later, earlier) as a float."""
arity = 2
arg_joiner = " - "
template = "EXTRACT(EPOCH FROM (%(expressions)s))"
output_field = models.FloatField()
def as_sqlite(self, compiler, connection, **extra_context):
return self.as_sql(
compiler,
connection,
template="(JULIANDAY(%(expressions)s)) * 86400.0",
arg_joiner=") - JULIANDAY(",
**extra_context,
)from django.db.models import Avg
Order.objects.filter(shipped_at__isnull=False).aggregate(
avg_fulfilment=Avg(SecondsBetween("shipped_at", "placed_at"))
)Declaring output_field matters in two ways. It avoids the FieldError Django raises when it cannot infer a type from mixed operands, and it determines the Python conversion applied to the result. The driver may return a Decimal for EXTRACT on PostgreSQL, and the FloatField converts it. If a timedelta is acceptable, F("shipped_at") - F("placed_at") already resolves to a DurationField with no custom class.
Ordered-set aggregates do not fit the default template because the column sits outside the parentheses, which makes them a good example of overriding template.
class PercentileCont(Aggregate):
function = "PERCENTILE_CONT"
template = "%(function)s(%(fraction)s) WITHIN GROUP (ORDER BY %(expressions)s)"
output_field = models.FloatField()
def __init__(self, expression, fraction, **extra):
fraction = float(fraction)
if not 0.0 <= fraction <= 1.0:
raise ValueError("fraction must be between 0 and 1")
super().__init__(expression, fraction=fraction, **extra)from django.db.models import Q
OrderLine.objects.aggregate(
p95_quantity=PercentileCont(
"quantity", 0.95, filter=Q(sku__startswith="A"), default=0
)
)
# SELECT COALESCE(PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY ...)
# FILTER (WHERE ...), 0.0) AS "p95_quantity"filter and default come from Aggregate and work with no extra code. This class is PostgreSQL-specific and fails on SQLite with a syntax error, so tests that exercise it need a PostgreSQL run.
The aggregate template API changed in Django 6.0, which added order_by support to aggregates. Retest custom aggregates when you upgrade.
Keyword arguments passed through **extra are interpolated directly into the template. They are not bound by the driver. That is why the constructor above forces fraction through float() and a range check. Never pass request data into template, function, or extra values. Anything user-controlled belongs in a Value() expression or a normal argument, so that it reaches params.
A literal % in a Func template must be quadrupled to %%%%, because the string is interpolated twice: once in as_sql() and once by the database cursor.
Subqueries solve a problem that joins cannot: aggregating more than one related table without row multiplication. This query looks correct and is not:
from django.db.models import Count
customer = Customer.objects.annotate(
order_count=Count("orders"),
line_count=Count("orders__lines"),
).get(pk=customer_id)
# One order with two lines gives order_count == 2.Both aggregates run over the same joined rows, so each order is counted once per line. Count(..., distinct=True) repairs counts, but nothing repairs a Sum over a fanned-out join. Correlated subqueries keep each aggregate independent.
from django.db.models import Count, Exists, OuterRef, Subquery
from django.db.models.functions import Coalesce
class CustomerQuerySet(models.QuerySet):
def with_order_stats(self):
placed = Order.objects.placed().filter(customer=OuterRef("pk"))
latest = placed.order_by("-placed_at", "-pk")
counts = (
placed.order_by() # drop any inherited ordering
.values("customer") # group per outer row
.annotate(n=Count("pk"))
.values("n") # exactly one output column
)
return self.annotate(
last_order_at=Subquery(latest.values("placed_at")[:1]),
order_count=Coalesce(Subquery(counts), 0),
has_open_order=Exists(
Order.objects.filter(customer=OuterRef("pk"), status="open")
),
)
class Customer(models.Model):
...
objects = CustomerQuerySet.as_manager()Each part of that shape has a reason. The first values() sets the grouping, the second reduces the output to one column, and order_by() clears inherited ordering, because an explicitly ordered column joins the GROUP BY. Coalesce is needed because a customer with no rows gets NULL from the subquery instead of zero.
[:1] on a non-aggregated subquery makes PostgreSQL raise "more than one row returned by a subquery used as an expression", but only once the data actually produces two rows. Tests with one row per parent will not catch it.OuterRef raises ValueError: This queryset contains a reference to an outer query and may only be used in a subquery. when evaluated alone.(customer_id, placed_at DESC), and read the plan. Three correlated subqueries against one table may be slower than one GROUP BY materialized in the application, or than a small denormalized summary table.Exists can stop at the first match, and it works directly in filter().Put query logic on a QuerySet subclass, which composes, rather than on a Manager, which does not.
from django.db.models import Sum
class OrderQuerySet(models.QuerySet):
def placed(self):
return self.exclude(status=Order.Status.DRAFT)
def overdue(self, cutoff):
return self.filter(
status=Order.Status.PAID,
shipped_at__isnull=True,
placed_at__lt=cutoff,
)
def with_total(self):
totals = (
OrderLine.objects.filter(order=OuterRef("pk"))
.order_by()
.values("order")
.annotate(total=Sum("line_total_cents"))
.values("total")
)
return self.annotate(
total_cents=Coalesce(
Subquery(totals), 0, output_field=models.BigIntegerField()
)
)
class Order(models.Model):
...
objects = OrderQuerySet.as_manager()These methods chain in any order, and they work from related managers: customer.orders.placed().with_total(). A few conventions keep such a class maintainable.
with_* methods add columns. Callers can then see which calls change the row set and which change the SELECT.overdue(cutoff) is testable without freezing the clock and stays deterministic within a request.get_queryset() taxes every row fetch, including callers who never read it. Use alias() when a computed value is only needed for filtering, because it is not selected.Prefetch("orders", queryset=Order.objects.placed().with_total(), to_attr="placed_orders") reuses the logic instead of duplicating it.When you also need manager-only methods, combine the two with from_queryset:
class OrderManager(models.Manager.from_queryset(OrderQuerySet)):
def create_draft(self, customer):
return self.create(customer=customer, status=Order.Status.DRAFT)Do not filter rows in the first manager on a model. A soft-delete manager that hides rows as objects also hides them from the admin and from dumpdata, which is the wrong default for a recovery path. Keep the unfiltered manager first and add a named one, such as live, for the narrowed view. Forward relation access such as order.customer uses _base_manager, which ignores your default manager, so a filtering manager does not protect that path. Scoping that must hold everywhere, such as tenant isolation, needs more than a manager.
Application validation can be bypassed by bulk_create, update(), a management shell, or a second service. A database constraint cannot. Treat constraints as the definition of an invariant, and treat validation as the friendly error message on top.
from django.db.models import F, Q
from django.db.models.functions import Lower
class Order(models.Model):
...
class Meta:
constraints = [
models.CheckConstraint(
condition=Q(shipped_at__isnull=True)
| Q(shipped_at__gte=F("placed_at")),
name="order_shipped_after_placed",
violation_error_code="shipped_before_placed",
violation_error_message="An order cannot ship before it is placed.",
),
models.CheckConstraint(
condition=Q(status="draft") | Q(placed_at__isnull=False),
name="order_placed_has_timestamp",
),
# At most one draft per customer.
models.UniqueConstraint(
fields=["customer"],
condition=Q(status="draft"),
name="order_one_draft_per_customer",
),
models.UniqueConstraint(
fields=["customer", "reference"],
condition=~Q(reference=""),
name="order_reference_unique_when_set",
),
]
class Customer(models.Model):
...
class Meta:
constraints = [
models.UniqueConstraint(Lower("email"), name="customer_email_ci_uniq"),
]On Django 5.2 use condition=. The older check= keyword was deprecated in 5.1 and removed in 6.0. Running tests with python -Wa shows the warning before an upgrade turns it into an error.
Conditional and expression-based unique constraints are partial and functional unique indexes underneath. UniqueConstraint also accepts nulls_distinct=False on PostgreSQL 15 and later, which treats NULL values as equal, and deferrable= for constraints checked at commit. A deferrable unique constraint cannot be combined with condition or expressions.
A common blind spot is three-valued logic. A CHECK passes when its condition evaluates to NULL, not only when it is true. Nothing stops a draft with shipped_at set and placed_at empty, since the comparison yields NULL. When a column should be required under a condition, state it with an explicit __isnull test, as the second constraint does.
Uniqueness compares with equality. An exclusion constraint generalizes that to any commutative operator, which is the only reliable way to prevent overlapping bookings. A check-then-insert in application code races under concurrency, unless every writer takes a lock.
from django.contrib.postgres.constraints import ExclusionConstraint
from django.contrib.postgres.fields import (
DateTimeRangeField,
RangeBoundary,
RangeOperators,
)
from django.db import models
from django.db.models import Func, Q
class TsTzRange(Func):
function = "TSTZRANGE"
output_field = DateTimeRangeField()
class Reservation(models.Model):
room = models.ForeignKey("Room", on_delete=models.CASCADE)
starts_at = models.DateTimeField()
ends_at = models.DateTimeField()
cancelled = models.BooleanField(default=False)
class Meta:
constraints = [
ExclusionConstraint(
name="reservation_no_overlap",
expressions=[
(
TsTzRange("starts_at", "ends_at", RangeBoundary()),
RangeOperators.OVERLAPS,
),
("room", RangeOperators.EQUAL),
],
condition=Q(cancelled=False),
),
]The equality comparison on room inside a GiST index requires the btree_gist extension. Add it in a migration that runs before the constraint, and add django.contrib.postgres to INSTALLED_APPS:
from django.contrib.postgres.operations import BtreeGistExtension
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [("myapp", "0003_reservation")]
operations = [BtreeGistExtension()]RangeBoundary() defaults to [), so one booking can end exactly when the next begins. Creating an extension needs elevated database privileges on many systems, so check that your migration role has them, or have an administrator create the extension first.
Model.full_clean() calls validate_constraints(), and ModelForm and the admin call it too. Plain save() does not. One subtle rule: a constraint is skipped when a field it references is excluded. A ModelForm that omits shipped_at does not validate order_shipped_after_placed at all, so the violation arrives as an IntegrityError at save time.
Since validation is a separate query from the write, concurrent requests can both pass it. Your write path must handle the database error anyway. On PostgreSQL with psycopg, the constraint name is available from the driver's exception:
from django.db import IntegrityError, transaction
def book(room, starts_at, ends_at):
try:
with transaction.atomic():
return Reservation.objects.create(
room=room, starts_at=starts_at, ends_at=ends_at
)
except IntegrityError as exc:
diag = getattr(exc.__cause__, "diag", None)
if diag and diag.constraint_name == "reservation_no_overlap":
raise RoomUnavailable(room) from exc
raiseThe atomic() block sits inside the try because a failed statement aborts the surrounding transaction on PostgreSQL, and the savepoint rollback is what makes the connection usable again.
Adding a CHECK scans the table under a lock that blocks reads and writes. On a large PostgreSQL table, split it with the operations in django.contrib.postgres.operations. AddConstraintNotValid enforces the rule for new writes without scanning old rows, and ValidateConstraint in a later migration checks old rows under a weaker lock. NOT VALID does not apply to unique or exclusion constraints, so those build their index in the same step and need a maintenance plan on large tables. Clean up violating rows first in every case.
GeneratedField stores a database-computed column. The database owns the value, so it cannot drift from its inputs.
class OrderLine(models.Model):
...
line_total_cents = models.GeneratedField(
expression=F("quantity") * F("unit_cents"),
output_field=models.BigIntegerField(),
db_persist=True,
)
sku_key = models.GeneratedField(
expression=Lower("sku"),
output_field=models.CharField(max_length=40),
db_persist=True,
)This is the field with_total() summed earlier. The important constraints:
db_persist=True stores the value on disk. Django 5.2's PostgreSQL backend supports only stored columns, while SQLite and MySQL also allow virtual ones.timestamptz to a date depends on the session time zone, so it fails.RETURNING, but it stays stale after an update: change quantity, call save(), and the attribute still holds the old total until refresh_from_db(). Later releases change this, so recheck on upgrade.AttributeError: Cannot read a generated field from an unsaved model. Validation code cannot depend on it before the first save.QuerySet.update(line_total_cents=1) does not raise on 5.2. The field is omitted, and the update changes nothing.Modifying GeneratedFields is not supported. You remove the field and add it again, which rewrites a stored column for every row and drops indexes built on it.| Need | Use | Main cost |
|---|---|---|
| Reusable filter condition | Custom Lookup | Global registration, hidden from casual readers |
| Derived value used in filters | Transform with expression index | Index and query expressions must match |
| Derived value read often, same row | GeneratedField | Disk space, table rewrite when redefined |
| Value depends on other rows | Subquery annotation | Runs on every query, needs an index |
| Named business query | QuerySet method | Very little, the default choice |
| Rule that must always hold | Check or unique constraint | Migration locking, error handling on write |
| No overlapping ranges | ExclusionConstraint | PostgreSQL only, GiST write overhead |
ORM extensions fail in ways that normal behavior tests miss: extra queries, wrong placeholders, or a missing index. Test the behavior, the query count, and the SQL shape separately, and run the suite on the production database engine. Passing on SQLite says little about PostgreSQL.
# myapp/tests/test_orm.py
from datetime import timedelta
from unittest import skipUnless
from django.db import IntegrityError, connection, transaction
from django.test import TestCase
from django.test.utils import CaptureQueriesContext
from django.utils import timezone
from myapp.models import Customer, Order, OrderLine
class CustomLookupTests(TestCase):
def test_hasbits_requires_every_bit(self):
match = Customer.objects.create(email="a@example.com", flags=6)
Customer.objects.create(email="b@example.com", flags=2)
self.assertQuerySetEqual(
Customer.objects.filter(flags__hasbits=6), [match]
)
def test_hasbits_binds_the_value_twice(self):
qs = Customer.objects.filter(flags__hasbits=6)
sql, params = qs.query.sql_with_params()
self.assertEqual(params, (6, 6))
self.assertIn("&", sql)
class QueryCountTests(TestCase):
def test_order_stats_are_one_query_at_any_size(self):
Customer.objects.bulk_create(
[Customer(email=f"u{i}@example.com") for i in range(20)]
)
with self.assertNumQueries(1):
rows = list(Customer.objects.with_order_stats())
self.assertEqual(rows[0].order_count, 0)
def test_subquery_runs_inline(self):
with CaptureQueriesContext(connection) as ctx:
list(Order.objects.with_total())
self.assertEqual(len(ctx), 1)
self.assertIn("COALESCE", ctx.captured_queries[0]["sql"])
class ConstraintTests(TestCase):
def setUp(self):
self.customer = Customer.objects.create(email="a@example.com")
def test_update_cannot_bypass_check(self):
now = timezone.now()
order = Order.objects.create(
customer=self.customer, status="paid", placed_at=now
)
with self.assertRaises(IntegrityError), transaction.atomic():
Order.objects.filter(pk=order.pk).update(
shipped_at=now - timedelta(days=1)
)
def test_second_draft_is_rejected(self):
Order.objects.create(customer=self.customer)
with self.assertRaises(IntegrityError), transaction.atomic():
Order.objects.create(customer=self.customer)
class GeneratedFieldTests(TestCase):
def test_value_requires_refresh_after_update(self):
customer = Customer.objects.create(email="a@example.com")
order = Order.objects.create(customer=customer)
line = OrderLine.objects.create(
order=order, sku="AB", quantity=2, unit_cents=500
)
line.quantity = 3
line.save()
line.refresh_from_db()
self.assertEqual(line.line_total_cents, 1500)
class IndexUsageTests(TestCase):
@skipUnless(connection.vendor == "postgresql", "PostgreSQL planner test")
def test_domain_filter_can_use_expression_index(self):
qs = Customer.objects.filter(email__domain="example.com")
with connection.cursor() as cursor:
cursor.execute("SET LOCAL enable_seqscan = off")
self.assertIn("customer_email_domain_idx", qs.explain())transaction.atomic() inside assertRaises is required, because TestCase wraps each test in a transaction that the failed statement would otherwise break. SET LOCAL enable_seqscan = off is needed because the planner correctly prefers a sequential scan on a small test table. That proves the index is usable, not that production will choose it.
When inspecting SQL, remember that str(queryset.query) shows values without proper quoting and is not always valid SQL. Use it for reading, and use sql_with_params() or CaptureQueriesContext for assertions. Keep SQL assertions narrow. Matching a full query string breaks on harmless changes in Django upgrades.
Run these from the project root against a database that matches production:
# Run tests with all warnings visible and the SQL printed on failure
python -Wa manage.py test myapp --debug-sql
# Fail if models and migrations have drifted
python manage.py makemigrations --check --dry-run
# Run system checks, including database-tagged checks
python manage.py check --database default
# Review the DDL and the migration order before applying
python manage.py sqlmigrate myapp 0004
python manage.py migrate --planThen inspect the result in psql:
-- Constraints and indexes as the database sees them
\d myapp_order
\d myapp_customer
-- Confirm a slow annotation uses the intended index
EXPLAIN (ANALYZE, BUFFERS)
SELECT id FROM myapp_customer
WHERE LOWER(SPLIT_PART(email, '@', 2)) = 'example.com';Before merging, confirm each item:
params, including repeated fragments.template, function, or **extra.[:1] with a deterministic order_by, and were tested with more than one matching row.assertNumQueries test with several rows, not one.update() or bulk_create().IntegrityError inside transaction.atomic().EXPLAIN output was checked on a realistic data volume.Found a mistake in this tutorial? Let us know and we will fix it.