Saltar a contenido

Modelos y campos

Cada modelo dorm es una clase Python que hereda de dorm.Model y declara un campo por columna. La metaclase construye un registro _meta que la suite de migraciones, el query builder y el adaptador Pydantic introspectan.

Anatomía de un modelo

import dorm


class Author(dorm.Model):
    name = dorm.CharField(max_length=100)
    age = dorm.IntegerField()
    email = dorm.EmailField(unique=True, null=True, blank=True)

    class Meta:
        db_table = "authors"      # default: "<applabel>_<lowercase_name>"
        ordering = ["name"]       # orden por defecto en cada queryset

Si no declaras una primary key, dorm añade automáticamente una id implícita (un BigAutoField).

Referencia de campos

Strings

Campo Tipo BD Notas
CharField(max_length=N) VARCHAR(N) max_length obligatorio
TextField() TEXT sin límite
EmailField() VARCHAR(254) valida el formato al asignar
URLField() VARCHAR(200)
SlugField() VARCHAR(50) letras/dígitos/-/_, indexado
UUIDField() UUID (PG) / CHAR(36) (SQLite)
IPAddressField() / GenericIPAddressField() VARCHAR(45)

Números

Campo Tipo BD
IntegerField() INTEGER
SmallIntegerField() SMALLINT
BigIntegerField() BIGINT
PositiveIntegerField() / PositiveSmallIntegerField() / PositiveBigIntegerField() (3.1+) con CHECK
FloatField() DOUBLE PRECISION / REAL
DecimalField(max_digits=N, decimal_places=M) DECIMAL(N, M)

Tiempo

Campo Tipo BD
DateField() DATE
TimeField() TIME
DateTimeField(auto_now_add=False, auto_now=False) TIMESTAMP
DurationField() INTERVAL (PG) / BIGINT µs (SQLite)

auto_now_add rellena al insertar; auto_now reescribe en cada save.

DurationField almacena un datetime.timedelta. En PostgreSQL mapea a INTERVAL nativo (psycopg adapta timedelta directamente). SQLite no tiene tipo intervalo, así que dorm registra un adaptador de sqlite3 que guarda la duración como microsegundos enteros en un BIGINT — el valor Python siempre es un timedelta, la codificación es invisible.

import datetime

class Job(dorm.Model):
    timeout = dorm.DurationField()
    grace = dorm.DurationField(null=True, blank=True)

Job.objects.create(timeout=datetime.timedelta(minutes=5))

Booleanos

BooleanField()BOOLEAN (PG) / INTEGER 0|1 (SQLite). Los defaults se emiten conscientes del vendor (DEFAULT TRUE vs DEFAULT 1).

Enumeraciones

EnumField(enum_cls, max_length=None) almacena un miembro de enum.Enum. El tipo de columna se deriva del tipo subyacente del enum: enums con valores string mapean a VARCHAR(max_length), enums con valores int a INTEGER. La instancia Python siempre lleva el miembro del enum; las lecturas desde BD rehidratan vía enum_cls(value). choices se autopobla para capas de admin / formularios.

import enum

class Status(enum.Enum):
    ACTIVE = "active"
    ARCHIVED = "archived"

class Article(dorm.Model):
    status = dorm.EnumField(Status, default=Status.ACTIVE)

Article.objects.filter(status=Status.ACTIVE)   # miembro
Article.objects.filter(status="active")        # también acepta valor crudo

Texto case-insensitive

CITextField() — columna de texto case-insensitive. Mapea a CITEXT de PostgreSQL (la BD necesita la extensión citext; instálala con RunSQL("CREATE EXTENSION IF NOT EXISTS citext") desde una migración). En SQLite cae a TEXT COLLATE NOCASE para que las comparaciones de igualdad / LIKE se comporten igual sin la extensión.

class User(dorm.Model):
    email = dorm.CITextField(unique=True)

# las dos triunfan y encuentran la misma fila:
User.objects.get(email="Alice@example.com")
User.objects.get(email="alice@example.com")

Datos estructurados

Campo Tipo BD
JSONField() JSONB (PG) / TEXT (SQLite)
BinaryField() BYTEA / BLOB
ArrayField(base_field) <inner>[] (solo PG — falla en SQLite)

