What's new in 4.3¶

Minor release. No breaking changes vs 4.2 — every addition is opt-in.

Bug fixes (release-blocker round)¶

Manager.upsert(returning=True) previously passed Field objects to bulk_create; now uses field names + skips M2M.
InboxRecord docstring requires Meta.unique_together = [("message_id", "handler_name")] on the concrete subclass so the DB serialises concurrent duplicates.
Saga.run() inside an outer atomic() logs a WARNING — step commits become savepoints and lose durability if the outer transaction rolls back.
anonymize_model uses cursor pagination (pk__gt=last_pk) instead of offset — O(N) on million-row tables.
two_phase_commit rejects nested atomic() blocks with a clear RuntimeError.
@temporal docstring warns that bulk_create / bulk_update / queryset.update / queryset.delete bypass post_save / post_delete and miss the temporal mirror.

Tier 1 — strategic¶

Cross-vendor UPSERT¶

Model.objects.upsert(
    [Model(slug="a", score=1), Model(slug="b", score=2)],
    unique_fields=["slug"],
)

Sugar over bulk_create(update_conflicts=True). Defaults update_fields to every non-PK / non-unique column.

Temporal tables¶

from dorm.contrib.temporal import temporal, as_of

@temporal
class Article(dorm.Model):
    title = dorm.CharField(max_length=200)

snapshot = as_of(Article, t)   # rows valid at time t
history = Article.history.filter(pk=42)  # every version

Sibling <Model>Temporal table mirrors every write with valid_from / valid_to / operation columns.

`dorm.contrib.tasks` — Celery-lite¶

from dorm.contrib.tasks import TaskQueue, task

queue = TaskQueue(model=BackgroundTask, channel="tasks")

@task(queue, name="send-welcome")
def send_welcome(user_id: int) -> None: ...

with dorm.transaction.atomic():
    user = User.objects.create(...)
    send_welcome.delay(user.id)

# Worker process:
queue.run()

Built on outbox + LISTEN/NOTIFY. Polling fallback when LISTEN/NOTIFY isn't available.

Tier 2 — extra field types¶

from dorm.contrib.extra_fields import (
    MoneyField, SemverField, PhoneField, ColorField, JSONSchemaField
)

MoneyField(currency="EUR") — Decimal + ISO-4217 currency.
SemverField() — SemVer 2.0.0 grammar.
PhoneField() — E.164.
ColorField() — #RRGGBB / #RRGGBBAA.
JSONSchemaField(schema={...}) — JSONField + per-assignment JSON Schema validation (requires jsonschema).

Tier 3 — async/sync ergonomics¶

`SyncDataLoader`¶

loader = SyncDataLoader(lambda ids: Author.objects.in_bulk(ids))
for book in books:
    loader.load(book.author_id)
loader.flush()
for book in books:
    author = loader.load(book.author_id)

`BackgroundTasks`¶

from dorm.contrib.background import BackgroundTasks

bg = BackgroundTasks(concurrency=8)
bg.add(send_email, user.email)
bg.add(warm_cache)
await bg.run()

Tier 4 — distributed transactions¶

Saga¶

from dorm.contrib.saga import Saga, Step

saga = Saga(steps=[
    Step("reserve_inventory", reserve, release),
    Step("charge_card", charge, refund),
    Step("ship", ship, cancel_shipping),
])
run = saga.run({"order_id": 42})
if not run.ok:
    log.error("saga failed at %s, compensated %s",
              run.failure[0], run.compensated)

Two-phase commit (PG)¶

from dorm.contrib.two_phase import two_phase_commit

with two_phase_commit(["primary", "warehouse"]) as txn:
    txn.execute("primary", "INSERT INTO orders ...")
    txn.execute("warehouse", "UPDATE stock ...")

Inbox (exactly-once consumer)¶

from dorm.contrib.inbox import InboxRecord, idempotent

class Inbox(InboxRecord):
    class Meta:
        db_table = "inbox"

@idempotent(Inbox)
def handle_payment(message_id: str, payload: dict) -> None: ...

Tier 5 — observability / perf¶

Plan-drift history(tag=None) + clear_history — bounded ring (50/tag).
dorm_pool_warmup_seconds{alias} Prometheus gauge (auto-recorded by warmup_pool).
RowCache(Model, maxsize=...) — per-model LRU with auto-invalidation on save/delete.
suggest_fix(template, model_hint=...) — N+1 detector returns the specific select_related / prefetch_related recommendation.

Tier 6 — DX¶

dorm init --template fastapi-postgres
dorm init --template litestar-sqlite

Scaffolds settings.py, app/__init__.py, app/models.py and app/main.py ready to run.

Tier 7 — sugar¶

F("nick").coalesce("anon")
# ↓ same as Coalesce(F("nick"), Value("anon"))

subtree_filter(Node, root_pk=1, where_sql="is_active = ?", where_params=[True])

Tier 8 — security¶

from dorm.contrib.anonymizer import anonymize_model

anonymize_model(User, {
    "email": "random_email",
    "phone": "random_phone",
    "notes": "redact",
    "credit_card": lambda v: "•••• •••• •••• ••••",
})

from dorm.contrib.auth.tokens import rotate_short_lived_token

new_tok, old_tok = rotate_short_lived_token(current_token)
# Persist new + leave old active during the grace window.

from dorm.db.utils import add_sensitive_pattern, reset_sensitive_patterns

add_sensitive_pattern("ssn", "credit_card", "iban")

Tier 2 expansion — TaskQueue scheduled / cron / priorities¶

from dorm.contrib.tasks import TaskQueue, task, run_cron_tick

queue = TaskQueue(model=BackgroundTask, channel="tasks", max_attempts=5)

