Base de datos y asincronía

Aquí está la diferencia real con Django. FastAPI no trae base de datos: hay que elegir ORM, montar la sesión, configurar las migraciones y decidir si todo eso es asíncrono.

Y la asincronía es el argumento técnico por el que se elige FastAPI, así que conviene medir si cumple. Cumple, y solo si se usa bien: hacerlo mal es peor que no hacerlo.

Lo que hay que montar#

pip install "sqlalchemy[asyncio]" aiosqlite alembic

SQLAlchemy 2.0.54 | Alembic 1.20.0

Los modelos, con la sintaxis moderna de SQLAlchemy 2, que usa anotaciones de tipo igual que el resto del bloque:

from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
        pass


class Editorial(Base):
        __tablename__ = 'editoriales'

    id: Mapped[int] = mapped_column(primary_key=True)
    nombre: Mapped[str] = mapped_column(String(100), unique=True)
    pais: Mapped[str] = mapped_column(String(60), default='')

    libros: Mapped[list['Libro']] = relationship(back_populates='editorial')


class Libro(Base):
        __tablename__ = 'libros'

    id: Mapped[int] = mapped_column(primary_key=True)
    titulo: Mapped[str] = mapped_column(String(200), index=True)
    isbn: Mapped[str] = mapped_column(String(17), unique=True)
    precio: Mapped[Decimal]
    editorial_id: Mapped[int] = mapped_column(ForeignKey('editoriales.id'))
    creado: Mapped[datetime] = mapped_column(server_default=func.now())

    editorial: Mapped[Editorial] = relationship(back_populates='libros')

Tres diferencias con Django que se ven en ese código:

El nombre de la tabla es explícito. Django lo deduce; aquí se escribe.

La relación se declara en los dos lados, con el back_populates uniéndolos. Django genera el inverso solo.

Y no hay nombres verbales, así que la ventaja de la lección 14 —etiquetas en español gratis— no existe. Lo que documenta aquí son los modelos de pydantic.

El motor y la fábrica de sesiones:

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

motor = create_async_engine('sqlite+aiosqlite:///./catalogo.db')
Sesion = async_sessionmaker(motor, expire_on_commit=False)

Ese expire_on_commit=False no es un detalle: sin él, después de guardar, leer cualquier atributo del objeto dispara otra consulta, y en asíncrono eso lanza una excepción en lugar de funcionar despacio. Es de las primeras cosas con las que choca todo el mundo.

Las dependencias#

Aquí aparece la pieza central de FastAPI, que hasta ahora no había hecho falta:

async def obtener_sesion() -> AsyncGenerator[AsyncSession]:
        async with Sesion() as sesion:
                yield sesion

@app.get('/api/libros/')
async def listar(sesion: AsyncSession = Depends(obtener_sesion)) -> list[LibroSalida]:
        libros = await sesion.scalars(select(Libro).limit(20))
    return list(libros)

Una dependencia es una función que produce lo que la ruta necesita, y FastAPI la llama antes y limpia después. El yield es la clave: lo de antes se ejecuta al empezar, lo de después al terminar, pase lo que pase.

Lo interesante es que sirve para todo: la sesión de base de datos, el usuario autenticado —que es la lección siguiente—, la paginación, un cliente HTTP compartido. Y se anidan, de modo que la dependencia del usuario puede pedir la de la sesión sin que la ruta se entere.

Para no repetir el tipo largo en cada función:

SesionBD = Annotated[AsyncSession, Depends(obtener_sesion)]


@app.get('/api/libros/')
async def listar(sesion: SesionBD) -> list[LibroSalida]:
        ...

Es el mismo patrón de tipo anotado reutilizable de la lección anterior, y es como se escribe hoy.

De la base de datos al modelo de salida#

El puente son dos líneas:

class LibroSalida(BaseModel):
        model_config = ConfigDict(from_attributes=True)

    id: int
    titulo: str
    precio: Decimal
    editorial: EditorialSalida

Con eso, devolver un objeto de SQLAlchemy desde una ruta anotada funciona: pydantic lee atributos en vez de claves. Y sigue haciendo el recorte de la lección 24, que aquí importa todavía más, porque un modelo de base de datos tiene columnas que no se publican.

Lo que cuesta no cargar las relaciones#

El N+1 otra vez, ahora en SQLAlchemy. Serializando cincuenta libros con su editorial:

sin optimizar (una consulta por libro) : 51 consultas
con selectinload                       :  2 consultas
con joinedload                         :  1 consulta

Las dos formas de arreglarlo:

from sqlalchemy.orm import joinedload, selectinload

select(Libro).options(selectinload(Libro.editorial))   # dos consultas
select(Libro).options(joinedload(Libro.editorial))     # una, con JOIN

Y no son intercambiables: el de unión vale para las relaciones de uno a uno y de muchos a uno; el otro es el que hay que usar con las de uno a muchos, porque una unión multiplica las filas y con un límite devuelve datos incompletos.

La regla práctica: la unión para ir «hacia el padre», el otro para ir «hacia los hijos». Son select_related y prefetch_related de la lección 15 del módulo 4 con otros nombres y la misma distinción.

La asincronía, medida#

Y ahora la razón por la que se elige FastAPI. Diez peticiones simultáneas, cada una esperando una décima de segundo:

async + await asyncio.sleep       0,10 s
async + time.sleep                1,05 s
def normal + time.sleep           0,15 s

ideal concurrente: ~0,10 s   |   secuencial: ~1,00 s

La primera línea es lo que promete la asincronía: diez esperas de 0,1 segundos en 0,1 segundos, porque mientras una espera, las otras avanzan. Eso es lo que no hace un servidor síncrono con un solo proceso.

La tercera es la interesante: una función sin async también sale bien, 0,15 segundos, porque FastAPI detecta que no es asíncrona y la ejecuta en un hilo aparte. Es una decisión de diseño excelente y poco conocida: no te obliga a nada.