Custom managers, raw SQL and EXPLAIN

nivas/chained.py

"""The notice board's question, what does this resident see today, asked with the methods of NoticeQuerySet in
notices/models.py, chained after Notice.objects. The day is fixed, 1 October 2026, so that the run reads the same on
any day. Run it next to manage.py.
"""

from datetime import date

import seed  # noqa: F401  (starts Django, creates the tables, adds the rows)
from django.db import connection
from django.test.utils import CaptureQueriesContext

from notices.models import Complaint, Notice, Resident
from seed import where

DAY = date(2026, 10, 1)
asha = Resident.objects.select_related('flat').get(name='Asha K.')
rohan = Resident.objects.select_related('flat').get(name='Rohan M.')

print(f'1. Notice.objects.active(on={DAY}).for_flat({asha.flat}).visible_to({asha})')
board = Notice.objects.active(on=DAY).for_flat(asha.flat).visible_to(asha)
with CaptureQueriesContext(connection) as queries:
    titles = [notice.title for notice in board]
print(f'   {len(queries)} query, {len(titles)} notices: {" | ".join(titles)}')
print('  ', where(queries[0]['sql']))

print(f'2. The same chain for {rohan} ({rohan.flat}, on the committee)')
for notice in Notice.objects.active(on=DAY).for_flat(rohan.flat).visible_to(rohan):
    print('  ', notice.title)

print('3. Which methods are where')
print('   Notice.objects.active:', hasattr(Notice.objects, 'active'), '| Notice.objects.post:',
      hasattr(Notice.objects, 'post'))
print('   Notice.objects.all().active:', hasattr(Notice.objects.all(), 'active'),
      '| Notice.objects.all().post:', hasattr(Notice.objects.all(), 'post'))
print('   Notice.objects.post() created:', Notice.objects.post('Garden tap repaired', by=rohan, category='water'))

print("4. Complaint.objects = ComplaintQuerySet.as_manager()")
print("   Complaint.objects.still_open().in_tower('B').count():", Complaint.objects.still_open().in_tower('B').count())

1. Notice.objects.active(on=2026-10-01).for_flat(B-1203).visible_to(Asha K.)
   1 query, 2 notices: Water cut in tower B, 10:00 to 14:00 | Cleanliness drive on Sunday
   WHERE ((notice.ends_on IS NULL OR notice.ends_on >= '2026-10-01')
     AND notice.published = 1
     AND notice.starts_on <= '2026-10-01'
     AND (notice.tower = '' OR notice.tower = 'B')
     AND NOT (notice.audience = 'committee'))
2. The same chain for Rohan M. (B-1204, on the committee)
   Water cut in tower B, 10:00 to 14:00
   Cleanliness drive on Sunday
   Security guard rota for October
   Tower B committee meeting notes
3. Which methods are where
   Notice.objects.active: True | Notice.objects.post: True
   Notice.objects.all().active: True | Notice.objects.all().post: False
   Notice.objects.post() created: Garden tap repaired
4. Complaint.objects = ComplaintQuerySet.as_manager()
   Complaint.objects.still_open().in_tower('B').count(): 2

notices/models.py · nivas/notices/models.py

from django.db import models
from django.db.models import Q
from django.db.models.functions import Lower
from django.utils import timezone


class Flat(models.Model):
    code = models.CharField(max_length=10, unique=True)  # 'B-1203': tower B, flat 1203
    tower = models.CharField(max_length=1)

    def __str__(self):
        return self.code


class Resident(models.Model):
    name = models.CharField(max_length=100)
    email = models.EmailField()
    phone = models.CharField(max_length=15)
    flat = models.ForeignKey(Flat, on_delete=models.CASCADE, related_name='residents')
    on_committee = models.BooleanField(default=False)

    class Meta:
        constraints = [
            # one account per email, whatever the capital letters: a unique index on LOWER(email)
            models.UniqueConstraint(Lower('email'), name='one_resident_per_email'),
        ]

    def __str__(self):
        return self.name


