django_utils package

Subpackages

Submodules

django_utils.aggregates module

Aggregates computed in a subquery, immune to JOIN fan-out.

The classic footgun: annotate(Count('review'), Count('topping')) implements each aggregate as a JOIN, so aggregating two one-to-many relations together counts the cartesian product: a sandwich with 2 reviews and 3 toppings reports 6 of each. Each helper here runs its aggregate in an independent (SELECT ... FROM (subquery)) instead, so combining any number of them stays correct:

from django.db.models import OuterRef
from django_utils.aggregates import SubqueryCount, SubquerySum

Sandwich.objects.annotate(
    reviews=SubqueryCount(Review.objects.filter(sandwich=OuterRef('pk'))),
    topping_total=SubquerySum(
        Topping.objects.filter(sandwich=OuterRef('pk')), 'price'
    ),
)

Annotations built this way support filter() and order_by() like any other. SubqueryCount of an empty set is 0; the column aggregates return NULL (Python None) for an empty set, matching SQL. Wrap in django.db.models.functions.Coalesce for a default.

Backend honesty: the correlated subquery lives inside a FROM-clause derived table (FROM (SELECT ...) _agg/_count). MySQL earlier than 8.0.14 cannot reference the outer query's columns from within a derived table and fails loudly with an unknown-column error. Verified on SQLite (this repo's CI); expected to work on PostgreSQL and on MySQL/MariaDB 8.0.14+, but not CI-verified on either.

The column aggregates (SubquerySum, SubqueryAvg, SubqueryMin, SubqueryMax) reserve agg_value as the inner annotation alias. A queryset whose model has a real field or annotation named agg_value raises Django's annotation-conflict error.

class django_utils.aggregates.SubqueryAvg(queryset: models.QuerySet[Any] | str, column: str, *, output_field: models.Field[Any, Any] | None = None, **extra: Any)[source]

Bases: _SubqueryColumnAggregate

AVG(column); defaults to FloatField output because SQL AVG of integers is fractional.

function = 'AVG'
class django_utils.aggregates.SubqueryCount(queryset: QuerySet | str, **extra: Any)[source]

Bases: _RelationNameMixin, Subquery

COUNT(*) over queryset, evaluated as its own subquery.

Unsliced inner querysets get their ordering stripped (pointless in an aggregate); sliced ones keep it (a top-N slice needs its ordering).

queryset may also be a relation name on the model being annotated (SubqueryCount('review')), resolved at resolve_expression time against reverse FK, reverse M2M, or forward M2M relations.

template = '(SELECT COUNT(*) FROM (%(subquery)s) _count)'
class django_utils.aggregates.SubqueryMax(queryset: models.QuerySet[Any] | str, column: str, *, output_field: models.Field[Any, Any] | None = None, **extra: Any)[source]

Bases: _SubqueryColumnAggregate

MAX(column); None for an empty set.

function = 'MAX'
class django_utils.aggregates.SubqueryMin(queryset: models.QuerySet[Any] | str, column: str, *, output_field: models.Field[Any, Any] | None = None, **extra: Any)[source]

Bases: _SubqueryColumnAggregate

MIN(column); None for an empty set.

function = 'MIN'
class django_utils.aggregates.SubquerySum(queryset: models.QuerySet[Any] | str, column: str, *, output_field: models.Field[Any, Any] | None = None, **extra: Any)[source]

Bases: _SubqueryColumnAggregate

SUM(column); None for an empty set.

function = 'SUM'

django_utils.auth module

The auth helpers Django is missing.