Archivos

FileField(upload_to="", *, storage=None, max_length=255) almacena un fichero vía un storage pluggable. La columna en BD es un VARCHAR(max_length) que guarda el nombre del storage (path relativo / clave S3); el valor Python es un wrapper FieldFile que devuelve el descriptor.

class Document(dorm.Model):
    name = dorm.CharField(max_length=100)
    attachment = dorm.FileField(upload_to="docs/%Y/%m/", null=True, blank=True)

doc = Document(name="Informe Q1")
doc.attachment = dorm.ContentFile(b"bytes del PDF", name="q1.pdf")
doc.save()                     # escribe en storage, guarda nombre en BD

doc.attachment.url             # storage.url(name) — path local o URL S3
doc.attachment.size            # storage.size(name)
with doc.attachment.open("rb") as fh:
    payload = fh.read()
doc.attachment.delete()        # borra el fichero + limpia la columna

upload_to acepta:

  • un string estático ("docs/").
  • una plantilla strftime ("docs/%Y/%m/", se expande al guardar).
  • un callable f(instance, filename) -> str para paths totalmente dinámicos — ver Rutas dinámicas abajo.

storage acepta una instancia Storage, un alias resuelto contra settings.STORAGES (por defecto "default"), o None para diferir la búsqueda hasta el primer uso. Sobre null=True mira el bloque de config más abajo — muy recomendado.

Rutas dinámicas

Cuando necesitas calcular la ruta del storage a partir de la instancia — carpetas aisladas por tenant, ruteo por extensión, layouts content-addressed — pasa un callable en lugar de un string. dorm lo invoca como upload_to(instance, filename) al guardar y usa la cadena devuelta como nombre completo del storage.

def upload_owner_scoped(instance, filename):
    """Los uploads de cada usuario viven bajo su propio prefijo para
    que un ACL mal configurado no filtre datos entre cuentas."""
    return f"users/{instance.owner_id}/{filename}"


class Document(dorm.Model):
    owner = dorm.ForeignKey(User, on_delete=dorm.CASCADE)
    attachment = dorm.FileField(upload_to=upload_owner_scoped, null=True)

El callable recibe la instancia completamente poblada, así que cualquier atributo que esté seteado al guardar es libre uso:

import os, hashlib

def upload_by_extension(instance, filename):
    """Rutea los uploads a buckets por mime-type para que las reglas
    de cache del CDN puedan tratar cada tipo distinto."""
    bucket = {".pdf": "documents", ".png": "images", ".jpg": "images"}
    _, ext = os.path.splitext(filename)
    return f"{bucket.get(ext.lower(), 'other')}/{filename}"


def upload_content_addressed(instance, filename):
    """Layout content-addressed — el nombre del storage es el hash de
    la identidad del modelo. Útil para storage que dedupe."""
    digest = hashlib.sha256(
        f"{instance.owner_id}|{filename}".encode()
    ).hexdigest()[:16]
    _, ext = os.path.splitext(filename)
    return f"cas/{digest}{ext}"

Los lambdas también funcionan:

attachment = dorm.FileField(
    upload_to=lambda instance, filename: f"by-name/{instance.slug}/{filename}",
)

Round-trip de migraciones. dorm makemigrations puede serializar un callable a nivel de módulo emitiendo upload_to=upload_owner_scoped más el import correspondiente from yourapp.uploads import upload_owner_scoped en la cabecera de la migración. Los lambdas y funciones anidadas no se pueden round-tripear (no tienen nombre importable estable); el writer deja una marca FIXME en el archivo generado y el usuario lo edita a mano. Así que si el modelo va a pasar por makemigrations alguna vez, declara el callable a nivel de módulo:

# yourapp/uploads.py — module-level, importable.
def upload_owner_scoped(instance, filename):
    return f"users/{instance.owner_id}/{filename}"

# yourapp/models.py
from .uploads import upload_owner_scoped

class Document(dorm.Model):
    attachment = dorm.FileField(upload_to=upload_owner_scoped)