class NoticeQuerySet(models.QuerySet):
    """The questions the notice board asks, each a method that returns a QuerySet, so they chain."""

    def active(self, on=None):
        """Published, started, and not yet ended on the given day (today if none is given)."""
        on = on or timezone.localdate()
        return self.filter(Q(ends_on__isnull=True) | Q(ends_on__gte=on), published=True, starts_on__lte=on)

    def for_flat(self, flat):
        """Notices for the whole society (no tower) or for the flat's own tower."""
        return self.filter(Q(tower='') | Q(tower=flat.tower))

    def visible_to(self, resident):
        """Committee notices are for committee members only."""
        if resident.on_committee:
            return self
        return self.exclude(audience=Notice.Audience.COMMITTEE)


class NoticeManager(models.Manager):
    def post(self, title, by, **fields):
        """A method of the manager only: there is no QuerySet to start from when a notice is created."""
        return self.create(title=title, posted_by=by, starts_on=timezone.localdate(), **fields)


class Notice(models.Model):
    class Audience(models.TextChoices):
        EVERYONE = 'everyone'
        COMMITTEE = 'committee'

    title = models.CharField(max_length=120)
    category = models.CharField(max_length=20)  # water, lift, security, events …
    tower = models.CharField(max_length=1, blank=True)  # '' means the whole society
    audience = models.CharField(max_length=10, choices=Audience, default=Audience.EVERYONE)
    published = models.BooleanField(default=True)
    starts_on = models.DateField()
    ends_on = models.DateField(null=True, blank=True)
    posted_by = models.ForeignKey(Resident, on_delete=models.PROTECT, related_name='notices')

    objects = NoticeManager.from_queryset(NoticeQuerySet)()

    def __str__(self):
        return self.title


class ComplaintQuerySet(models.QuerySet):
    def still_open(self):
        return self.filter(closed_on__isnull=True)

    def in_tower(self, tower):
        return self.filter(tower=tower)


class Complaint(models.Model):
    tower = models.CharField(max_length=1)
    subject = models.CharField(max_length=120)
    raised_on = models.DateField()
    closed_on = models.DateField(null=True, blank=True)

    objects = ComplaintQuerySet.as_manager()  # the QuerySet's public methods, copied onto a manager

    def __str__(self):
        return self.subject


class PublishedManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().filter(published=True)


class Circular(models.Model):
    """The mistake: the filtered manager is declared first, so it is the default manager."""

    title = models.CharField(max_length=120)
    published = models.BooleanField(default=False)

    objects = PublishedManager()
    every = models.Manager()

    def __str__(self):
        return self.title


class Minutes(models.Model):
    """The fix: a plain manager first, the filtered one under its own name."""

    title = models.CharField(max_length=120)
    published = models.BooleanField(default=False)

    objects = models.Manager()
    published_only = PublishedManager()

    class Meta:
        verbose_name_plural = 'minutes'

    def __str__(self):
        return self.title


class Attachment(models.Model):
    circular = models.ForeignKey(Circular, on_delete=models.CASCADE, related_name='attachments')
    file_name = models.CharField(max_length=100)

    def __str__(self):
        return self.file_name

nivas/default_manager.py

"""Which manager Django's own tools use. Circular declares a manager that hides unpublished rows first, so it is the
default manager; Minutes declares a plain manager first and the filtered one under its own name. Both tables hold
three rows, one of them a draft. Run it next to manage.py.
"""

import io
import json

import seed  # noqa: F401  (starts Django, creates the tables, adds the rows)
from django.contrib import admin
from django.core.management import call_command
from django.test import RequestFactory

from notices.models import Attachment, Circular, Minutes


def dumped(label, *options):
    out = io.StringIO()
    call_command('dumpdata', label, *options, stdout=out)
    return [row['fields']['title'] for row in json.loads(out.getvalue())]