login_required and permission_required ship with Django; superuser_required/staff_required do not (django/new-features #47) and get re-implemented in nearly every project. Likewise the 'app_label.action_modelname' string every user.has_perm() call needs is hand-formatted everywhere (django/new-features #137); permission_string builds it from the model.

django_utils.auth.permission_string(model: type[Model], action: str) → str[source]

Return the 'app_label.action_modelname' string has_perm wants.

>>> from tests.test_app import models as test_models
>>> permission_string(test_models.Sandwich, 'change')
'test_app.change_sandwich'
django_utils.auth.staff_required(view_func: _View | None = None, *, login_url: str | None = None, raise_exception: bool = False) → _View | Callable[[_View], _View][source]

Allow only users with is_staff. See superuser_required.

Works on both sync and async views.

django_utils.auth.superuser_required(view_func: _View | None = None, *, login_url: str | None = None, raise_exception: bool = False) → _View | Callable[[_View], _View][source]

Allow only users with is_superuser.

Mirrors login_required's shape: use bare (@superuser_required) or parameterized (@superuser_required(raise_exception=True)). Failing users are redirected to login (login_url or settings.LOGIN_URL); with raise_exception=True they get PermissionDenied (HTTP 403) instead, matching permission_required's option of the same name. Works on both sync and async views.

django_utils.base_models module

class django_utils.base_models.CreatedAtModelBase(*args, **kwargs)[source]

Bases: ModelBase

class Meta[source]

Bases: object

abstract = False
db_table = 'django_utils_created_at_model_base'
created_at

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_next_by_created_at(*, field=<django.db.models.fields.DateTimeField: created_at>, is_next=True, **kwargs)
get_next_by_updated_at(*, field=<django.db.models.fields.DateTimeField: updated_at>, is_next=True, **kwargs)
get_previous_by_created_at(*, field=<django.db.models.fields.DateTimeField: created_at>, is_next=False, **kwargs)
get_previous_by_updated_at(*, field=<django.db.models.fields.DateTimeField: updated_at>, is_next=False, **kwargs)
updated_at

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

class django_utils.base_models.ModelBase(*args, **kwargs)[source]

Bases: Model

class Meta[source]

Bases: object

abstract = False
db_table = 'django_utils_model_base'
class django_utils.base_models.ModelBaseMeta(name: str, bases: tuple[type, ...], attrs: dict[str, Any], **kwargs: Any)[source]

Bases: ModelBase

Model base with more readable naming convention

Example: Assuming the model is called app.FooBarObject

Default Django table name: app_foobarobject Table name with this base: app_foo_bar_object

class django_utils.base_models.NameCreatedAtModelBase(*args, **kwargs)[source]

Bases: NameModelBase, CreatedAtModelBase

class Meta[source]

Bases: object

abstract = False
db_table = 'django_utils_name_created_at_model_base'
created_at

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_next_by_created_at(*, field=<django.db.models.fields.DateTimeField: created_at>, is_next=True, **kwargs)
get_next_by_updated_at(*, field=<django.db.models.fields.DateTimeField: updated_at>, is_next=True, **kwargs)
get_previous_by_created_at(*, field=<django.db.models.fields.DateTimeField: created_at>, is_next=False, **kwargs)
get_previous_by_updated_at(*, field=<django.db.models.fields.DateTimeField: updated_at>, is_next=False, **kwargs)
name: Any

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

updated_at

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

class django_utils.base_models.NameMixin[source]

Bases: object

Mixin to automatically get a unicode and repr string base on the name

>>> x = NameMixin()
>>> x.pk = 123
>>> x.name = 'test'
>>> repr(x)
'<NameMixin[123]: test>'
>>> str(x)
'test'
>>> str(str(x))
'test'
name: Any
pk: Any
class django_utils.base_models.NameModelBase(*args, **kwargs)[source]

Bases: NameMixin, ModelBase

class Meta[source]

Bases: object

abstract = False
db_table = 'django_utils_name_model_base'
name: Any

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

class django_utils.base_models.SlugCreatedAtModelBase(*args, **kwargs)[source]

Bases: SlugModelBase, CreatedAtModelBase

class Meta[source]

Bases: object

abstract = False
db_table = 'django_utils_slug_created_at_model_base'
created_at

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_next_by_created_at(*, field=<django.db.models.fields.DateTimeField: created_at>, is_next=True, **kwargs)
get_next_by_updated_at(*, field=<django.db.models.fields.DateTimeField: updated_at>, is_next=True, **kwargs)
get_previous_by_created_at(*, field=<django.db.models.fields.DateTimeField: created_at>, is_next=False, **kwargs)
get_previous_by_updated_at(*, field=<django.db.models.fields.DateTimeField: updated_at>, is_next=False, **kwargs)
name: Any

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

slug: Any

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

updated_at

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

class django_utils.base_models.SlugMixin[source]

Bases: NameMixin

Mixin to automatically slugify the name and add both a name and slug to the model

>>> x = NameMixin()
>>> x.pk = 123
>>> x.name = 'test'
>>> repr(x)
'<NameMixin[123]: test>'
>>> str(x)
'test'
>>> str(str(x))
'test'

Note: this mixin does not add a unique constraint on slug. The nested Meta.unique_together below is inert -- SlugMixin is not itself a Model, and a concrete subclass such as SlugModelBase declares its own Meta, which does not inherit from this one. Models that need the slug to be enforced unique at the database level should declare unique=True on their own slug field.

class Meta[source]

Bases: object

unique_together = ('slug',)
get_unique_slug(base: str) → str[source]

Return base, suffixed with a counter if already taken.

Probes with _base_manager -- Django's documented unfiltered manager -- rather than _default_manager, so a model with a filtered default manager (soft-delete style: objects = ActiveManager()) still gets checked against every row in the table, not just the ones its default manager happens to expose. Otherwise two rows hidden from each other by the filter can silently collide onto the same slug.

Note: the uniqueness check and the eventual insert are not atomic, so two concurrent saves can still race each other onto the same slug. SlugMixin does not add a unique constraint (see the class docstring), so callers with high write concurrency on the same name should declare unique=True on their own slug field and handle the resulting IntegrityError.

save(*args: Any, **kwargs: Any) → None[source]
slug: Any
slugify_max_attempts: ClassVar[int] = 1000
class django_utils.base_models.SlugModelBase(*args, **kwargs)[source]

Bases: SlugMixin, NameModelBase

class Meta[source]

Bases: object

abstract = False
db_table = 'django_utils_slug_model_base'
name: Any

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

slug: Any

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

django_utils.bulk module

Bulk upsert built on Django's native conflict handling.

bulk_update_or_create wraps bulk_create(update_conflicts=True) (Django 4.1+) with chunking and up-front validation. Unlike a loop of update_or_create (2N queries, racy between check and write), the upsert happens in one INSERT ... ON CONFLICT DO UPDATE statement per batch, atomic per row on the database side.

Backend honesty: PostgreSQL and SQLite use ON CONFLICT with unique_fields naming the conflict target; MySQL/MariaDB use ON DUPLICATE KEY UPDATE, which ignores unique_fields and fires on any unique constraint: identical behaviour when the model has one unique constraint, subtly broader when it has several.

Version honesty: on Django 5.0+ the returned objects have their primary keys populated (inserted and conflict-updated rows alike), where the backend can return rows from a bulk insert at all: PostgreSQL and SQLite can, MariaDB can, vanilla MySQL never can (its Django backend disables row-returning inserts on every Django version). On Django 4.2, bulk_create(update_conflicts=True) cannot return IDs anywhere (Django ticket #34698, fixed in 5.0), so every returned object has pk=None even though its row was written.

Field names accept the same spellings bulk_create does: field names (owner), foreign-key attnames (owner_id), and the 'pk' alias.

django_utils.bulk.bulk_update_or_create(objs: Iterable[M], *, unique_fields: Sequence[str], update_fields: Sequence[str], batch_size: int = 1000) → list[M][source]

Insert objs, updating update_fields on conflicts.

Rows whose unique_fields already exist are updated instead of raising IntegrityError; new rows are inserted. Returns the input objects as a list. All validation errors raise ValueError before any query runs -- including when objs is empty, so a bad batch_size or field list is never masked by an empty input.

Each batch is committed independently; there is no transaction spanning all batches. That is deliberate -- a crash partway through a large run leaves the earlier batches durable and the operation resumable. Wrap the call in django.db.transaction.atomic() yourself if you need all-or-nothing semantics instead.

Writes always go through the model's default database (Model._base_manager, no using override); routing to a non-default alias is a known gap -- there is no using= parameter.

django_utils.choices module

Usage

Create a Choices class and add Choice objects to the class to define your choices.

Example with explicit values:

The normal Django version:

class Human(models.Model):
    GENDER = (
        ('m', 'Male'),
        ('f', 'Female'),
        ('o', 'Other'),
    )
    gender = models.CharField(max_length=1, choices=GENDER)

The Django Utils Choices version:

from django_utils import choices


class Human(models.Model):
    class Gender(choices.Choices):
        Male = choices.Choice('m')
        Female = choices.Choice('f')
        Other = choices.Choice('o')

    gender = models.CharField(max_length=1, choices=Gender)

To reference these properties:

Human.create(gender=Human.Gender.Male)

Example with implicit values:

The normal Django version:

class SomeModel(models.Model):
    SOME_ENUM = (
        (1, 'foo'),
        (2, 'bar'),
        (3, 'spam'),
        (4, 'eggs'),
    )
    enum = models.IntegerField(choices=SOME_ENUM, default=1)

The Django Utils Choices version:

from django_utils import choices


class SomeModel(models.Model):
    class Enum(choices.Choices):
        Foo = choices.Choice()
        Bar = choices.Choice()
        Spam = choices.Choice()
        Eggs = choices.Choice()

    enum = models.IntegerField(choices=Enum, default=Enum.Foo)

To reference these properties:

SomeModel.create(enum=SomeModel.Enum.Spam)

Excluding constants

Any plain str, int or float class attribute becomes a choice. To keep a constant alongside your choices, list it in _ignore_:

class Gender(choices.Choices):
    _ignore_ = ('MAX_LENGTH',)
    MAX_LENGTH = 1

    Male = choices.Choice('m')
    Female = choices.Choice('f')

Like the standard library's enum, _ignore_ also accepts a single string of names separated by whitespace and/or commas, which is split for you:

class Gender(choices.Choices):
    _ignore_ = 'MAX_LENGTH'
    MAX_LENGTH = 1

    Male = choices.Choice('m')
    Female = choices.Choice('f')

Attaching metadata to choices

Unlike Django's TextChoices, a Choice can carry arbitrary extra data, reachable as attributes:

class Status(choices.Choices):
    Active = choices.Choice('a', 'Active', color='green')
    Closed = choices.Choice('c', 'Closed', color='red')


Status.choices['a'].color  # 'green'

Grouped choices

Give choices a group and hand Django the nested structure it renders as <optgroup>:

class Product(choices.Choices):
    Apple = choices.Choice('ap', 'Apple', group='Fruit')
    Carrot = choices.Choice('ca', 'Carrot', group='Vegetable')


field = models.CharField(max_length=2, choices=Product.choices.grouped())

Getting a real Enum

Members of a Choices class are raw values, so they can be passed straight to Django fields. When you want isinstance checks or a match on real enum members, ask for an enum:

class Gender(choices.Choices):
    Male = choices.Choice('m', 'Male')
    Female = choices.Choice('f', 'Female')


GenderEnum = Gender.as_enum()
GenderEnum('m') is GenderEnum.Male  # True
class django_utils.choices.Choice(value: Any = None, label: str | StrPromise | None = None, **metadata: Any)[source]

Bases: object

The choice object has an optional label and value. If the value is not given an autoincrementing id (starting from 1) will be used

>>> choice = Choice('value', 'label')
>>> choice
<Choice[1]:label>
>>> str(choice)
'label'

>>> choice = Choice()
>>> choice
<Choice[2]:None>
>>> str(choice)
'None'
deconstruct() → tuple[str, tuple[Any, str | StrPromise | None], dict[str, Any]][source]
order: int = 0
class django_utils.choices.Choices[source]

Bases: object

The choices class is what you should inherit in your Django models

>>> choices = Choices()
>>> choices.choices[0]
Traceback (most recent call last):
...
KeyError: 'Key 0 does not exist'
>>> choices.choices
OrderedDict()
>>> str(choices.choices)
'OrderedDict()'
>>> choices.choices.items()
[]
>>> choices.choices.keys()
[]
>>> choices.choices.values()
[]
>>> list(choices)
[]

>>> class ChoiceTest(Choices):
...     a = Choice()
>>> choices = ChoiceTest()
>>> choices.choices.items()
[(0, <Choice[...]:a>)]
>>> choices.a
0
>>> choices.choices['a']
<Choice[...]:a>
>>> choices.choices[0]
<Choice[...]:a>
>>> choices.choices.keys()
[0]
>>> choices.choices.values()
['a']
>>> list(choices)
[(0, <Choice[...]:a>)]
>>> list(ChoiceTest)
[(0, <Choice[...]:a>)]
choices = OrderedDict()
class django_utils.choices.ChoicesDict[source]

Bases: object

The choices dict is an object that stores a sorted representation of the values by key and database value

by_key() → OrderedDict[str, Choice][source]

The choices keyed by their declared attribute name.

grouped() → list[tuple[str | StrPromise, list[tuple[Any, str | StrPromise]]]][source]

The choices as Django's <optgroup> structure.

Choices declaring a group metadata key are collected under it, in declaration order; ungrouped choices land under ''.

Labels (and group keys) are passed through unchanged, so a gettext_lazy label stays a lazy proxy -- resolved at render time, not when grouped() is called. This matters because the module docstring's own example calls grouped() at model-field definition time (import time), so eagerly resolving here would freeze translations in whatever language happened to be active at import.

items() → list[tuple[Any, Choice]][source]
keys() → list[Any][source]
values() → list[str][source]
class django_utils.choices.ChoicesMeta(name: str, bases: tuple[type, ...], attrs: dict[str, Any])[source]

Bases: type

The choices metaclass is where all the magic happens, this automatically creates a ChoicesDict to get a sorted list of keys and values

as_enum() → type[Enum][source]

Build a real enum.Enum from these choices.

The result is memoised on the class: repeated calls return the same enum class, so Gender.as_enum() is Gender.as_enum() and Gender.as_enum().Male is Gender.as_enum().Male both hold, and the enum can be used as (or as part of) a dict/cache key. The cache is stored in cls.__dict__ and looked up there directly (never via getattr, which walks the MRO), so a subclass builds and caches its own enum rather than inheriting its parent's.

The returned class is built dynamically via enum.Enum's functional API with its __module__ set to django_utils.choices rather than the caller's module, so its members are not picklable with the default pickle protocol (pickling an enum member looks the class up by __module__ + qualified name, which won't resolve back to a class that was never assigned a name in that module).

The original class is unchanged: its members stay raw values so they can be handed to Django fields. Use the enum where you want isinstance checks or a match on real enum members.

choices: ChoicesDict
class django_utils.choices.LiteralChoices[source]

Bases: Choices

Special version of the Choices class that uses the label as the value

>>> class Role(LiteralChoices):
...     admin = Choice()
...     user = Choice()
...     guest = Choice()

>>> Role.choices.values()
['admin', 'user', 'guest']
>>> Role.choices.keys()
['admin', 'user', 'guest']

>>> class RoleWithImplicitChoice(LiteralChoices):
...     ADMIN = 'admin'
...     USER = 'user'
...     GUEST = 'guest'

>>> Role.choices.values()
['admin', 'user', 'guest']
>>> Role.choices.keys()
['admin', 'user', 'guest']
>>> Role.admin
'admin'
choices = OrderedDict()

django_utils.context module

Current-request/current-user access, contextvars-native.

RequestContextMiddleware stores each request in a contextvars.ContextVar for exactly the duration of its request/response cycle, making the request reachable from code with no request argument: model save() methods, signal handlers, log filters, template-independent helpers.

Why not a thread-local (django-crum, django-currentuser)? Under ASGI a single thread's event loop interleaves many requests, so thread-local state leaks between them. A ContextVar is isolated per asyncio task and propagated into sync_to_async threads by asgiref, which makes it correct under WSGI and ASGI alike.

Everything here is opt-in: without the middleware (or the current_request context manager) the getters simply return None.

class django_utils.context.RequestContextMiddleware(get_response: Callable[[HttpRequest], HttpResponseBase] | Callable[[HttpRequest], Awaitable[HttpResponseBase]])[source]

Bases: object

Store each request in the context variable for its whole cycle.

Add to MIDDLEWARE (any position; before auth middleware is fine because get_current_user reads the user lazily). The variable is reset in a finally so an exception anywhere downstream cannot leak one request into the next. That reset happens as soon as the view returns a response, before a StreamingHttpResponse body iterator runs -- so get_current_request() called from inside one returns None.

async_capable = True
sync_capable = True
django_utils.context.current_request(request: HttpRequest) → Generator[HttpRequest, None, None][source]

Make request current for the duration of the block.

For tests, shell sessions and management commands, anywhere no middleware runs. Nests: the previous request is restored on exit.

django_utils.context.get_current_request() → HttpRequest | None[source]

Return the request currently being served, or None outside one.

django_utils.context.get_current_user() → _User | None[source]

Return request.user for the current request.

None when there is no current request or when the request has no user attribute (AuthenticationMiddleware absent). The user is read lazily off the stored request, so middleware order relative to RequestContextMiddleware does not matter, only that auth middleware ran before this call.

django_utils.crypto_fields module

Fernet-encrypted model fields, behind the crypto extra.

EncryptedCharField, EncryptedTextField and EncryptedJSONField store a base64 Fernet token in a plain TEXT column (get_internal_type() returns 'TextField' for all three, so the column is always sized for ciphertext, never plaintext). Encryption uses cryptography's Fernet/MultiFernet exclusively -- nothing hand-rolled, no hazmat primitives touched directly.

Keys come from settings.DJANGO_UTILS_FERNET_KEYS, a list of urlsafe-base64 32-byte keys (Fernet.generate_key()). The FIRST key encrypts; EVERY key is tried on decrypt (MultiFernet). Rotate by prepending a new key and redeploying: old rows keep decrypting under the old key until they are next saved, at which point they are re-encrypted under the new (first) key. There is no bulk re-encryption command here -- touch (.save()) the rows you want migrated, or run your own queryset_iterator-based pass over the table.

Non-goals, loudly: this is encryption at rest for values you never need to query, sort or index by. Ordering is not blocked (Django has no field-level hook for it) but sorts by ciphertext -- meaningless. And anything reading through the ORM sees plaintext: an admin export action (django_utils.admin.export.ExportMixin) on an encrypted model streams decrypted values, and dumpdata writes plaintext fixtures. There is no queryable/searchable mode, no per-field keys, and no deterministic (same-plaintext-same-ciphertext) mode -- Fernet salts every encryption, so even two rows with identical plaintext get different ciphertext, and equality can never match at the database level. Every lookup except isnull therefore raises NotImplementedError; filter in Python after decrypting, or maintain a separate searchable hash column alongside the encrypted one if you need to find rows by value.

Requires the crypto extra (pip install 'django-utils2[crypto]'): importing this module works without cryptography installed, but instantiating any of the three fields raises ImproperlyConfigured with that install hint. Fields instantiate as part of executing a model's class body, so a model that declares one of them fails at app-import/django.setup() time without the extra -- the whole app fails to boot, loudly and immediately, not a deferred per-use error.

class django_utils.crypto_fields.EncryptedCharField(*args: Any, **kwargs: Any)[source]

Bases: _EncryptedField

Fernet-encrypted str, stored as a TEXT column.

max_length validates the PLAINTEXT via a MaxLengthValidator (mirroring plain CharField's own behaviour); it never sizes the column, which stores the necessarily-longer ciphertext instead.

formfield(**kwargs: Any) → Field | None[source]

Return a django.forms.Field instance for this field.

class django_utils.crypto_fields.EncryptedJSONField(*args: Any, **kwargs: Any)[source]

Bases: _EncryptedField

Fernet-encrypted JSON-serializable value, stored as TEXT.

formfield(**kwargs: Any) → Field | None[source]

Return a django.forms.Field instance for this field.

class django_utils.crypto_fields.EncryptedTextField(*args: Any, **kwargs: Any)[source]

Bases: _EncryptedField

Fernet-encrypted, unbounded str, stored as a TEXT column.

formfield(**kwargs: Any) → Field | None[source]

Return a django.forms.Field instance for this field.

django_utils.fields module

class django_utils.fields.RecursiveField(field_name: str | None = None, parent_field: str = 'parent', default: Any = None)[source]

Bases: object

PREFIX: ClassVar[str] = 'get_'
contribute_to_class(cls: type, name: str) → None[source]
get(instance: Any) → Any[source]

django_utils.middleware module

Standalone middleware.

FetchMetadataMiddleware implements modern, header-based CSRF protection (the approach django/new-features #98 proposes for core and Go's standard library ships): browsers have sent the Sec-Fetch-Site request header on every request since roughly 2019, so cross-site state-changing requests can be rejected without tokens, template tags or AJAX header plumbing.

Deliberately defense-in-depth: run it alongside CsrfViewMiddleware, never instead of it. A request carrying neither Sec-Fetch-Site nor Origin (an old browser, curl, a server-side client) is allowed through. Rejecting those is exactly the token middleware's job, which is why this middleware alone is not sufficient protection.

class django_utils.middleware.FetchMetadataMiddleware(get_response: Callable[[HttpRequest], HttpResponseBase] | Callable[[HttpRequest], Awaitable[HttpResponseBase]])[source]

Bases: object

Reject cross-site state-changing requests by request headers.

Policy, in order:

  1. Safe methods (GET/HEAD/OPTIONS/TRACE) always pass.

  2. Views marked @fetch_metadata_exempt pass.

  3. Sec-Fetch-Site present: allow same-origin/same-site/ none (browser UI, e.g. the address bar); reject anything else with 403. Unknown values fail closed.

  4. No Sec-Fetch-Site: compare Origin to this request's scheme://host; mismatch is rejected.

  5. Neither header: allow. Token CSRF remains the backstop.

Behind a TLS-terminating proxy, step 4 can false-reject: without SECURE_PROXY_SSL_HEADER configured, request.scheme reads http while a legacy browser's Origin header (the client sees only the outer HTTPS connection) is https://..., so the exact-match comparison fails and the request is rejected as cross-site. Modern browsers are unaffected -- they send Sec-Fetch-Site, which step 3 handles first. Configure SECURE_PROXY_SSL_HEADER per the Django docs to fix request.scheme itself.

async_capable = True
process_view(request: HttpRequest, callback: Callable[[...], Any], callback_args: tuple[Any, ...], callback_kwargs: dict[str, Any]) → HttpResponseBase | None[source]
sync_capable = True
django_utils.middleware.fetch_metadata_exempt(view_func: _View) → _View[source]

Mark a view as exempt from FetchMetadataMiddleware.

Same shape as csrf_exempt; works on sync and async views.

django_utils.pg_enum module

PostgreSQL ENUM-backed model field, with explicit migration operations.

EnumField is a CharField wired to a django_utils.choices Choices class: choices/max_length are derived from it, and on PostgreSQL the column's real type is a native CREATE TYPE ... AS ENUM type instead of VARCHAR -- the database itself then rejects a row that doesn't hold one of the declared values, on top of (not instead of) Django's own choice validation. On every other backend db_type() falls back to plain VARCHAR, so a model using EnumField stays portable -- SQLite never sees a Postgres-specific type name.

The field never issues DDL for the enum type itself -- that is the job of three explicit migrations.Operation subclasses you add to a migration BY HAND. Django's autodetector has no concept of "create this standalone database object first", so these are never auto-generated by makemigrations:

CreateEnumType(name, values)

CREATE TYPE <name> AS ENUM (<values>). Add it to the same migration as -- and before -- the AddField/CreateModel operation that introduces the EnumField column using it; the type must exist before a column can reference it.

AddEnumValue(name, value)

ALTER TYPE <name> ADD VALUE IF NOT EXISTS <value>. PostgreSQL cannot run this statement inside a transaction block on versions before 12, and even on 12+ a value added in one transaction can't be used in that same transaction. It sets atomic = False, but that alone is NOT enough: Django's migration executor opens its schema editor -- and with it, the wrapping transaction -- keyed on the Migration's atomic attribute (default True), before any operation's own atomic flag is ever consulted; the operation-level flag can only add extra wrapping inside that, never escape it. Put AddEnumValue in its own migration with atomic = False set as a CLASS attribute on that Migration (same convention as Django's own AddIndexConcurrently). Skip that and database_forwards raises NotSupportedError instead of running -- it never applies the value from inside a transaction. It is also irreversible -- PostgreSQL has no DROP VALUE -- so reversing it raises IrreversibleError. The recreate-and-migrate escape hatch (create a new type with the reduced value list, migrate the column and data across, drop the old type) is documented in the README.

DropEnumType(name, values)

The reverse of CreateEnumType. It takes the same values as CreateEnumType so that ITS OWN reversal (recreating the type) has something to recreate.

All three are DB-only: state_forwards is a no-op (there is no model state for a standalone database type -- EnumField doesn't need one either, its deconstruct() carries the Choices class, not the type's existence) and database_forwards/database_backwards are no-ops on every non-PostgreSQL vendor, so a migration using them still applies cleanly against SQLite or MySQL -- just without the enum type's extra database-level integrity check. SQL is always built with schema_editor.quote_name() for identifiers and schema_editor.quote_value() for values, never raw string interpolation of unquoted input.

class django_utils.pg_enum.AddEnumValue(*args, **kwargs)[source]

Bases: Operation

Add a value to an existing PostgreSQL ENUM type, by hand, in its own migration entry -- see the module docstring for why this is atomic = False and irreversible.

atomic = False on THIS operation is necessary but not sufficient: Django's migration executor opens its schema editor -- and with it, the wrapping transaction -- keyed on the Migration's atomic attribute (default True), before any operation's own atomic flag is consulted. Put this operation in its own migration with atomic = False set as a CLASS attribute on that Migration (same convention as Django's own AddIndexConcurrently). Without that, database_forwards raises NotSupportedError rather than running inside a transaction it cannot safely run inside.

atomic = False
database_backwards(app_label: str, schema_editor: BaseDatabaseSchemaEditor, from_state: ProjectState, to_state: ProjectState) → None[source]

Perform the mutation on the database schema in the reverse direction - e.g. if this were CreateModel, it would in fact drop the model's table.

database_forwards(app_label: str, schema_editor: BaseDatabaseSchemaEditor, from_state: ProjectState, to_state: ProjectState) → None[source]

Perform the mutation on the database schema in the normal (forwards) direction.

describe() → str[source]

Output a brief summary of what the action does.

state_forwards(app_label: str, state: ProjectState) → None[source]

Take the state from the previous migration, and mutate it so that it matches what this migration would perform.

class django_utils.pg_enum.CreateEnumType(*args, **kwargs)[source]

Bases: Operation

Create a PostgreSQL ENUM type. Add to a migration BY HAND, before the operation that adds a column using it -- see the module docstring.

database_backwards(app_label: str, schema_editor: BaseDatabaseSchemaEditor, from_state: ProjectState, to_state: ProjectState) → None[source]

Perform the mutation on the database schema in the reverse direction - e.g. if this were CreateModel, it would in fact drop the model's table.

database_forwards(app_label: str, schema_editor: BaseDatabaseSchemaEditor, from_state: ProjectState, to_state: ProjectState) → None[source]

Perform the mutation on the database schema in the normal (forwards) direction.

describe() → str[source]

Output a brief summary of what the action does.

state_forwards(app_label: str, state: ProjectState) → None[source]

Take the state from the previous migration, and mutate it so that it matches what this migration would perform.

class django_utils.pg_enum.DropEnumType(*args, **kwargs)[source]

Bases: Operation

Drop a PostgreSQL ENUM type. The reverse of CreateEnumType -- takes the same values so reversing THIS operation (recreating the type) has something to recreate.

database_backwards(app_label: str, schema_editor: BaseDatabaseSchemaEditor, from_state: ProjectState, to_state: ProjectState) → None[source]

Perform the mutation on the database schema in the reverse direction - e.g. if this were CreateModel, it would in fact drop the model's table.

database_forwards(app_label: str, schema_editor: BaseDatabaseSchemaEditor, from_state: ProjectState, to_state: ProjectState) → None[source]

Perform the mutation on the database schema in the normal (forwards) direction.

describe() → str[source]

Output a brief summary of what the action does.

state_forwards(app_label: str, state: ProjectState) → None[source]

Take the state from the previous migration, and mutate it so that it matches what this migration would perform.

class django_utils.pg_enum.EnumField(choices_class: type[Choices], *, enum_type: str | None = None, **kwargs: Any)[source]

Bases: CharField

CharField whose choices/max_length come from a django_utils.choices.Choices class, with a native PostgreSQL ENUM column type (plain VARCHAR on every other backend).

choices is always derived from choices_class -- there is no override; passing a choices= keyword too raises TypeError (this __init__ rejects it explicitly, before deriving anything). max_length is derived as the longest value's length unless given explicitly. enum_type defaults to choices_class's name in snake_case (SandwichStatus -> 'sandwich_status') and names the PostgreSQL type this field's column references -- that type must already exist by the time this field's table/column is created; see CreateEnumType in the module docstring.

db_type(connection: BaseDatabaseWrapper) → str | None[source]

Return the database column data type for this field, for the provided connection.

deconstruct() → tuple[str, str, Sequence[Any], dict[str, Any]][source]

Return enough information to recreate the field as a 4-tuple:

  • The name of the field on the model, if contribute_to_class() has been run.

  • The import path of the field, including the class, e.g. django.db.models.IntegerField. This should be the most portable version, so less specific may be better.

  • A list of positional arguments.

  • A dict of keyword arguments.

Note that the positional or keyword arguments must contain values of the following types (including inner values of collection types):

  • None, bool, str, int, float, complex, set, frozenset, list, tuple, dict

  • UUID

  • datetime.datetime (naive), datetime.date

  • top-level classes, top-level functions - will be referenced by their full import path

  • Storage instances - these have their own deconstruct() method

This is because the values here must be serialized into a text format (possibly new Python code, possibly JSON) and these are the only types with encoding handlers defined.

There's no need to return the exact way the field was instantiated this time, just ensure that the resulting field is the same - prefer keyword arguments over positional ones, and omit parameters with their default values.

django_utils.query_debug module

Query budgets that are safe to leave in production code paths.

query_budget counts every query a block executes and logs a structured warning (warn_at) or raises (raise_at) when the block exceeds its budget: an always-on guard against N+1 regressions, unlike test-only tools (assertNumQueries) or dev-only profilers (django-silk, django-debug-toolbar).

The load-bearing design decision: counting is implemented with connection.execute_wrapper(), which is public, documented, and active regardless of DEBUG, and free of the per-connection query-log accumulation that force_debug_cursor/connection.queries would cause in a long-lived production process.

exception django_utils.query_debug.QueryBudgetExceeded(count: int, raise_at: int, last_sql: str)[source]

Bases: Exception

A block exceeded its raise_at query budget.

class django_utils.query_debug.query_budget(warn_at: int | None = None, raise_at: int | None = None, *, using: str | None = None)[source]

Bases: ContextDecorator

Count queries in a block; warn and/or raise over budget.

Usable as a context manager or a decorator:

with query_budget(warn_at=20, raise_at=100):
    ...


@query_budget(warn_at=20)
def view(request): ...

warn_at logs a single warning on exit when exceeded (with the most-repeated statement, the classic N+1 signature). raise_at raises QueryBudgetExceeded from the first over-budget query. using limits counting to one connection alias; the default counts every configured connection. Each decorated call gets a fresh budget; instances are single-use per with (not reentrant). Instances are not thread-safe; the decorator form is safe because each call gets a fresh instance.

django_utils.queryset module

django_utils.queryset.queryset_iterator(queryset: ~django.db.models.query.QuerySet, chunksize: int = 1000, getfunc: ~collections.abc.Callable[[~typing.Any, str], ~typing.Any] = <built-in function getattr>, *, pk_field: str = 'pk', start_after: ~typing.Any = None, gc_collect: bool = False) → Generator[Any, None, None][source]

Iterate over a Django queryset in fixed-size chunks.

Uses keyset pagination (WHERE <pk_field> > <cursor> LIMIT chunksize) instead of a database cursor: each chunk is fetched as its own independent query rather than as part of one long-lived cursor kept open for the whole iteration. The database driver therefore never has to hold more than one chunk's worth of rows at a time -- whatever the driver does internally, it does it chunksize rows at a time.

Why this function exists

This function was written after QuerySet.iterator() exhausted memory on a very large table. Splitting one unbounded query into many bounded LIMIT queries fixed that, and the same query shape brings four distinct benefits along with it -- only the first of which is about the client driver.

Bounded memory on the Python side. Only chunksize model instances are ever alive at once, rather than the whole table. Core's own QuerySet.iterator(chunk_size=...) makes the same promise, but it only holds if the driver streams: with PostgreSQL and Django's default server-side cursors, the database -- not the driver -- holds the unfetched rows, so iterator() is the better choice there. Where the driver buffers instead of streaming -- verified for MySQL with mysqlclient, whose default cursor calls store_result() and materialises the entire result set client-side before Python sees the first row, in a C buffer chunk_size never touches -- the whole table is already sitting in the client process before chunk_size gets a chance to bound anything. Oracle (python-oracledb) and PostgreSQL with DISABLE_SERVER_SIDE_CURSORS = True are driver-dependent in the same way, though that is not established here the way mysqlclient's is. SQLite has no server-side cursor to disable either, but the stdlib sqlite3 module steps rows lazily rather than buffering the whole result set (see the benchmark section below).

Bounded memory on the database server. Each chunk is issued as WHERE <pk_field> > cursor ORDER BY <pk_field> LIMIT chunksize -- an index range scan the planner can satisfy incrementally and stop as soon as it has chunksize rows. A single unbounded SELECT ... ORDER BY over a large table can instead force the server to materialise and sort the entire result set before returning the first row, consuming work_mem (PostgreSQL) or sort_buffer_size (MySQL) and spilling to disk when it doesn't fit. The chunked form never asks the server to hold more than one chunk at a time. This follows from the query shape, not from measurement here.

Read-replica distribution. Because each chunk is its own independent statement, a Django database router or a connection pooler can spread the N chunk queries across read replicas. A single long-running query pins one connection to one server for its whole duration and cannot be load-balanced mid-flight, so chunked iteration can scale horizontally in a way one cursor cannot -- again by construction, not measured here.

Operational blast radius. A single heavy query is a single point of failure: it can exhaust server memory, hit a statement timeout, and hold its transaction snapshot open for as long as it runs -- which on PostgreSQL keeps vacuum from reclaiming dead rows and causes table bloat, and on MySQL/InnoDB grows the undo history. Short per-chunk queries release their snapshot between chunks instead, so a slow consumer doesn't hold the database hostage. They also fail gracefully and resumably: a chunk that fails partway through a run can be retried from the last cursor seen instead of restarting the whole query, which is exactly what start_after below is for.

None of points two through four were independently benchmarked in this repository -- they follow from the query shape described above. The only measured numbers here are the query counts and wall-clock figures below, and the cost is real: N small queries instead of 1, plus a modest wall-clock overhead.

Benchmark

benchmarks/queryset_iterator.py, SQLite, 20,000 rows, chunksize=1000. Wall clock is close to parity with QuerySet.iterator() (~1.09x-1.14x across runs; query count and wall clock are measured in separate passes so that query-logging instrumentation doesn't inflate the many-query path's timing more than the single-query path's). An earlier version of this function called gc.collect() after every chunk; that call alone added roughly 37-47% to queryset_iterator()'s default-mode wall-clock time for no measured reduction in Python-level memory, so it is off by default here (see gc_collect below).

The benchmark's tracemalloc peak figures are not evidence about the memory story either way: tracemalloc only sees Python-level allocations, so it cannot see a driver's C-level result buffer -- the thing this function exists to bound. SQLite also has no server-side cursor to disable in the first place, and its stdlib driver does not buffer the full result set either, so it structurally cannot reproduce the failure mode this function guards against. That failure mode is verified on MySQL with mysqlclient, and may also occur -- driver-dependent, not established here -- on Oracle and on PostgreSQL with DISABLE_SERVER_SIDE_CURSORS = True; it does not occur on SQLite, and none of this shows up in a tracemalloc trace on any backend. No OOM has been measured here; the claim is narrower: bounded driver-side memory by construction, not a demonstrated fix for a specific crash.

Choosing between the two

Prefer QuerySet.iterator(chunk_size=...) on PostgreSQL with server-side cursors enabled (the default): one query, and the database -- not the driver -- holds the unfetched rows. Reach for this function when that guarantee isn't available (MySQL, Oracle, SQLite, or PostgreSQL with server-side cursors disabled), where a single unbounded query can hand the driver an unbounded buffer. It is also useful whenever a single long-lived query/cursor is undesirable for another reason: because each chunk is its own query, iteration can resume on a connection that was reset or recycled between chunks, which one open cursor cannot.

Keyword-only options

pk_field

Column to order and paginate by, instead of the primary key. Must be unique, indexed, totally ordered and non-nullable so keyset pagination doesn't skip or repeat rows -- e.g. a monotonic created_at or integer id column. Useful when the primary key itself is a randomly-ordered UUID that would make every WHERE pk > cursor query scan out of index order. A NULL value read from pk_field raises ValueError, since the keyset cursor cannot resume from NULL. A non-unique value silently truncates: if several rows tie on the same pk_field value, only the ones that land in the current chunk are yielded and the rest of that tied group is skipped.

start_after

Resume from a previously-seen cursor value instead of starting from the beginning, so a batch job that died partway through can continue rather than restart. When given, the first chunk is filtered > start_after instead of being unfiltered. The value must be a value of pk_field (a plain pk by default):

last_seen = None
for row in queryset_iterator(
    MyModel.objects.all(),
    pk_field='created_at',
    start_after=last_seen,
):
    process(row)
    last_seen = row.created_at
    checkpoint(last_seen)  # survives a restart
gc_collect

Call gc.collect() after every chunk. Off by default: in the SQLite benchmark above it adds roughly 37-47% to this function's default-mode wall-clock time and made no measured difference to peak Python-level memory. It is available for runs where reclaiming Python-level garbage more aggressively than the allocator would on its own is worth that cost -- e.g. very large runs where resident set size matters more than throughput.

Note that the results are always ordered by pk_field.

django_utils.utils module

django_utils.utils.to_json(request: HttpRequest, data: Any) → HttpResponse[source]

django_utils.view_decorators module

class django_utils.view_decorators.EnvRequest[source]

Bases: HttpRequest

A request as seen inside an env-decorated view.

The env decorator injects these attributes at runtime; this subclass exists so type checkers know about them.

ajax: bool
context: dict[str, Any] | None
not_found: type[HttpResponseNotFound]
permanent_redirect: Callable[[...], HttpResponsePermanentRedirect]
redirect: Callable[[...], HttpResponseRedirect]
reverse: Callable[[...], str]
template: str
exception django_utils.view_decorators.UnknownViewResponseError[source]

Bases: ViewError

exception django_utils.view_decorators.ViewError[source]

Bases: Exception

django_utils.view_decorators.debug_allowed(request: HttpRequest) → bool[source]

Whether the ?debug=1 HTML view may render for this request.

Rendering a response body inside an HTML document is a debugging aid, so it is restricted the way django-debug-toolbar restricts its own panels: settings.DEBUG, or an explicitly whitelisted client address.

Note that behind a reverse proxy REMOTE_ADDR is the proxy's address, not the client's, so INTERNAL_IPS will match nothing (or everything, if the proxy's own address is listed). If your deployment sits behind a reverse proxy and needs the real client address (e.g. from X-Forwarded-For), reassign this function (django_utils.view_decorators.debug_allowed = my_check) rather than trusting that header here, since it is only meaningful when the proxy topology is known.

django_utils.view_decorators.env(function: Callable[[...], Any]) → Callable[[...], HttpResponse][source]
django_utils.view_decorators.env(function: None = None, login_required: bool = False, response_class: type[HttpResponse] = http.HttpResponse) → Callable[[Callable[[...], Any]], Callable[[...], HttpResponse]]

View decorator that automatically adds context and renders response

Keyword arguments: login_required -- is everyone allowed or only authenticated users

Adds a RequestContext (request.context) with the following context items: name -- current function name

Stores the template in request.template and assumes it to be in <app>/<view>.html

django_utils.view_decorators.json_default_handler(obj: Any) → str[source]
django_utils.view_decorators.permanent_redirect(url: str, *args: Any, **kwargs: Any) → HttpResponsePermanentRedirect[source]
django_utils.view_decorators.redirect(url: str = './', *args: Any, **kwargs: Any) → HttpResponseRedirect[source]

django_utils.views module

django_utils.views.error_400(request: HttpRequest, exception: Exception) → None[source]
django_utils.views.error_403(request: HttpRequest, exception: Exception) → None[source]
django_utils.views.error_404(request: HttpRequest, exception: Exception) → None[source]
django_utils.views.error_500(request: HttpRequest) → None[source]

Module contents