Seguridad de paths. El basename que devuelve el callable pasa por Storage.get_valid_name (elimina separadores de path, normaliza caracteres unsafe), y FileSystemStorage._resolve_path rechaza cualquier ruta final que escape de la raíz del storage. Aunque tu callable empalme accidentalmente un string controlado por el usuario en la parte del directorio, el writer subyacente no puede ser engañado para salirse de location.

Storage backends

La configuración sigue el mismo esquema BACKEND + OPTIONS que DATABASES:

# settings.py — sistema de archivos local (default si STORAGES no se define)
STORAGES = {
    "default": {
        "BACKEND": "dorm.storage.FileSystemStorage",
        "OPTIONS": {
            "location": "/var/app/media",
            "base_url": "/media/",
        },
    }
}

Para usar S3, instala el extra opcional y cambia el backend:

pip install 'djanorm[s3]'
# settings.py — AWS S3 producción
STORAGES = {
    "default": {
        "BACKEND": "dorm.contrib.storage.s3.S3Storage",
        "OPTIONS": {
            "bucket_name": "my-app-uploads",
            "region_name": "eu-west-1",
            # Las credenciales se toman del rol IAM / env vars / `~/.aws/`
            # por defecto — no las hardcodees en código fuente. Las
            # opciones ``access_key`` / ``secret_key`` existen para
            # escenarios de desarrollo (MinIO abajo); en producción
            # déjalas vacías para que boto3 use la cadena de credenciales
            # del entorno.
            "default_acl": "private",
            "querystring_auth": True,     # URLs firmadas
            "querystring_expire": 3600,
        },
    }
}

El mismo S3Storage funciona contra cualquier servicio compatible con S3 — MinIO para desarrollo local, Cloudflare R2, Backblaze B2, DigitalOcean Spaces. Configura endpoint_url y fuerza path-style (la mayoría de endpoints no-AWS no soportan sub-dominios virtual-hosted sobre IP):

# Levanta MinIO localmente — sin cuenta AWS, sin coste.
docker run -d --name minio -p 9000:9000 -p 9001:9001 \
  -e "MINIO_ROOT_USER=minioadmin" \
  -e "MINIO_ROOT_PASSWORD=minioadmin" \
  minio/minio server /data --console-address ":9001"

# Crea el bucket por la consola en http://localhost:9001
# (login: minioadmin / minioadmin) o con `mc`.
# settings.py — desarrollo local contra MinIO.
STORAGES = {
    "default": {
        "BACKEND": "dorm.contrib.storage.s3.S3Storage",
        "OPTIONS": {
            "bucket_name": "dev-uploads",
            "endpoint_url": "http://localhost:9000",
            "access_key": "minioadmin",
            "secret_key": "minioadmin",
            "region_name": "us-east-1",     # MinIO la ignora pero boto3 necesita *algo*
            "signature_version": "s3v4",
            "addressing_style": "path",     # obligatorio: MinIO sobre IP no soporta virtual-hosted
        },
    }
}

El código de la aplicación es idéntico — mismo FileField, mismo obj.attachment.save(...), misma obj.attachment.url. Cambiar entre FileSystemStorage local, MinIO y AWS es puramente un cambio de STORAGES.

Puedes mezclar backends — declara varios alias y eliges por campo:

class Avatar(dorm.Model):
    image = dorm.FileField(upload_to="avatars/", storage="public")
    backup = dorm.FileField(upload_to="archive/", storage="cold")

De serie dorm trae:

Backend Módulo Extra
FileSystemStorage dorm.storage core
S3Storage dorm.contrib.storage.s3 s3 (boto3)

Para enchufar el tuyo (Azure Blob, GCS, encriptado en reposo), hereda de dorm.storage.Storage e implementa _save, _open, delete, exists, size, url. Los métodos async heredan de la clase base (envuelven los sync con asyncio.to_thread); sobrescríbelos si tu SDK es nativamente async.