def in_admin(model):
    """The rows the admin's change list starts from: ModelAdmin.get_queryset()."""
    return admin.ModelAdmin(model, admin.site).get_queryset(RequestFactory().get('/admin/')).count()


print('Circular: objects = PublishedManager() comes first')
print('   default manager:', Circular._default_manager.name, '| rows in the table:', Circular.every.count(),
      '| Circular.objects.count():', Circular.objects.count())
print('   dumpdata notices.circular:', dumped('notices.circular'))
print('   dumpdata --all notices.circular:', len(dumped('notices.circular', '--all')), 'circulars | the admin lists',
      in_admin(Circular))
draft = Circular.every.get(published=False)
try:
    Circular.objects.get(pk=draft.pk)
except Circular.DoesNotExist:
    print(f'   Circular.objects.get(pk={draft.pk}): DoesNotExist')
attachment = Attachment.objects.get(file_name='agm-agenda.pdf')
print(f'   attachment.circular: {attachment.circular} (read with the base manager:',
      f'{type(Circular._base_manager).__name__})')

print('Minutes: objects = models.Manager() first, published_only = PublishedManager() second')
print('   default manager:', Minutes._default_manager.name, '| Minutes.published_only.count():',
      Minutes.published_only.count())
print('   dumpdata notices.minutes:', len(dumped('notices.minutes')), 'sets of minutes | the admin lists',
      in_admin(Minutes))

Circular: objects = PublishedManager() comes first
   default manager: objects | rows in the table: 3 | Circular.objects.count(): 2
   dumpdata notices.circular: ['Maintenance dues for 2026-27', 'New visitor parking rules']
   dumpdata --all notices.circular: 3 circulars | the admin lists 2
   Circular.objects.get(pk=1): DoesNotExist
   attachment.circular: AGM agenda (draft) (read with the base manager: Manager)
Minutes: objects = models.Manager() first, published_only = PublishedManager() second
   default manager: objects | Minutes.published_only.count(): 2
   dumpdata notices.minutes: 3 sets of minutes | the admin lists 3

notices/models.py · nivas/notices/models.py

from django.db import models
from django.db.models import Q
from django.db.models.functions import Lower
from django.utils import timezone


class Flat(models.Model):
    code = models.CharField(max_length=10, unique=True)  # 'B-1203': tower B, flat 1203
    tower = models.CharField(max_length=1)

    def __str__(self):
        return self.code


class Resident(models.Model):
    name = models.CharField(max_length=100)
    email = models.EmailField()
    phone = models.CharField(max_length=15)
    flat = models.ForeignKey(Flat, on_delete=models.CASCADE, related_name='residents')
    on_committee = models.BooleanField(default=False)

    class Meta:
        constraints = [
            # one account per email, whatever the capital letters: a unique index on LOWER(email)
            models.UniqueConstraint(Lower('email'), name='one_resident_per_email'),
        ]

    def __str__(self):
        return self.name


class NoticeQuerySet(models.QuerySet):
    """The questions the notice board asks, each a method that returns a QuerySet, so they chain."""

    def active(self, on=None):
        """Published, started, and not yet ended on the given day (today if none is given)."""
        on = on or timezone.localdate()
        return self.filter(Q(ends_on__isnull=True) | Q(ends_on__gte=on), published=True, starts_on__lte=on)

    def for_flat(self, flat):
        """Notices for the whole society (no tower) or for the flat's own tower."""
        return self.filter(Q(tower='') | Q(tower=flat.tower))

    def visible_to(self, resident):
        """Committee notices are for committee members only."""
        if resident.on_committee:
            return self
        return self.exclude(audience=Notice.Audience.COMMITTEE)


class NoticeManager(models.Manager):
    def post(self, title, by, **fields):
        """A method of the manager only: there is no QuerySet to start from when a notice is created."""
        return self.create(title=title, posted_by=by, starts_on=timezone.localdate(), **fields)


