Crear una conexión a la base de datos no es barato. El handshake TCP, la autenticación, la negociación de parámetros de sesión: todo eso suma entre 20 y 100 ms dependiendo de la red y el motor. Si cada request HTTP de tu API abre y cierra su propia conexión, estás pagando ese coste repetidamente, y además saturando el servidor de base de datos con conexiones efímeras. El pool de conexiones resuelve esto manteniendo un conjunto de conexiones abiertas y reutilizándolas.
SQLAlchemy gestiona este pool a través de create_engine. Cuando llamas a engine.connect(), no se abre una conexión nueva — se toma una prestada del pool, se usa, y al salir del contexto se devuelve. El parámetro pool_size controla cuántas conexiones permanentes mantiene el pool; max_overflow permite crear conexiones adicionales temporales cuando todas las permanentes están ocupadas. Si superas pool_size + max_overflow, el siguiente intento bloquea hasta que alguna quede libre (o lanza TimeoutError si configuras pool_timeout).
Las transacciones en SQLAlchemy 2.x tienen semántica explícita: engine.connect() abre una transacción implícitamente (BEGIN), pero el commit nunca es automático. Si no llamas a conn.commit() antes de salir del bloque with, se hace rollback. Esto es intencional — te protege de commits accidentales en código que falla a mitad de camino. Los isolation levels controlan qué ve una transacción respecto a cambios concurrentes: READ COMMITTED es el default en PostgreSQL y evita dirty reads; REPEATABLE READ o SERIALIZABLE son necesarios cuando necesitas garantías más fuertes, como en transferencias financieras, pero incrementan la contención.
Alembic es la herramienta de migraciones de esquema de SQLAlchemy. La idea central: tus modelos Python son la fuente de verdad, y Alembic compara esa definición contra el esquema real de la base de datos para generar scripts de migración versionados. Esos scripts se pueden aplicar hacia adelante (upgrade) o revertir (downgrade), lo cual es crítico en despliegues donde necesitas poder hacer rollback.
Si no usas Alembic —o algo equivalente— terminas sincronizando esquemas manualmente, que es exactamente el tipo de proceso que falla silenciosamente en producción cuando alguien olvida un paso.
# requirements: sqlalchemy>=2.0, alembic, psycopg2-binary
from sqlalchemy import (
create_engine, Column, Integer, String, Numeric,
text, event
)
from sqlalchemy.orm import DeclarativeBase, Session
from sqlalchemy.pool import QueuePool
import logging
logging.basicConfig()
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
DATABASE_URL = "postgresql+psycopg2://user:pass@localhost:5432/shop"
# pool_size: conexiones permanentes en el pool
# max_overflow: conexiones extra permitidas bajo demanda
# pool_timeout: segundos de espera antes de TimeoutError
# pool_pre_ping: ejecuta "SELECT 1" antes de entregar una conexión
# para detectar conexiones caídas (firewall timeouts, etc.)
engine = create_engine(
DATABASE_URL,
pool_size=5,
max_overflow=10,
pool_timeout=30,
pool_pre_ping=True,
poolclass=QueuePool,
)
class Base(DeclarativeBase):
pass
class Product(Base):
__tablename__ = "products"
id = Column(Integer, primary_key=True)
name = Column(String(200), nullable=False)
price = Column(Numeric(10, 2), nullable=False)
stock = Column(Integer, nullable=False, default=0)
class Order(Base):
__tablename__ = "orders"
id = Column(Integer, primary_key=True)
product_id = Column(Integer, nullable=False)
quantity = Column(Integer, nullable=False)
def transfer_stock(from_product_id: int, to_product_id: int, qty: int) -> None:
"""
Mueve unidades de stock entre dos productos de forma atómica.
Si cualquier paso falla, todo el bloque se revierte.
"""
# with engine.connect() toma una conexión del pool y abre BEGIN implícito.
# Al salir del bloque sin commit, SQLAlchemy hace ROLLBACK automático.
with engine.connect() as conn:
# REPEATABLE READ garantiza que el stock que leemos no cambia
# entre nuestro SELECT y nuestro UPDATE dentro de la misma transacción.
conn.execution_options(isolation_level="REPEATABLE READ")
source = conn.execute(
text("SELECT stock FROM products WHERE id = :id FOR UPDATE"),
{"id": from_product_id},
).fetchone()
if source is None:
raise ValueError(f"Product {from_product_id} not found")
if source.stock < qty:
raise ValueError(
f"Insufficient stock: have {source.stock}, need {qty}"
)
conn.execute(
text("UPDATE products SET stock = stock - :qty WHERE id = :id"),
{"qty": qty, "id": from_product_id},
)
conn.execute(
text("UPDATE products SET stock = stock + :qty WHERE id = :id"),
{"qty": qty, "id": to_product_id},
)
# Sin este commit, al salir del `with` se ejecuta ROLLBACK.
conn.commit()
def get_product(product_id: int) -> dict | None:
"""
Operación de solo lectura. No necesita commit porque no modifica datos,
pero igual devuelve la conexión al pool al salir del with.
"""
with engine.connect() as conn:
row = conn.execute(
text("SELECT id, name, price, stock FROM products WHERE id = :id"),
{"id": product_id},
).fetchone()
if row is None:
return None
return {"id": row.id, "name": row.name, "price": float(row.price), "stock": row.stock}
# --- Uso con Session (ORM layer) ---
def create_order(product_id: int, quantity: int) -> Order:
"""
Session gestiona su propia transacción. session.commit() es explícito.
session.rollback() ocurre automáticamente si el with-block lanza excepción
cuando usas begin() como context manager.
"""
with Session(engine) as session:
with session.begin(): # BEGIN aquí; COMMIT al salir si no hay excepción
product = session.get(Product, product_id)
if product is None:
raise ValueError(f"Product {product_id} not found")
if product.stock < quantity:
raise ValueError("Not enough stock")
product.stock -= quantity
order = Order(product_id=product_id, quantity=quantity)
session.add(order)
# COMMIT ocurre aquí automáticamente al salir del with session.begin()
return order
Configuración de Alembic
# En la raíz del proyecto pip install alembic alembic init migrations
Esto crea alembic.ini y el directorio migrations/. Edita migrations/env.py para conectar Alembic con tus modelos:
# migrations/env.py (fragmento relevante)
from myapp.models import Base # importa tu Base con los modelos definidos
from myapp.config import DATABASE_URL # tu URL de conexión
config.set_main_option("sqlalchemy.url", DATABASE_URL)
# target_metadata es lo que Alembic compara contra el esquema actual
target_metadata = Base.metadata
# Detecta diferencias entre modelos y esquema → genera un script de migración alembic revision --autogenerate -m "add products and orders tables" # Aplica todas las migraciones pendientes alembic upgrade head # Revierte la última migración alembic downgrade -1 # Ver el historial de migraciones aplicadas alembic history --verbose
El script generado en migrations/versions/ tiene este aspecto:
# migrations/versions/abc123_add_products_and_orders_tables.py
def upgrade() -> None:
op.create_table(
"products",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("name", sa.String(length=200), nullable=False),
sa.Column("price", sa.Numeric(10, 2), nullable=False),
sa.Column("stock", sa.Integer(), nullable=False),
sa.PrimaryKeyConstraint("id"),
)
# ... orders table
def downgrade() -> None:
op.drop_table("orders")
op.drop_table("products")
Nunca edites la base de datos a mano en producción y luego actualices los modelos. Alembic solo puede generar migraciones correctas si el esquema real coincide con lo que él cree que está aplicado.
Desglose del código
pool_pre_ping=True merece especial atención. Los pools mantienen conexiones abiertas indefinidamente, pero los firewalls y proxies como PgBouncer pueden cerrarlas por inactividad sin avisar. Sin pre_ping, la siguiente operación que use esa conexión “zombi” lanzará un error en mitad de una transacción. Con pre_ping, SQLAlchemy hace un SELECT 1 rápido antes de entregar la conexión; si falla, descarta esa conexión y toma otra del pool (o abre una nueva).
La separación entre engine.connect() y Session no es estética. connect() te da acceso directo al nivel de conexión — ideal para queries raw, bulk operations o cuando necesitas control preciso del isolation level por operación. Session añade la capa de identidad del ORM (el mismo objeto Python para la misma fila de base de datos dentro de la sesión) y gestiona el unit-of-work pattern: acumula cambios en memoria y los escribe todos en commit().
El isolation_level="REPEATABLE READ" en transfer_stock no es gratuito. Causa más contención porque PostgreSQL debe mantener versiones del dato durante más tiempo. Úsalo cuando el negocio requiere que dos lecturas de la misma fila dentro de una transacción devuelvan el mismo resultado, lo que importa en operaciones que leen para luego escribir condicionalmente. En operaciones de solo lectura o writes sin dependencia de reads previos, READ COMMITTED es suficiente y más eficiente.
El FOR UPDATE en el SELECT de transfer_stock es el detalle que cierra el círculo: dentro de REPEATABLE READ, dos transacciones podrían ambas leer el stock como 10 unidades y ambas intentar descontar 8. FOR UPDATE serializa ese acceso poniendo un lock de escritura en la fila en el momento del SELECT, no solo del UPDATE.
Errores que debes conocer
Error: Olvidar conn.commit() en operaciones de escritura. SQLAlchemy 2.x no hace autocommit — al salir del bloque with sin commit explícito, se ejecuta rollback silencioso.
# ❌ Incorrecto: los cambios se revierten al salir del with
with engine.connect() as conn:
conn.execute(text("UPDATE products SET stock = 0 WHERE id = 1"))
# ✅ Correcto
with engine.connect() as conn:
conn.execute(text("UPDATE products SET stock = 0 WHERE id = 1"))
conn.commit()
El cambio es una sola línea, pero el efecto es la diferencia entre persistir o perder datos.
Error: Confiar en --autogenerate para detectar todos los cambios. Alembic no detecta automáticamente cambios en constraints de CHECK, cambios en stored procedures, ni algunos cambios de tipo dependiendo del dialecto.
# ❌ Asumes que esto genera la constraint automáticamente
class Product(Base):
__tablename__ = "products"
price = Column(Numeric(10, 2), nullable=False)
# CheckConstraint("price > 0") añadida al modelo
# ✅ Verificas el script generado y lo completas si falta
# En el script de migración generado:
def upgrade() -> None:
op.create_check_constraint(
"ck_products_price_positive",
"products",
"price > 0"
)
Siempre revisa el script generado antes de aplicarlo. --autogenerate es un punto de partida, no una garantía.
Error: Usar el mismo engine en tests y crear las tablas con Base.metadata.create_all(engine) sin Alembic, lo que hace que los tests pasen en un esquema que no coincide con el de producción.
# ❌ Tests crean el esquema directamente desde los modelos
Base.metadata.create_all(engine) # bypasea las migraciones completamente
# ✅ Tests aplican las migraciones de Alembic igual que producción
from alembic.config import Config
from alembic import command
alembic_cfg = Config("alembic.ini")
command.upgrade(alembic_cfg, "head")
Esto garantiza que las migraciones se prueban en el mismo pipeline que el código que las usa, y que un downgrade incompleto se detecta antes de llegar a producción.
N° 183