Consejos

  • Declara siempre null=True, blank=True en campos de archivo opcionales. Un FileField sin set bindea NULL al insertar; una columna no-null rechazaría la fila.
  • MEDIA_URL es una preocupación solo del ORM — dorm no sirve los ficheros. Conecta tu framework (FastAPI StaticFiles, nginx alias, etc.) para exponer location en base_url.
  • default_storage es un proxy a nivel de módulo que se re-resuelve en cada llamada, así que dorm.configure(STORAGES=...) después del import surte efecto al instante.
  • Los ficheros escritos dentro de atomic() se limpian en rollback. FileField.pre_save registra un hook on_rollback que llama a storage.delete(name) si la transacción que lo rodea hace rollback, así que un BusinessRuleViolation a media de bloque no deja bytes huérfanos en disco / S3. Los rollbacks de savepoint limpian solo los ficheros escritos dentro de ese savepoint; el commit exterior (si lo hay) preserva el resto. Fuera de atomic() no se registra cleanup — los saves son fire-and-forget. Ver Transacciones: limpieza al rollback para la API subyacente.
  • Reemplazar un fichero no borra el antiguo. Reasignar obj.attachment = ContentFile(...) y guardar escribe el fichero nuevo pero deja el anterior en storage. Si necesitas semántica delete-on-replace, llama a obj.attachment.delete(save=False) antes de asignar el reemplazo, o programa la limpieza tú mismo vía on_commit.

ImageField

ImageField(upload_to="", *, storage=None, max_length=255) es un FileField especializado que valida que la subida sea realmente una imagen antes de escribirla en storage — así un usuario no puede colar un .exe con la extensión cambiada.

class Avatar(dorm.Model):
    user = dorm.ForeignKey(User, on_delete=dorm.CASCADE)
    image = dorm.ImageField(upload_to="avatars/%Y/%m/", null=True, blank=True)

La validación usa Pillow si está instalada; si no, hace fallback a un sniff de magic-bytes que reconoce PNG / JPEG / GIF / WebP / TIFF / BMP. Instala el extra opcional image para que Pillow sea el validador canónico (y para poder leer dimensiones o re-codificar antes de guardar desde tu código):

pip install 'djanorm[image]'

Todo lo que se puede hacer con FileField (upload_to dinámico, aliases de STORAGES, S3 / MinIO, limpieza atómica al rollback) funciona igual con ImageField — la única diferencia es la comprobación de content-type en el momento de la asignación.

Tipos de rango (solo PostgreSQL)

Campo Tipo BD
IntegerRangeField() int4range
BigIntegerRangeField() int8range
DecimalRangeField() numrange
DateRangeField() daterange
DateTimeRangeField() tstzrange

El tipo de valor Python es dorm.Range(lower, upper, bounds="[)"). bounds son dos caracteres con la inclusividad de los extremos: "[)" (el por defecto), "(]", "[]" o "()". Cualquiera de los dos extremos puede ser None para indicar "sin cota por ese lado".

import datetime

class Reservation(dorm.Model):
    during = dorm.DateTimeRangeField()
    seats = dorm.IntegerRangeField(null=True, blank=True)

Reservation.objects.create(
    during=dorm.Range(
        datetime.datetime(2026, 1, 1, 9, tzinfo=datetime.timezone.utc),
        datetime.datetime(2026, 1, 1, 17, tzinfo=datetime.timezone.utc),
    ),
    seats=dorm.Range(1, 10),
)

PostgreSQL canoniza los rangos discretos (int4range, int8range, daterange) al salir — (1, 5] siempre vuelve como [2, 6). Los rangos continuos (numrange, tstzrange) preservan los bounds escritos. SQLite no tiene tipo de rango nativo; usar uno de estos campos contra una conexión SQLite levanta NotImplementedError desde db_type() para que la limitación aparezca al hacer migrate, no en la primera query.

Relaciones

class Book(dorm.Model):
    title = dorm.CharField(max_length=200)
    # uno-a-muchos
    author = dorm.ForeignKey(
        Author, on_delete=dorm.CASCADE, related_name="books"
    )
    # uno-a-uno
    cover = dorm.OneToOneField(
        "Cover", on_delete=dorm.SET_NULL, null=True
    )

class Article(dorm.Model):
    title = dorm.CharField(max_length=200)
    tags = dorm.ManyToManyField("Tag", related_name="articles")

on_delete acepta CASCADE, PROTECT, SET_NULL, SET_DEFAULT, DO_NOTHING, RESTRICT — semántica idéntica a Django.

