1. Predice constraints, filas y SELECT antes de tocar el código#

Usa tres predicciones separadas. Una respuesta HTTP no puede sustituir ninguna de ellas:

La activación obliga a separar integridad, atomicidad y coste de carga. Las tres viven cerca del ORM, pero fallan por mecanismos distintos.

2. Decide quién garantiza qué#

Una regla puede aparecer en varias capas sin que sean equivalentes:

3. Separa la foreign key persistente de Relationship#

La columna FK guarda identidad y la base puede validarla. Relationship añade navegación y sincronización del grafo Python:

from sqlalchemy import event
from sqlmodel import Field, Relationship, SQLModel, create_engine

class AssetTable(SQLModel, table=True):
    __tablename__ = "assets"

    id: int | None = Field(default=None, primary_key=True)
    name: str
    organization_id: int = Field(
        foreign_key="organizations.id",
        ondelete="RESTRICT",
    )
    organization: "OrganizationTable" = Relationship(
        back_populates="assets",
    )

class OrganizationTable(SQLModel, table=True):
    __tablename__ = "organizations"

    id: int | None = Field(default=None, primary_key=True)
    name: str
    assets: list[AssetTable] = Relationship(
        back_populates="organization",
        passive_deletes="all",
    )

back_populates enlaza ambos atributos: asignar asset.organization puede actualizar organization.assets en memoria. Eso no es un commit y no sustituye organization_id ni su constraint.

SQLite exige una decisión adicional. La documentación oficial mantiene foreign keys desactivadas por compatibilidad y pide habilitarlas en cada conexión:

engine = create_engine(sqlite_url, connect_args={"check_same_thread": False})

@event.listens_for(engine, "connect")
def enable_foreign_keys(dbapi_connection, _connection_record) -> None:
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA foreign_keys=ON")
    cursor.close()

La frontera pública sigue usando modelos separados:

class AssetPublic(SQLModel):
    id: int
    name: str
    organization_id: int

class OrganizationWithAssets(SQLModel):
    id: int
    name: str
    assets: list[AssetPublic]

No expongas el table model con todas sus relaciones: una relación bidireccional puede crear ciclos, overfetching o I/O implícito durante serialización.

4. Declara la invariante en metadata y pruébala en el motor#

Las constraints compuestas pertenecen a la tabla:

from sqlalchemy import CheckConstraint, UniqueConstraint

class TransferEntryTable(SQLModel, table=True):
    __tablename__ = "transfer_entries"
    __table_args__ = (
        UniqueConstraint(
            "transfer_id",
            "location",
            name="uq_transfer_entries_transfer_location",
        ),
        CheckConstraint(
            "delta <> 0",
            name="ck_transfer_entries_nonzero_delta",
        ),
    )

    id: int | None = Field(default=None, primary_key=True)
    transfer_id: int = Field(foreign_key="transfers.id")
    location: str
    delta: int

El nombre estable ayuda a inspeccionar DDL, escribir migraciones y clasificar errores en motores que lo reportan. create_all sigue siendo bootstrap local: añadir esta metadata a una base existente no migra la tabla; ese flujo pertenece a 4.4.

5. La operación observable posee begin → flush → commit o rollback#

La Session comienza una transacción al necesitarla. flush envía DML sin confirmar; commit hace durable todo el trabajo; rollback revierte la transacción y recupera el estado ORM tras un fallo.

request
  │
  ├─ add cabecera
  ├─ flush ──────────────── INSERT; id y constraints visibles
  ├─ add fila 1
  ├─ add fila 2
  ├─ commit ─────────────── todo durable
  │
  └─ si flush/commit falla ─ rollback ─ ninguna fila durable

Un helper interior que hace commit rompe esa figura:

def save_entry(session: Session, entry: TransferEntryTable) -> None:
    session.add(entry)
    session.commit()  # rompe la atomicidad del caso de uso exterior

La alternativa mínima muta o hace flush, y deja la decisión exterior:

def add_entry(session: Session, entry: TransferEntryTable) -> None:
    session.add(entry)

No hace falta inventar una capa universal. Una función de coordinación aparece porque esta regla necesita un solo límite transaccional.

6. Tras IntegrityError, rollback precede a cualquier consulta#

Cuando un flush falla, la transacción DB aborta y SQLAlchemy deja la Session inactiva hasta un rollback explícito. Capturar la excepción no la recupera:

try:
    session.add(asset_tag)
    session.commit()
except IntegrityError as error:
    session.rollback()
    if not is_asset_tag_unique_conflict(error):
        raise
    existing = session.exec(
        select(AssetTagTable).where(AssetTagTable.tag == payload.tag)
    ).one()
    raise HTTPException(
        status_code=409,
        detail={"code": "asset_tag_conflict", "existing_id": existing.id},
    )