class Notice(models.Model):
    class Audience(models.TextChoices):
        EVERYONE = 'everyone'
        COMMITTEE = 'committee'

    title = models.CharField(max_length=120)
    category = models.CharField(max_length=20)  # water, lift, security, events …
    tower = models.CharField(max_length=1, blank=True)  # '' means the whole society
    audience = models.CharField(max_length=10, choices=Audience, default=Audience.EVERYONE)
    published = models.BooleanField(default=True)
    starts_on = models.DateField()
    ends_on = models.DateField(null=True, blank=True)
    posted_by = models.ForeignKey(Resident, on_delete=models.PROTECT, related_name='notices')

    objects = NoticeManager.from_queryset(NoticeQuerySet)()

    def __str__(self):
        return self.title


class ComplaintQuerySet(models.QuerySet):
    def still_open(self):
        return self.filter(closed_on__isnull=True)

    def in_tower(self, tower):
        return self.filter(tower=tower)


class Complaint(models.Model):
    tower = models.CharField(max_length=1)
    subject = models.CharField(max_length=120)
    raised_on = models.DateField()
    closed_on = models.DateField(null=True, blank=True)

    objects = ComplaintQuerySet.as_manager()  # the QuerySet's public methods, copied onto a manager

    def __str__(self):
        return self.subject


class PublishedManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().filter(published=True)


class Circular(models.Model):
    """The mistake: the filtered manager is declared first, so it is the default manager."""

    title = models.CharField(max_length=120)
    published = models.BooleanField(default=False)

    objects = PublishedManager()
    every = models.Manager()

    def __str__(self):
        return self.title


class Minutes(models.Model):
    """The fix: a plain manager first, the filtered one under its own name."""

    title = models.CharField(max_length=120)
    published = models.BooleanField(default=False)

    objects = models.Manager()
    published_only = PublishedManager()

    class Meta:
        verbose_name_plural = 'minutes'

    def __str__(self):
        return self.title


class Attachment(models.Model):
    circular = models.ForeignKey(Circular, on_delete=models.CASCADE, related_name='attachments')
    file_name = models.CharField(max_length=100)

    def __str__(self):
        return self.file_name

nivas/plan_login.py

"""The login lookup, a resident by email whatever the capital letters, planned by PostgreSQL 18: read with explain(),
then with explain(analyze=True), which runs the query. The table holds 5,000 residents (residents.py) and the unique
index one_resident_per_email on LOWER(email) (notices/models.py); the lookup is Django's iexact. PostgreSQL runs on a
throwaway server of its own (pgserver.py). SQLite plans the same query in the last line.

Times change on every run, so the run prints them as …; everything else is as PostgreSQL printed it.
"""

import re

import pgserver

pgserver.start()

import residents  # noqa: E402  (5,000 residents and fresh statistics)
from django.db import connection  # noqa: E402
from django.test.utils import CaptureQueriesContext  # noqa: E402

from notices.models import Resident  # noqa: E402
from seed import where  # noqa: E402

TIME = re.compile(r'(actual time=)[\d.]+\.\.[\d.]+|(Time: )[\d.]+')


def plan(text):
    return '\n'.join('     ' + line for line in TIME.sub(lambda m: (m.group(1) or m.group(2)) + '…', text).splitlines())


login = Resident.objects.filter(email__iexact='asha.k@example.com')
print(f'{residents.server()} | {Resident.objects.count():,} residents in {residents.PAGES} pages')
with CaptureQueriesContext(connection) as sent:
    print("email__iexact='asha.k@example.com':", [resident.name for resident in login])
print('  ', where(sent[0]['sql']))
print('   explain():')
print(plan(login.explain()))
print('   explain(analyze=True):')
print(plan(login.explain(analyze=True)))
print('SQLite, explain():', login.using('sqlite').explain())