El descriptor de FK expone:

  • book.author → la instancia Author relacionada (fetch + caché)
  • book.author_id → el PK entero crudo (tipado como int | None)

Para que el type-checker vea <fk>_id, añade una anotación de clase:

class Book(dorm.Model):
    author = dorm.ForeignKey(Author, ...)
    author_id: int | None        # ← lo verán ty/mypy/pyright

Claves primarias compuestas

CompositePrimaryKey(*field_names) declara que la clave primaria de la tabla abarca más de una columna. Los campos componentes son campos concretos que también declaras en el cuerpo del modelo — el composite solo le indica al writer de migraciones que emita PRIMARY KEY (col1, col2) y al ORM que direccione filas por tupla.

class OrderLine(dorm.Model):
    order_id = dorm.IntegerField()
    line_no = dorm.IntegerField()
    sku = dorm.CharField(max_length=50)
    qty = dorm.IntegerField(default=1)

    pk = dorm.CompositePrimaryKey("order_id", "line_no")

CRUD por tupla pk:

line = OrderLine.objects.create(order_id=1, line_no=1, sku="A", qty=2)
line.pk                             # (1, 1)

OrderLine.objects.get(pk=(1, 1))    # lookup por tupla
OrderLine.objects.filter(pk=(1, 1)) # se descompone en WHERE por componente
line.delete()                       # usa (order_id=…, line_no=…)

Limitaciones a conocer de antemano:

  • Una CompositePrimaryKey no puede ser el destino de una ForeignKey — las FK de una sola columna no pueden referenciar una clave multi-columna. Si necesitas referenciar entre tablas, declara una PK sintética y un UniqueConstraint sobre las columnas compuestas.
  • Ningún componente es auto-incrementable; tú aportas los dos valores en el insert.
  • filter(pk__in=[...]) sobre claves compuestas no está soportado. Usa objetos Q con cláusulas explícitas por componente.

Relaciones genéricas (FKs polimórficas)

Para el caso en que un modelo necesita apuntar a "cualquier otro modelo" — piensa en tags, comentarios, registros de auditoría — usa los helpers de dorm.contrib.contenttypes. Reflejan el django.contrib.contenttypes de Django: un registro ContentType más dos tipos de campo que componen content_type (FK a ContentType) + object_id (columna entera) en una FK polimórfica.

Añade la app a tu settings y ejecuta las migraciones una vez para que exista la tabla django_content_type:

# settings.py
INSTALLED_APPS = ["dorm.contrib.contenttypes", "myapp"]
dorm makemigrations
dorm migrate

Después declara el lado polimórfico y el accesor inverso:

import dorm
from dorm.contrib.contenttypes import (
    ContentType,
    GenericForeignKey,
    GenericRelation,
)

class Article(dorm.Model):
    title = dorm.CharField(max_length=200)
    tags = GenericRelation("Tag")          # accesor inverso — sin columna

class Book(dorm.Model):
    name = dorm.CharField(max_length=200)
    tags = GenericRelation("Tag")

class Tag(dorm.Model):
    label = dorm.CharField(max_length=50)
    content_type = dorm.ForeignKey(ContentType, on_delete=dorm.CASCADE)
    content_type_id: int | None
    object_id = dorm.PositiveIntegerField()
    target = GenericForeignKey("content_type", "object_id")

Acceso forward:

article = Article.objects.create(title="Hello")
tag = Tag(label="featured")
tag.target = article                       # asigna content_type + object_id
tag.save()

reloaded = Tag.objects.get(pk=tag.pk)
isinstance(reloaded.target, Article)       # True

Acceso inverso vía GenericRelation:

article.tags.create(label="urgent")
list(article.tags.all())                   # todos los Tags apuntando a article
article.tags.filter(label__startswith="u").count()

Los caminos asíncronos existen en paralelo a los síncronos:

ct = await ContentType.objects.aget_for_model(Article)
target = await tag.target.aget(tag) if tag.target is None else tag.target

ContentType.objects.get_for_model(MyModel) memoiza la fila por proceso — los lookups polimórficos repetidos no pagan un round-trip por acceso. Si tus tests recrean modelos o truncan la tabla, llama a ContentType.objects.clear_cache() para invalidar.

