1. Predice qué sobrevive antes de cambiar el almacenamiento#

La API modular del bloque anterior puede conservar recursos en una lista o un diccionario. Eso permite aprender contrato y composición, pero el dato pertenece al proceso Python: otra instancia empieza vacía y un reinicio lo pierde.

Haz primero este experimento:

1. crea un recurso y guarda su id;
2. construye otra instancia de la aplicación;
3. solicita ese id desde la nueva instancia;
4. predice la respuesta antes de ejecutar.

Un 200 en el mismo TestClient solo prueba que el estado sigue accesible dentro de ese proceso. La evidencia nueva de esta unidad es más fuerte: dos instancias independientes apuntan al mismo archivo y la segunda recupera la fila creada por la primera.

2. Separa cuatro ciclos de vida#

El engine conoce cómo obtener conexiones para una URL de base de datos. La sesión mantiene objetos y trabajo dentro de un contexto transaccional. Ninguno de los dos es la base de datos, y una sesión tampoco es “la conexión global”.

3. Del registro en memoria a una fila relacional#

SQLite guarda una base relacional en un archivo local. Una tabla define columnas; cada fila representa un registro; una clave primaria identifica una fila. SQL expresa operaciones sobre esas estructuras.

Un ORM añade un mapeo entre clases/objetos Python y tablas/filas. SQLModel combina modelos compatibles con Pydantic y el mapeo de SQLAlchemy. No elimina SQL ni las restricciones del motor: traduce trabajo y permite inspeccionar el SQL resultante.

AssetTable(id=7, name="Router", category="network")
           │
           └── mapeo ORM ──► assets(id, name, category, internal_note)

Esta unidad usa SQLite para aislar el mecanismo. PostgreSQL, despliegue y diferencias entre motores llegan en la Unidad 4.4; no se debe presentar el archivo local como arquitectura de producción universal.

4. Separa tabla, creación y salida#

La misma entidad participa en tres fronteras distintas:

from sqlmodel import Field, SQLModel


class AssetBase(SQLModel):
    name: str = Field(min_length=2, max_length=80)
    category: str = Field(min_length=2, max_length=40)


class AssetCreate(AssetBase):
    pass


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

    id: int | None = Field(default=None, primary_key=True)
    internal_note: str


class AssetPublic(AssetBase):
    id: int

Solo AssetTable lleva table=True: participa en el mapeo y registra una tabla en SQLModel.metadata. AssetCreate valida lo que puede enviar el cliente. AssetPublic expresa lo que se promete devolver.

El id de tabla es int | None porque todavía no existe antes del INSERT; después lo genera la base. En la salida es int porque una representación pública persistida debe tenerlo. internal_note existe en la tabla, pero no en creación ni salida.

Una base sin table=True solo debe contener la intersección real. No fuerces defaults, ids u opciones que cambian de significado entre entrada, tabla y salida para ahorrar unas líneas.

5. Un engine conoce la URL; metadata conoce las tablas#

from sqlmodel import SQLModel, create_engine


sqlite_url = "sqlite:///./assets.db"
engine = create_engine(
    sqlite_url,
    echo=True,
    connect_args={"check_same_thread": False},
)


def create_db_and_tables() -> None:
    SQLModel.metadata.create_all(engine)

sqlite:///./assets.db apunta a un archivo relativo. connect_args permite el patrón síncrono usado por FastAPI con SQLite; no convierte la sesión en segura para compartir. En general, crea un engine estable por aplicación o proceso, no uno por petición.

SQLModel.metadata solo conoce las clases de tabla que Python ya importó. El orden importa:

# app/bootstrap.py
from app import models  # registra AssetTable en metadata
from app.database import engine
from sqlmodel import SQLModel

SQLModel.metadata.create_all(engine)

Si ejecutas create_all antes de importar AssetTable, puede terminar sin error y sin crear esa tabla. Comprueba SQLModel.metadata.tables.keys() y el archivo real antes de culpar al endpoint.

6. La Session es una unidad de trabajo, no un repositorio global#

La sesión mantiene un mapa de identidad y coordina trabajo con una transacción. Observa la secuencia de una creación:

new object
   │ add
   ▼
pending ── flush/commit ──► INSERT and primary key
                                  │ commit confirms
                                  ▼
                         attributes may expire
                                  │ refresh
                                  ▼
                         state read from the DB

• add(obj) registra intención; no promete todavía una fila confirmada.
• flush() envía cambios pendientes dentro de la transacción sin confirmarla. commit() hace flush si hace falta y confirma.
• refresh(obj) vuelve a leer el estado, útil para ids y defaults generados por la base y para serializar evidencia actual.
• rollback() revierte la transacción activa. Después de un fallo de flush, es obligatorio antes de seguir usando esa misma sesión.
• close() libera recursos y termina el contexto; no sustituye la decisión de confirmar o revertir.

En esta primera operación feliz usarás commit y refresh. La política para varias escrituras atómicas, fallos intermedios y rollback observable se estudiará en la Unidad 4.3.

7. Adquiere una sesión por petición y ciérrala con yield#

Recupera el patrón de lifecycle de la Unidad 3.2 y aplica ahora un recurso real:

from collections.abc import Iterator
from typing import Annotated

from fastapi import Depends
from sqlmodel import Session


def get_session() -> Iterator[Session]:
    with Session(engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]

Antes de yield se abre el contexto; la operación recibe la sesión; al salir del with, la sesión se cierra también si la operación falla. La dependencia administra adquisición y cierre. La operación que conoce la intención de negocio decide commit.

request A: Session A ── operation ── close A
request B: Session B ── operation ── close B
engine:     ────────────────────────────────
DB file:    ─────────────────────────────────────────►

No coloques un commit automático después de yield en esta unidad: ocultaría la frontera transaccional a quien lee la operación y mezclaría cleanup con política de escritura.

8. Completa una creación: validar, añadir, confirmar y refrescar#

from fastapi import APIRouter, HTTPException, status

router = APIRouter(prefix="/assets", tags=["assets"])


@router.post("", response_model=AssetPublic, status_code=status.HTTP_201_CREATED)
def create_asset(payload: AssetCreate, session: SessionDep) -> AssetTable:
    db_asset = AssetTable.model_validate(
        payload,
        update={"internal_note": "registered"},
    )
    session.add(db_asset)
    session.commit()
    session.refresh(db_asset)
    return db_asset


@router.get("/{asset_id}", response_model=AssetPublic)
def read_asset(asset_id: int, session: SessionDep) -> AssetTable:
    db_asset = session.get(AssetTable, asset_id)
    if db_asset is None:
        raise HTTPException(status_code=404, detail="Asset not found")
    return db_asset

model_validate cruza el contrato de creación hacia el modelo persistente y añade el dato controlado por servidor. El endpoint devuelve el objeto refrescado, pero AssetPublic filtra la representación.

El GET usa búsqueda por clave primaria solo para demostrar que la fila existe. Consultas, filtros, paginación y PATCH se trabajan en 4.2; añadirlos ahora diluiría la evidencia del lifecycle.

9. Prueba la propiedad que una lista no puede ofrecer#

Una prueba decisiva usa un directorio temporal y la misma ruta de archivo:

def test_asset_survives_app_recreation(tmp_path):
    database_path = tmp_path / "assets.db"

    first_app = create_app(database_path)
    with TestClient(first_app) as first_client:
        created = first_client.post(
            "/assets",
            json={"name": "Router", "category": "network"},
        )
        asset_id = created.json()["id"]

    second_app = create_app(database_path)
    with TestClient(second_app) as second_client:
        recovered = second_client.get(f"/assets/{asset_id}")

    assert recovered.status_code == 200
    assert recovered.json() == {
        "id": asset_id,
        "name": "Router",
        "category": "network",
    }

Esta prueba no exige una estructura concreta de módulos. Sí exige que la fábrica no reemplace el archivo ni conserve la única copia de los datos en memoria. Añade otra observación de lifecycle para demostrar una sesión distinta por petición y cierre en éxito y error.

10. Diagnostica por la primera evidencia que contradice tu modelo#

No empieces creando más capas. Primero identifica si falla metadata, URL, lifecycle, transacción o serialización. Una abstracción adicional no corrige una tabla nunca registrada ni una sesión global.

11. Decisiones y límites de esta unidad#

Esta unidad publica una sola rebanada vertical: SQLite, un modelo plano, un engine estable, bootstrap local con create_all, una sesión por petición y una creación con commit y refresh.

Quedan fuera deliberadamente:

• listados, filtros, paginación y PATCH, que pertenecen a 4.2;
• unicidad, relaciones, varias filas, rollback operativo y transacciones atómicas, que pertenecen a 4.3;
• Alembic, PostgreSQL y evolución del esquema, que pertenecen a 4.4;
• sesiones asíncronas y rendimiento concurrente, que se recuperan en 6.2;
• repositorios, servicios o una arquitectura de carpetas obligatoria.

La condición de parada exige:

• tres modelos con responsabilidades observables;
• table=True solo en el modelo persistente;
• metadata poblada antes del bootstrap;
• un engine estable y una sesión nueva por petición;
• decisión de commit visible en la operación y cierre garantizado;
• id generado por la base disponible después de refresh;
• una segunda instancia que recupera la fila;
• ningún campo interno en body u OpenAPI.

12. Criterios de dominio#

• Explicas por qué un estado en memoria puede pasar tests HTTP y aun así no ser persistente.
• Distingues base, engine, conexión, Session, transacción, objeto Python y fila.
• Predices qué registra table=True y por qué el orden de imports afecta a metadata.
• Justificas id: int | None en tabla e id: int en salida.
• Ordenas y observas add, commit, refresh y close sin tratarlos como sinónimos.
• Explicas cuándo hace falta rollback antes de reutilizar una sesión fallida.
• Demuestras sesión por petición y fila más allá del proceso con pruebas diferentes.
• Conservas el contrato público mientras cambia el mecanismo de almacenamiento.
• Rechazas create_all como sistema de migraciones y detienes el alcance antes de 4.2–4.4.

Prácticas relacionadas#

• EX-B4-02 · Conectar modelo, sesión y contrato público
• EX-B4-S01-01 · Checkpoint: hacer persistente una factura

Fuentes#