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) -> strpara 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:
# 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=Trueen campos de archivo opcionales. UnFileFieldsin set bindeaNULLal insertar; una columna no-null rechazaría la fila. MEDIA_URLes una preocupación solo del ORM — dorm no sirve los ficheros. Conecta tu framework (FastAPIStaticFiles, nginxalias, etc.) para exponerlocationenbase_url.default_storagees un proxy a nivel de módulo que se re-resuelve en cada llamada, así quedorm.configure(STORAGES=...)después del import surte efecto al instante.- Los ficheros escritos dentro de
atomic()se limpian en rollback.FileField.pre_saveregistra un hookon_rollbackque llama astorage.delete(name)si la transacción que lo rodea hace rollback, así que unBusinessRuleViolationa 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 deatomic()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 aobj.attachment.delete(save=False)antes de asignar el reemplazo, o programa la limpieza tú mismo víaon_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):
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 instanciaAuthorrelacionada (fetch + caché)book.author_id→ el PK entero crudo (tipado comoint | 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
CompositePrimaryKeyno puede ser el destino de unaForeignKey— las FK de una sola columna no pueden referenciar una clave multi-columna. Si necesitas referenciar entre tablas, declara una PK sintética y unUniqueConstraintsobre 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 objetosQcon 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:
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.name→Field[str](el descriptor en sí, para introspección de migraciones y_meta)author.name→str(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.