Cuando iteras una queryset de filas con tags polimórficos, usa prefetch_related("target") — el get(pk=…) por fila del descriptor se reduce a 1 + 1 + K queries (una para los tags, una para todos los ContentType referenciados en bulk, una por cada modelo target concreto). Ver FKs polimórficas en prefetch_related en la guía de consultas.

Opciones comunes de campo

Todo campo acepta:

Opción Efecto
null=True la columna permite NULL (a nivel BD)
blank=True string vacío permitido (validación, no BD)
unique=True añade restricción UNIQUE
db_index=True crea un índice
db_column="x" override del nombre de la columna
default=value o default=callable default a nivel Python (dispara cuando el constructor no recibe valor)
db_default=value o db_default=RawSQL("now()") default a nivel servidor — aparece en CREATE TABLE como DEFAULT <literal>; cubre INSERTs raw que omiten la columna
db_comment="..." (3.1+) comentario de columna almacenado en el field para tooling de documentación. El DDL que lo traduce a COMMENT ON COLUMN (PG / MySQL) llega en 3.2; hoy el valor es accesible vía field.db_comment para inspección / migraciones custom
validators=[fn, ...] se ejecutan al asignar y en full_clean()
choices=[(value, label), …] restringe a un conjunto fijo
editable=False oculto a forms / serializers
help_text="..." string de docs

default vs db_default

import dorm
from dorm.expressions import RawSQL

class Event(dorm.Model):
    # Default Python: dispara cuando ``Event(...)`` se construye sin
    # el kwarg. Dinámico — corre cada vez en la aplicación.
    correlation_id = dorm.UUIDField(default=uuid.uuid4)

    # Default servidor: aparece en DDL como ``DEFAULT now()``. INSERTs
    # raw que omiten la columna (sistema externo escribiendo a la
    # tabla directamente) siguen recibiendo un valor sano.
    created_at = dorm.DateTimeField(db_default=RawSQL("now()"))

    # Ambos a la vez: ``default`` cubre escrituras Python, ``db_default``
    # cubre escrituras SQL raw. Atacan paths distintos, no conflictan.
    revision = dorm.IntegerField(default=1, db_default=1)

RawSQL es la salida de emergencia para defaults específicos del vendor (now(), gen_random_uuid(), llamadas a secuencia). El string se inserta verbatim — usa uno que tu motor reconozca.

Opciones Meta

class Author(dorm.Model):
    ...
    class Meta:
        db_table = "authors"
        ordering = ["name", "-age"]            # orden por defecto
        unique_together = [("first_name", "last_name")]
        indexes = [dorm.Index(fields=["name"], name="author_name_idx")]
        abstract = False                       # True para mixins
        app_label = "blog"                     # rara vez necesario

Index extras (3.1+)

Index(include=[...]) emite un covering index PostgreSQL:

indexes = [
    dorm.Index(
        fields=["email"],
        name="ix_user_email_cover",
        include=["full_name", "is_active"],
    ),
]
# PG: CREATE INDEX ... ON "users" ("email") INCLUDE ("full_name", "is_active")
# SQLite / MySQL: ignoran INCLUDE silenciosamente

Las columnas incluidas viajan con las páginas del índice — el planner resuelve SELECT email, full_name, is_active WHERE email = ? sin tocar el heap.

UniqueConstraint(deferrable=, include=) (3.1+)

constraints = [
    # Unique check diferido — se evalúa en COMMIT, no a fin de
    # statement. Permite swap de dos filas con valor único dentro
    # de una transacción sin disparar la restricción a mitad.
    dorm.UniqueConstraint(
        fields=["slot"], name="uq_slot_deferred",
        deferrable="deferred",
    ),
    # Unique covering — mismo INCLUDE que Index.
    dorm.UniqueConstraint(
        fields=["email"], name="uq_email_cover",
        include=["last_login_at"],
    ),
]

deferrable= acepta "deferred" (check al COMMIT) o "immediate" (check al fin de statement, switchable mid-tx con SET CONSTRAINTS ... DEFERRED). Solo PostgreSQL — SQLite + MySQL descartan el clause silenciosamente.

ExclusionConstraint (3.1+, solo PG)

