Rutas y Bases de Datos Asíncronas
Combina rutas async def con sesiones asíncronas de SQLAlchemy 2.0 y asyncpg.
Receta
Tarjeta de receta de referencia rápida, lista para copiar y pegar.
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy import text
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db", pool_pre_ping=True)
Session = async_sessionmaker(engine, expire_on_commit=False)
app = FastAPI()
async def get_session():
async with Session() as s:
yield s
@app.get("/ping")
async def ping(session: AsyncSession = Depends(get_session)):
await session.execute(text("SELECT 1"))
return {"db": "ok"}Cuándo usar esto:
- APIs de E/S de alta concurrencia
- PostgreSQL con asyncpg
- Llamadas upstream HTTP no bloqueantes
Ejemplo Funcional
from sqlalchemy import select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import Depends, FastAPI
class Base(DeclarativeBase):
pass
class Item(Base):
__tablename__ = "items"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column()
# asume Session/get_session de la receta
app = FastAPI()
@app.get("/items")
async def list_items(session: AsyncSession = Depends(get_session)) -> list[dict]:
rows = (await session.execute(select(Item).limit(20))).scalars().all()
return [{"id": r.id, "name": r.name} for r in rows]Lo que esto demuestra:
- Consultas ORM asíncronas
- Sesiones por ámbito de solicitud
- Estilo
selectde SQLAlchemy 2.0
Análisis Profundo
Cómo Funciona
- Uvicorn programa corrutinas en un bucle de eventos.
- Los motores asíncronos usan asyncpg para PostgreSQL.
- Mantén las transacciones cortas por solicitud.
Errores Comunes
- Validación de límites omitida: Llegan datos inválidos a las capas de persistencia. Solución: Valida con Pydantic o formularios del framework en el borde.
- Fugas de trazas de pila: Los clientes ven errores internos. Solución: Mapea excepciones a respuestas HTTP estables.
- Bloqueo de bucles de eventos asíncronos: Los trabajadores se detienen bajo carga concurrente. Solución: Usa drivers asíncronos o envoltorios de
threadpool. - Secretos en control de código fuente: Credenciales filtradas a través del historial de git. Solución: Carga secretos desde variables de entorno o una bóveda en tiempo de ejecución.
- Falta de observabilidad: Los incidentes son difíciles de depurar. Solución: Añade logs estructurados, métricas e IDs de solicitud.
Alternativas
| Alternativa | Usar Cuando | No Usar Cuando |
|---|---|---|
| Framework alternativo en este cookbook | Estándar del equipo o monolito existente | API nueva con restricciones diferentes |
| BaaS Gestionado | MVP solo CRUD | Autenticación personalizada, flujos de trabajo o necesidades de cumplimiento |
| gRPC | RPC interno de alto rendimiento | Clientes HTTP públicos y acceso al navegador |
Preguntas Frecuentes
¿Cuándo debo adoptar el acceso a bases de datos asíncronas de FastAPI?
Úsalo cuando los patrones y las compensaciones de esta página coincidan con tu API o el límite de datos.
¿Cuál es el principal error de producción con el acceso a bases de datos asíncronas de FastAPI?
Omitir la validación, los tiempos de espera o los contratos de error explícitos en el borde HTTP.
¿Cómo pruebo el acceso a bases de datos asíncronas de FastAPI?
Usa el cliente de prueba del framework, anula las dependencias y verifica el estado y la forma del JSON.
¿El acceso a bases de datos asíncronas de FastAPI funciona con Python 3.14?
Sí, los ejemplos se dirigen a Python 3.14 con versiones de framework fijadas del pie de página de la pila.
¿Cómo se relaciona el acceso a bases de datos asíncronas de FastAPI con Pydantic 2?
Valida y serializa en los límites; mantén los servicios funcionando con objetos de dominio tipados.
¿Síncrono o asíncrono?
Prefiere rutas asíncronas cuando la E/S domina; mantén el trabajo de CPU pequeño o descárgalo a trabajadores.
¿Dónde debe vivir la lógica de negocio?
Handlers delgados; los servicios poseen las reglas; los repositorios poseen las consultas.
¿Cómo documento las APIs?
Publica documentación OpenAPI o de esquema que coincida con los modelos de respuesta en el código.
¿Cómo manejo el versionado?
Versionado explícito de URL o encabezado con ventanas de deprecación; evita roturas silenciosas.
¿Qué debo leer a continuación?
Sigue los enlaces Relacionados para la siguiente capa de profundidad en esta sección.
¿Cómo me mantengo seguro?
Autentica a los llamadores, autoriza por recurso, limita la tasa y nunca registres secretos.
¿Primer paso para el rendimiento?
Mide la latencia de la base de datos y del upstream antes de cambiar de framework.
Relacionado
- Fundamentos de FastAPI - Rutas y modelos principales
- Inyección de Dependencias - Proveedores por solicitud
- Pruebas de FastAPI - Patrones de cliente de prueba
- Despliegue de FastAPI - Servicio en producción
Versiones de la pila: Esta página fue escrita para Python 3.14.0 (estable 3.14, mantenimiento 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, y uv 0.6+.