Novedades en 4.1

Versión menor. Sin cambios incompatibles vs 4.0 — toda novedad es opt-in o gratis cuando no se usa.

1. Backend CockroachDB

CockroachDB habla el protocolo PostgreSQL, así que dorm reutiliza la pipeline psycopg y publica una subclase con puerto 26257 + helper de reintento ante errores de serialización.

# settings.py
DATABASES = {
    "default": {
        "ENGINE": "cockroachdb",
        "NAME": "defaultdb",
        "HOST": "localhost",
        "PORT": 26257,
        "USER": "root",
        "PASSWORD": "",
        "OPTIONS": {"sslmode": "disable"},
    }
}

Instalación: pip install 'djanorm[cockroachdb]' (alias de [postgresql]). vendor permanece "postgresql" para que cada feature exclusiva de PG (CreatePGEnum, copy_from, particionado, materialised views, HStoreField, …) siga funcionando; el nuevo atributo dialect = "cockroachdb" es la salida opt-in cuando hace falta divergir.

Reintento ante fallo de serialización (SQLSTATE 40001 / 40003):

import dorm
from dorm.contrib.cockroach import retry_on_serialization, with_retry


def transfer() -> None:
    with dorm.transaction.atomic():
        src = Account.objects.select_for_update().get(pk=1)
        src.balance -= 100
        src.save()

# Forma funcional:
retry_on_serialization(transfer, max_attempts=5)


# Forma decorador (detecta async automáticamente):
@with_retry(max_attempts=5)
def transfer_money(src_id: int, dst_id: int, amount: int) -> None:
    with dorm.transaction.atomic():
        ...

2. AsyncSchemaEditor

Fachada async sobre SchemaEditor — cada helper DDL delega en asyncio.to_thread para que un hook de arranque FastAPI / Litestar no bloquee el event loop.

from dorm.migrations.schema import AsyncSchemaEditor
from dorm.db.connection import get_connection


async def init_schema() -> None:
    async with AsyncSchemaEditor(get_connection()) as se:
        await se.acreate_model(Article)
        await se.aadd_field(Article, "summary", dorm.TextField(null=True))

3. Migraciones Row-Level Security (PostgreSQL)

Ciclo completo de políticas RLS como operaciones de migración.

from dorm.migrations.operations import (
    AlterPolicy,
    CreatePolicy,
    DropPolicy,
    EnableRowLevelSecurity,
    ForceRowLevelSecurity,
)

operations = [
    EnableRowLevelSecurity("articles"),
    ForceRowLevelSecurity("articles"),
    CreatePolicy(
        "p_tenant_isolation",
        "articles",
        command="ALL",
        using="tenant_id = current_setting('app.tenant_id')::int",
        check="tenant_id = current_setting('app.tenant_id')::int",
    ),
]

CreatePolicy rechaza políticas SELECT/UPDATE/DELETE/ALL sin using= explícito (e INSERT sin check=) para que ningún predicado silencioso se cuele en una migración sensible. Ops PostgreSQL-only, no-op en otros backends.

4. Etiquetado PII

class User(dorm.Model):
    email = dorm.EmailField(pii=True)
    full_name = dorm.CharField(max_length=120, pii=True)
    username = dorm.CharField(max_length=40, unique=True)

El flag no afecta al esquema SQL. Consumidores:

from dorm.contrib.pii import (
    anonymize_row,
    has_pii_fields,
    mask_dict,
    mask_instance,
    pii_fields,
)

pii_fields(User)
# [<EmailField: email>, <CharField: full_name>]

mask_instance(user)
# user.email == "[REDACTED]", user.full_name == "[REDACTED]"

anonymize_row(user)   # enmascarado + .save() para derecho al olvido

Activa enmascarado del histórico para modelos @track_history:

dorm.configure(
    HISTORY_MASK_PII=True,
    ...,
)

5. Compatibilidad PgBouncer (transaction pool)

PgBouncer en transaction-pool reutiliza conexiones backend entre transacciones cliente, así que las prepared statements sobreviven al cliente que las creó. El nuevo ajuste PGBOUNCER_MODE las deshabilita en cada conexión.

DATABASES = {
    "default": {
        "ENGINE": "postgresql",
        "NAME": "app",
        # True, "transaction" o "statement" — desactiva la caché de
        # prepared statements de psycopg.
        "PGBOUNCER_MODE": "transaction",
    }
}

El ajuste sobreescribe cualquier PREPARE_THRESHOLD explícito y emite un warning cuando lo hace.

6. Middleware ASGI

Tres middlewares ASGI-3 puros, agnósticos al framework:

from fastapi import FastAPI

from dorm.contrib.asgi import (
    NPlusOneMiddleware,
    OTelDormMiddleware,
    QueryBudgetMiddleware,
)

app = FastAPI()
app.add_middleware(NPlusOneMiddleware, threshold=10)
app.add_middleware(QueryBudgetMiddleware, timeout_ms=2000, max_rows=10_000)
app.add_middleware(OTelDormMiddleware)

El orden importa — el más externo (último add_middleware) ve la request primero. OTel va fuera para que budget / N+1 / queries dorm queden anidadas bajo el span de la request.

Cada middleware ignora scopes lifespan / websocket — no son per-request.

7. Plugin Litestar

from litestar import Litestar
from dorm.contrib.litestar import dorm_plugin

app = Litestar(
    plugins=[
        dorm_plugin(
            budget_timeout_ms=2000,
            nplusone_threshold=10,
            warmup_pool=5,  # abre 5 conexiones al arrancar
        )
    ],
)

Conecta los tres middlewares ASGI vía DefineMiddleware y registra hooks async de startup / shutdown que calientan el pool al arranque y lo drenan limpiamente al apagar. Instalación: pip install 'djanorm[litestar]'.

8. connection_for(alias) / aconnection_for(alias)

Acceso público al wrapper backend por alias, equivalente al connections['default'] de Django:

import dorm

conn = dorm.connection_for("replica")
rows = conn.execute("SELECT 1")

aconn = await dorm.aconnection_for("replica")
rows = await aconn.execute("SELECT 1")

Sustituye al dorm.transaction._connection_for privado.

9. QuerySet.bulk_update_when / abulk_update_when

Compila una lista de (Q, {field: value}) en un único UPDATE … SET col = CASE WHEN … THEN … END WHERE ….

from dorm.expressions import Q

Article.objects.filter(published=True).bulk_update_when(
    [
        (Q(score__gte=90), {"label": "A", "featured": True}),
        (Q(score__gte=70), {"label": "B"}),
    ],
    default={"label": "C", "featured": False},
)

Sin default, la columna se preserva (ELSE col) en filas que no cumplan ninguna condición. QuerySet.update también acepta expresiones Case directamente.

10. Volcado QueryLog

Exporta consultas capturadas a JSON, JSONL o Parquet para pipelines de replay / análisis:

from dorm.contrib.querylog import QueryLog

with QueryLog() as log:
    do_work()

log.dump_json("queries.json")
log.dump_jsonl("queries.jsonl")           # streaming, memoria O(1)
log.dump_parquet("queries.parquet")       # requiere djanorm[parquet]

Los parámetros se excluyen por defecto — pasa include_params=True para incluirlos, y combínalo con dorm.contrib.pii.mask_dict(Model, row) para enmascarado previo.

---

Cambios completos en el CHANGELOG del proyecto. Subir de 4.0 → 4.1 no necesita modificar código.