EXCLUDE PostgreSQL — garantiza que no haya dos filas que satisfagan el mismo operador sobre las expresiones nombradas. Caso canónico: exclusión por solapamiento de rangos:

import dorm

class Reservation(dorm.Model):
    room_id = dorm.IntegerField()
    slot = dorm.RangeField(...)  # tstzrange

    class Meta:
        constraints = [
            dorm.ExclusionConstraint(
                name="no_overlap_room",
                expressions=[("room_id", "="), ("slot", "&&")],
                index_type="gist",   # default
            ),
        ]

Cualquier insert/update que produzca un (room_id, slot) que solape una fila existente lanza IntegrityError. SQLite + MySQL no emiten nada — usa otra estrategia de unicidad ahí.

Clases base abstractas

class TimestampedModel(dorm.Model):
    created_at = dorm.DateTimeField(auto_now_add=True)
    updated_at = dorm.DateTimeField(auto_now=True)

    class Meta:
        abstract = True


class Post(TimestampedModel):                 # hereda timestamps
    title = dorm.CharField(max_length=200)

abstract = True significa: sin tabla en BD, sin migraciones; las subclases concretas heredan los campos como si los hubieran declarado.

Campos custom con descriptores

Una subclase normal de Field escribe el valor directo en el dict de la instancia — Model.__init__ llama a field.to_python(value) y guarda el resultado. Suficiente para el 95% de tipos de columna.

Algunos campos, sin embargo, necesitan reaccionar a la asignación: trackear un upload pendiente, invalidar un cache, capturar el valor anterior. Para eso, sobreescribe __get__ y __set__ y opta-in al class-descriptor path con una línea:

import dorm


class MyEncryptedField(dorm.CharField):
    uses_class_descriptor = True

    def contribute_to_class(self, cls, name):
        # Reinstala como descriptor a nivel de clase — la metaclass
        # eliminaría las instancias de Field de los class attrs si no.
        super().contribute_to_class(cls, name)
        setattr(cls, name, self)

    def __get__(self, instance, owner=None):
        if instance is None:
            return self
        ...

    def __set__(self, instance, value):
        # Lógica propia — encriptado, audit logging, lazy decrypt.
        instance.__dict__[self.attname] = self._encrypt(value)

uses_class_descriptor = True es el opt-in documentado: cuando Model.__init__ ve ese flag (o encuentra el field instalado en la clase directamente), enruta Model(field=value) por setattr para que dispare __set__. FileField es el ejemplo canónico — guarda un File pendiente hasta que model.save() lo flushea al storage.

Tipado

Cada campo es Field[T] (un Generic parametrizado por el tipo Python que almacena). El __get__ sobrecargado del descriptor hace que:

  • Author.nameField[str] (el descriptor en sí, para introspección de migraciones y _meta)
  • author.namestr (el valor real)

Así user.name + " hi" está bien, user.age + " hi" lo flagea el type-checker. Mismo truco que SQLAlchemy 2.0 introdujo con Mapped[T].

Validación

La validación a nivel de campo se ejecuta al asignar/construir:

>>> Author(name="x", age=10, email="not-an-email")
ValidationError: {'email': "'not-an-email' is not a valid email address."}

Para lógica más rica, sobrescribe clean() en el modelo y llama a obj.full_clean() antes de guardar:

class Author(dorm.Model):
    name = dorm.CharField(max_length=100)
    age = dorm.IntegerField()

    def clean(self):
        if self.age < 0:
            raise dorm.ValidationError({"age": "debe ser >= 0"})

full_clean() ejecuta clean_fields()clean()validate_unique().

Señales

from dorm.signals import pre_save, post_save

def slugify(sender, instance, **kwargs):
    if not instance.slug:
        instance.slug = slugify(instance.title)

pre_save.connect(slugify, sender=Article)

Señales disponibles: pre_save, post_save, pre_delete, post_delete, pre_query, post_query. Disparan tanto en operaciones sync como async.

Para la referencia completa — kwargs que recibe cada señal, la diferencia entre sender para save/delete (clase modelo) y sender para señales de query (string del vendor), dispatch_uid para registración idempotente, referencias débiles, y los pitfalls sobre tragado de excepciones y recursión — mira la guía de Señales.