SQLAlchemy 2.0 en Python: engine, Session, modelos ORM y queries con select()

SQLAlchemy es el toolkit de base de datos más completo de Python. Con la versión 2.0 (2023) se rediseñó la API para ser más consistente: una única forma de hacer queries, mejor integración con los type checkers y soporte nativo para código asíncrono. La curva de aprendizaje es real, pero la inversión merece la pena cuando el proyecto crece.

Conectar y definir modelos con DeclarativeBase

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

# Crear el engine (conexión a la BD)
engine = create_engine("sqlite:///mi_base.db", echo=True)
# echo=True imprime el SQL generado; útil en desarrollo

class Base(DeclarativeBase):
    pass

class Categoria(Base):
    __tablename__ = "categorias"

    id: Mapped[int] = mapped_column(primary_key=True)
    nombre: Mapped[str] = mapped_column(String(100))
    productos: Mapped[list["Producto"]] = relationship(back_populates="categoria")

class Producto(Base):
    __tablename__ = "productos"

    id: Mapped[int] = mapped_column(primary_key=True)
    nombre: Mapped[str] = mapped_column(String(200))
    precio: Mapped[float]
    stock: Mapped[int] = mapped_column(default=0)
    categoria_id: Mapped[int] = mapped_column(ForeignKey("categorias.id"))
    categoria: Mapped[Categoria] = relationship(back_populates="productos")

# Crear todas las tablas
Base.metadata.create_all(engine)

CRUD con Session

from sqlalchemy.orm import Session

# CREATE
with Session(engine) as session:
    cat = Categoria(nombre="Electrónica")
    session.add(cat)
    session.flush()  # obtener el ID sin commit

    prod1 = Producto(nombre="Teclado", precio=49.99, stock=10, categoria_id=cat.id)
    prod2 = Producto(nombre="Ratón", precio=29.99, stock=25, categoria_id=cat.id)
    session.add_all([prod1, prod2])
    session.commit()
    print(f"Categoría creada con id={cat.id}")

# READ
from sqlalchemy import select

with Session(engine) as session:
    stmt = select(Producto).where(Producto.precio < 50)
    productos = session.scalars(stmt).all()
    for p in productos:
        print(f"{p.nombre}: {p.precio:.2f} €")

# UPDATE
with Session(engine) as session:
    stmt = select(Producto).where(Producto.nombre == "Teclado")
    teclado = session.scalars(stmt).one()
    teclado.precio = 44.99
    session.commit()

# DELETE
with Session(engine) as session:
    stmt = select(Producto).where(Producto.nombre == "Ratón")
    raton = session.scalars(stmt).one_or_none()
    if raton:
        session.delete(raton)
        session.commit()

Queries avanzadas con select()

from sqlalchemy import select, func, and_, or_

with Session(engine) as session:
    # Filtros compuestos
    stmt = select(Producto).where(
        and_(Producto.precio > 20, Producto.stock > 0)
    )

    # ORDER BY y LIMIT
    stmt = (
        select(Producto)
        .order_by(Producto.precio.desc())
        .limit(5)
    )

    # Joins
    stmt = (
        select(Producto, Categoria.nombre)
        .join(Categoria)
        .where(Categoria.nombre == "Electrónica")
    )

    # Agregaciones
    stmt = select(func.count(Producto.id), func.avg(Producto.precio))
    count, avg = session.execute(stmt).one()
    print(f"{count} productos, precio medio: {avg:.2f} €")

    # Subquery
    subq = select(func.avg(Producto.precio)).scalar_subquery()
    stmt = select(Producto).where(Producto.precio > subq)
    caros = session.scalars(stmt).all()

Cargar relaciones con selectinload

from sqlalchemy.orm import selectinload

with Session(engine) as session:
    # Sin selectinload: N+1 queries (lazy loading por defecto)
    categorias = session.scalars(select(Categoria)).all()
    for cat in categorias:
        print(cat.productos)  # query extra por cada categoría

    # Con selectinload: 2 queries en total
    stmt = select(Categoria).options(selectinload(Categoria.productos))
    categorias = session.scalars(stmt).all()
    for cat in categorias:
        print(f"{cat.nombre}: {len(cat.productos)} productos")  # sin queries extra

AsyncSession para FastAPI

# pip install sqlalchemy[asyncio] aiosqlite

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

async_engine = create_async_engine("sqlite+aiosqlite:///async_base.db")

AsyncSessionLocal = sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

# En FastAPI:
from fastapi import FastAPI, Depends
from sqlalchemy import select

app = FastAPI()

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/productos")
async def listar_productos(db: AsyncSession = Depends(get_db)):
    stmt = select(Producto).order_by(Producto.nombre)
    result = await db.execute(stmt)
    return result.scalars().all()

Migraciones con Alembic

# pip install alembic
# alembic init alembic   — crea la estructura de migraciones

# alembic/env.py — conectar con los modelos
from mi_app.models import Base
target_metadata = Base.metadata

# Generar migración automática tras cambiar los modelos:
# alembic revision --autogenerate -m "añadir campo activo"

# Aplicar migraciones pendientes:
# alembic upgrade head

# Revertir la última migración:
# alembic downgrade -1

# Ver historial de migraciones:
# alembic history --verbose

Las claves de SQLAlchemy 2.0 que marcan la diferencia: Mapped[T] y mapped_column() en lugar de Column() dan tipado estático real; session.scalars(stmt) devuelve objetos directamente en lugar de Row tuples; y selectinload() es la forma recomendada de evitar el problema N+1. Para un proyecto FastAPI con PostgreSQL, la combinación SQLAlchemy 2.0 + asyncpg + Alembic cubre todo el stack de base de datos.