PostgreSQL 18.6 | 5,000 residents in 62 pages
email__iexact='asha.k@example.com': ['Asha K.']
   WHERE UPPER(resident.email::text) = UPPER('asha.k@example.com')
   explain():
     Seq Scan on notices_resident  (cost=0.00..137.00 rows=25 width=67)
       Filter: (upper((email)::text) = 'ASHA.K@EXAMPLE.COM'::text)
   explain(analyze=True):
     Seq Scan on notices_resident  (cost=0.00..137.00 rows=25 width=67) (actual time=… rows=1.00 loops=1)
       Filter: (upper((email)::text) = 'ASHA.K@EXAMPLE.COM'::text)
       Rows Removed by Filter: 4999
       Buffers: shared hit=62
     Planning Time: … ms
     Execution Time: … ms
SQLite, explain(): 2 0 216 SCAN notices_resident

nivas/residents.py

"""Adds 4,995 residents to the five of seed.py, in every database the settings name, and gathers PostgreSQL's
statistics with ANALYZE, as autovacuum would. The plan examples import it after pgserver.start().
"""

import seed  # noqa: F401  (starts Django, creates the tables, adds the rows)
from django.db import connection, connections

from notices.models import Flat, Resident

for alias in connections:
    flats = list(Flat.objects.using(alias).order_by('pk'))
    Resident.objects.using(alias).bulk_create([
        Resident(name=f'Resident {n}', email=f'resident{n}@nivas.example', phone=f'90000{n:05}', flat=flats[n % 5])
        for n in range(6, 5001)
    ])
with connection.cursor() as cursor:
    cursor.execute('ANALYZE notices_resident')
    cursor.execute("SELECT relpages, reltuples FROM pg_class WHERE relname = 'notices_resident'")
    PAGES, ROWS = cursor.fetchone()


def server():
    """PostgreSQL 18.6, as the server reports its version: 180006."""
    return f'PostgreSQL {connection.pg_version // 10000}.{connection.pg_version % 10000}'

nivas/pgserver.py

"""A throwaway PostgreSQL 18 server for one example: start() creates a database cluster in a folder of its own, starts
the server listening only on 127.0.0.1, an address no other computer can reach, and tells Django its port through
NIVAS_PG_PORT (config/settings.py reads it). The server is stopped and its folder deleted when the example ends, also
when it fails. Autovacuum is off, so the table statistics the plans use are the ones the example gathers itself, with
ANALYZE, and every run plans the same way.

It needs PostgreSQL's server programs, initdb and pg_ctl, on PATH (or in the mise installation beside this Python),
and the psycopg package, which Django uses to talk to PostgreSQL.
"""

import atexit
import os
import shutil
import socket
import subprocess
import sys
import tempfile
from pathlib import Path


def _postgres_bin():
    on_path = shutil.which('pg_ctl')
    if on_path:
        return Path(on_path).parent
    installs = Path(sys.executable).resolve().parents[3]  # <mise>/installs/python/3.14.8/bin/python3
    found = sorted(installs.glob('postgres/18.*/bin/pg_ctl'))
    if not found:
        sys.exit('PostgreSQL 18 was not found: install it so that pg_ctl is on PATH')
    return found[-1].parent


def start():
    pg = _postgres_bin()
    work = Path(tempfile.mkdtemp(prefix='pg', dir='.')).resolve()
    data = work / 'data'
    subprocess.run([pg / 'initdb', '-D', data, '-U', 'postgres', '-A', 'trust', '-E', 'UTF8', '--no-locale'],
                   check=True, capture_output=True)
    with socket.socket() as probe:  # ask the operating system for a port that nothing uses
        probe.bind(('127.0.0.1', 0))
        port = probe.getsockname()[1]
    with open(data / 'postgresql.conf', 'a') as conf:
        conf.write(f"listen_addresses = '127.0.0.1'\nport = {port}\nunix_socket_directories = ''\nautovacuum = off\n")
    subprocess.run([pg / 'pg_ctl', '-D', data, '-l', work / 'server.log', '-w', 'start'], check=True,
                   capture_output=True)

    def stop():
        subprocess.run([pg / 'pg_ctl', '-D', data, '-m', 'fast', '-w', 'stop'], capture_output=True)
        shutil.rmtree(work, ignore_errors=True)

    atexit.register(stop)
    os.environ['NIVAS_PG_PORT'] = str(port)