is_asset_tag_unique_conflict representa una clasificación probada para el backend activo. No conviertas todo IntegrityError en 409: una FK mal configurada, una tabla ausente o un fallo inesperado necesitan otra respuesta y otra investigación.

7. Usa flush para enlazar filas sin confirmar a medias#

Una cabecera necesita id antes de construir sus filas. flush resuelve esa dependencia sin cerrar la transacción:

def build_transfer(
    session: Session,
    payload: TransferCreate,
) -> TransferTable:
    transfer = TransferTable(reference=payload.reference)
    session.add(transfer)
    session.flush()

    for item in payload.entries:
        session.add(
            TransferEntryTable(
                transfer_id=transfer.id,
                location=item.location,
                delta=item.delta,
            )
        )
    return transfer

try:
    transfer = build_transfer(session, payload)
    session.commit()
except IntegrityError as error:
    session.rollback()
    if not is_transfer_entry_conflict(error):
        raise
    raise HTTPException(
        status_code=409,
        detail={"code": "transfer_entry_conflict"},
    )

session.refresh(transfer)
return transfer

La prueba fuerte induce una posición o ubicación repetida y cuenta cabeceras e hijos desde otra Session. Observar una lista Python vacía después de rollback no basta: puede ser estado expirado o no recargado.

8. Una relación lazy puede convertir un listado en 1+N#

Relationship no dice cuándo llegan los datos. Con la estrategia lazy por defecto, esta secuencia parece inocente:

organizations = session.exec(select(OrganizationTable)).all()
for organization in organizations:
    serialize(organization.assets)

pero su traza puede ser:

SELECT ... FROM organizations;              -- 1
SELECT ... FROM assets WHERE org_id = 1;     -- +1
SELECT ... FROM assets WHERE org_id = 2;     -- +1
SELECT ... FROM assets WHERE org_id = 3;     -- +1

Con N organizaciones, el coste crece como 1 + N. El defecto no es “usar relaciones”; es acceder repetidamente a una relación no cargada dentro de un camino que la respuesta siempre necesita.

Mide statements con un listener en una prueba acotada. Un cronómetro en local mezcla caché, máquina y tamaño de datos; no demuestra la forma causal.

9. Elige la carga desde la forma de respuesta#

Para una página acotada de padres cuya respuesta siempre incluye una colección, selectinload suele encajar:

from sqlalchemy.orm import selectinload

statement = (
    select(OrganizationTable)
    .options(selectinload(OrganizationTable.assets))
    .order_by(OrganizationTable.id)
)
organizations = session.exec(statement).all()

La primera consulta carga organizaciones; otra usa sus claves en IN (...) para agrupar activos. El conjunto principal no se multiplica y el presupuesto deja de depender de N para esta página.

No conviertas selectin en configuración global por comodidad. Otra operación que solo necesita organization_id no debería cargar activos.

10. Diagnostica con DDL, estado transaccional, filas y SQL#

11. Decide lo necesario y detén el alcance#

En esta unidad eliges de forma explícita:

• constraint y nombre para cada invariante compartida;
• nulabilidad y política restrictiva de FK;
• qué función exterior posee commit/rollback;
• qué conflicto conocido merece 409;
• qué relación necesita carga explícita en cada query;
• qué presupuesto de SELECT demuestra la mejora.

La condición de parada excluye many-to-many, cascade delete automático, savepoints y éxito parcial, aislamiento, locks, retries, sesiones async, migraciones y PostgreSQL. Esos mecanismos necesitan contratos y evidencia propios.

12. Comprueba dominio, no solo una suite verde#

Puedes cerrar 4.3 cuando demuestras que:

• distingues columna FK, constraint y atributo Relationship;
• verificas que SQLite aplica foreign keys en cada conexión;
• explicas por qué un SELECT previo no sustituye unique;
• localizas flush, commit y rollback en una línea temporal real;
• una operación fallida deja cero filas parciales desde otra Session;
• la Session se recupera antes de traducir un conflicto conocido;
• reconoces 1+N en una traza y justificas selectinload para esa colección;
• cuerpo, orden y cardinalidad permanecen iguales al cambiar la estrategia.

13. Aísla mecanismos y transfiere sin pistas#

La práctica del Bloque IV publica cuatro intentos:

1. EX-B4-07: recuperar una Session tras una violación unique y devolver 409 estable;
2. EX-B4-10: reducir un listado anidado de 1+N a dos SELECT sin cambiar el body;
3. EX-B4-11: eliminar commits interiores para que una transferencia sea atómica;
4. EX-B4-S01-03: diseñar cabecera y líneas de factura desde criterios visibles, sin pistas ni solución.

Conserva en cada intento la predicción inicial, baseline, fallo visible, cambio, prueba final y límite de parada. Los starters guiados ofrecen apoyo diferente; el checkpoint no prescribe carpetas, helpers ni orden de implementación.

14. Fuentes y vigencia#