@task(queue, name="reindex", priority=5)
def reindex(): ...

@task(queue, name="nightly", cron="0 3 * * *")
def nightly_job(): ...

# Schedule delayed work:
reindex.delay(eta=datetime.utcnow() + timedelta(minutes=10))
reindex.delay(delay_seconds=30, priority=1)

# Cron tick — wire to APScheduler / external cron / dorm worker:
run_cron_tick(queue)

Helpers:

retry_with_backoff(queue, event) — exponential ETA for relay implementations.
dead_letters(queue) — queryset over status='dead'.

Tier 3 expansion — Migration linter v2 + new ops¶

dorm lint-migrations

New codes:

DORM-M005 — operation is irreversible (reversible=False).
DORM-M006 — destructive RunSQL (DROP TABLE / DROP COLUMN / TRUNCATE) with empty reverse_sql.

New operations:

from dorm.migrations.operations import (
    AddCheckConstraintOnline,
    SeederMigration,
)

operations = [
    # PG: ADD CONSTRAINT NOT VALID + VALIDATE CONSTRAINT — no
    # ACCESS EXCLUSIVE rewrite scan.
    AddCheckConstraintOnline("orders", "chk_pos_amount", "amount > 0"),

    # Seed reference data inside a migration:
    SeederMigration(
        [
            {"model": "shop.Currency", "fields": {"code": "USD", "name": "US Dollar"}},
            {"model": "shop.Currency", "fields": {"code": "EUR", "name": "Euro"}},
        ],
    ),
]

Tier 4 expansion — DX¶

DORM_PROFILE=prod dorm migrate
# Imports settings.prod from settings/prod.py transparently.

Tier 6 expansion — new field types¶

from dorm.contrib.extra_fields import (
    IPRangeField, TimezoneField, PathField,
    PercentageField, CountryField, autoslug,
)

class User(dorm.Model):
    # UUIDv7: time-ordered, PK-friendly.
    id = dorm.UUIDField(primary_key=True, version=7)
    email = dorm.EmailField(pii=True)
    network = IPRangeField(null=True)           # "10.0.0.0/24"
    tz = TimezoneField(default="UTC")
    avatar_path = PathField()
    completion = PercentageField()              # [0, 100]
    country = CountryField()                    # "ES", "US", ...
    slug = dorm.SlugField(default=autoslug("name"))

Tier 7 expansion — permissions DSL + rate limiting¶

from dorm.contrib.permissions import requires, requires_any

@requires("article.edit", scope="article")
def edit(user, article, payload): ...

@requires_any("admin.read", "audit.read")
def view_audit(user): ...

from dorm.contrib.rate_limit import TokenBucket, rate_limited

bucket = TokenBucket(rate_per_second=10, burst=20)

@rate_limited(bucket, key=lambda req: req.ip)
def signup(req): ...

Tier 8 — concurrency primitives¶

from dorm.contrib.concurrency import (
    named_lock, SerializableSnapshot, with_optimistic_lock,
)

# Distributed mutex (PG advisory lock; in-proc fallback elsewhere).
with named_lock("nightly-report"):
    run_nightly_report()

# SERIALIZABLE block with auto-retry on 40001.
SerializableSnapshot(max_attempts=5).run(transfer_money)

# Version-column optimistic concurrency control.
class Account(dorm.Model):
    balance = dorm.DecimalField(max_digits=12, decimal_places=2)
    version = dorm.IntegerField(default=0)

with_optimistic_lock(account)  # raises OptimisticLockError if stale

Tier 9 — hybrid vector + FTS rerank¶

from dorm.contrib.pgvector.reranker import rerank

results = rerank(
    Document,
    vector_field="embedding",
    text_field_tsv="search_vector",
    query_vector=embed("machine learning"),
    query_text="machine learning",
    candidates=200,
    weight_vector=0.7,
)

Tier 11 — active/passive router¶

from dorm.contrib.active_passive import ActivePassiveRouter

DATABASE_ROUTERS = [
    ActivePassiveRouter(aliases=["node_a", "node_b"], probe_seconds=10.0),
]

Probes pg_is_in_recovery() per alias on a cached schedule; writes flow to the discovered primary, reads to replicas.

Also: await awarmup_pool(target=5) for ASGI startup hooks.

Tier 12 — serialization helpers¶

from dorm.contrib.serializers import (
    stream_msgpack, avro_schema_for, openapi_schema_for,
)

# Streaming MessagePack (requires `pip install msgpack`):
for chunk in stream_msgpack(Article.objects.iterator()):
    yield chunk

avro_schema = avro_schema_for(Article)
openapi_components = {
    "schemas": {
        "Article": openapi_schema_for(Article),
    }
}

Tier 13 expansion — sugar¶

from dorm.expressions import F

# range predicate as F method
Person.objects.filter(F("age").between(18, 65))

# exists-first variant of get_or_create
exists, user = User.objects.exists_or_create(
    email="x@y", defaults={"name": "X"}
)

Tier 14 — slow-transaction detector¶

from dorm.contrib import slow_tx

slow_tx.install(threshold_ms=500)  # at process start

Every atomic() block longer than 500 ms emits a WARNING on dorm.contrib.slow_tx and attaches a dorm.slow_tx event to the active OTel span.

Deferred to v4.4¶

MSSQL / ClickHouse / Snowflake / Spanner backends.
Composite Foreign Keys (query compiler refactor).
TimescaleDB hypertable helpers.
dorm shell --notebook (Jupyter kernel).
pytest-djanorm factories (sibling package release).
Migration squash preserving RunPython (data-ops risk).

Upgrading 4.2 → 4.3 needs no code changes.