nivas/ci_lookup.py

"""The ci lookup of notices/lookups.py against Django's iexact: the SQL each sends, and the plan PostgreSQL 18 and
SQLite choose for the login lookup on 5,000 residents (residents.py), with the unique index on LOWER(email). PostgreSQL
runs on a throwaway server of its own (pgserver.py).

Times change on every run, so the run prints them as …; everything else is as PostgreSQL printed it.
"""

import re

import pgserver

pgserver.start()

import residents  # noqa: E402  (5,000 residents and fresh statistics)
from django.db import connection  # noqa: E402
from django.test.utils import CaptureQueriesContext  # noqa: E402

from notices.models import Resident  # noqa: E402
from seed import where  # noqa: E402

TIME = re.compile(r'(actual time=)[\d.]+\.\.[\d.]+|(Time: )[\d.]+')


def plan(text):
    return '\n'.join('     ' + line for line in TIME.sub(lambda m: (m.group(1) or m.group(2)) + '…', text).splitlines())


print(f'{residents.server()} | {Resident.objects.count():,} residents')
for lookup in ['iexact', 'ci']:
    search = Resident.objects.filter(**{f'email__{lookup}': 'Asha.K@example.COM'})
    with CaptureQueriesContext(connection) as sent:
        names = [resident.name for resident in search]
    print(f"email__{lookup}='Asha.K@example.COM': {names}")
    print('  ', where(sent[0]['sql']))
    if lookup == 'ci':
        print('   explain(analyze=True):')
        print(plan(search.explain(analyze=True)))
    print('   SQLite:', search.using('sqlite').explain())

PostgreSQL 18.6 | 5,000 residents
email__iexact='Asha.K@example.COM': ['Asha K.']
   WHERE UPPER(resident.email::text) = UPPER('Asha.K@example.COM')
   SQLite: 2 0 216 SCAN notices_resident
email__ci='Asha.K@example.COM': ['Asha K.']
   WHERE LOWER(resident.email) = LOWER('Asha.K@example.COM')
   explain(analyze=True):
     Index Scan using one_resident_per_email on notices_resident  (cost=0.28..8.30 rows=1 width=67) (actual time=… rows=1.00 loops=1)
       Index Cond: (lower((email)::text) = 'asha.k@example.com'::text)
       Index Searches: 1
       Buffers: shared hit=3
     Planning Time: … ms
     Execution Time: … ms
   SQLite: 3 0 39 SEARCH notices_resident USING INDEX one_resident_per_email (<expr>=?)

notices/lookups.py · nivas/notices/lookups.py

"""A case-insensitive equality that a unique index on LOWER(email) can serve: email__ci='Asha.K@Example.com' becomes
LOWER("email") = LOWER(%s). Django's own iexact compares UPPER(…) on PostgreSQL instead, which that index cannot
serve.
"""

from django.db.models import CharField, Lookup


@CharField.register_lookup  # EmailField and SlugField are CharFields, so they get it too
class CaseInsensitiveEqual(Lookup):
    lookup_name = 'ci'

    def as_sql(self, compiler, connection):
        lhs, lhs_params = self.process_lhs(compiler, connection)
        rhs, rhs_params = self.process_rhs(compiler, connection)
        # Since Django 6.0 the params of as_sql() are a tuple.
        return f'LOWER({lhs}) = LOWER({rhs})', (*lhs_params, *rhs_params)

notices/apps.py · nivas/notices/apps.py

from django.apps import AppConfig


class NoticesConfig(AppConfig):
    name = 'notices'

    def ready(self):
        from . import lookups  # noqa: F401  (registers the ci lookup on CharField and its subclasses)