# SQLAlchemy ORM en Python: bases de datos relacionales paso a paso
**SQLAlchemy** es la biblioteca de base de datos más popular en Python. Ofrece dos APIs: **Core** (SQL de bajo nivel) y **ORM** (mapeo objeto-relacional). Este tutorial cubre el ORM moderno (SQLAlchemy 2.x).
## 1. Instalación y configuración
```bash
pip install sqlalchemy # Core + ORM
pip install alembic # Migraciones
pip install psycopg2-binary # Para PostgreSQL
# Para MySQL: pip install pymysql
```
```python
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session
# SQLite (desarrollo)
engine = create_engine("sqlite:///mi_app.db", echo=True)
# PostgreSQL (producción)
# engine = create_engine("postgresql+psycopg2://usuario:password@localhost/midb")
# MySQL
# engine = create_engine("mysql+pymysql://usuario:password@localhost/midb")
class Base(DeclarativeBase):
pass
```
## 2. Definir modelos
```python
from sqlalchemy import String, Integer, Float, Boolean, DateTime, ForeignKey, Text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from datetime import datetime
from typing import Optional, List
class Base(DeclarativeBase):
pass
class Usuario(Base):
__tablename__ = "usuarios"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
nombre: Mapped[str] = mapped_column(String(100), nullable=False)
email: Mapped[str] = mapped_column(String(200), unique=True, nullable=False)
activo: Mapped[bool] = mapped_column(Boolean, default=True)
creado_en: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
# Relación uno-a-muchos: un usuario tiene muchos pedidos
pedidos: Mapped[List["Pedido"]] = relationship("Pedido", back_populates="usuario")
def __repr__(self) -> str:
return f"Usuario(id={self.id}, nombre={self.nombre!r})"
class Pedido(Base):
__tablename__ = "pedidos"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
usuario_id: Mapped[int] = mapped_column(ForeignKey("usuarios.id"))
total: Mapped[float] = mapped_column(Float, nullable=False)
estado: Mapped[str] = mapped_column(String(50), default="pendiente")
creado_en: Mapped[datetime] = mapped_column(DateTime, default=datetime.now)
notas: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
# Relación inversa
usuario: Mapped["Usuario"] = relationship("Usuario", back_populates="pedidos")
lineas: Mapped[List["LineaPedido"]] = relationship("LineaPedido", back_populates="pedido")
class LineaPedido(Base):
__tablename__ = "lineas_pedido"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
pedido_id: Mapped[int] = mapped_column(ForeignKey("pedidos.id"))
producto: Mapped[str] = mapped_column(String(200))
cantidad: Mapped[int] = mapped_column(Integer)
precio_unit: Mapped[float] = mapped_column(Float)
pedido: Mapped["Pedido"] = relationship("Pedido", back_populates="lineas")
# Crear todas las tablas
Base.metadata.create_all(engine)
```
## 3. CRUD básico con Session
```python
from sqlalchemy.orm import Session
# CREATE — Insertar registros
with Session(engine) as session:
usuario1 = Usuario(nombre="Ana García", email="ana@ejemplo.com")
usuario2 = Usuario(nombre="Carlos López", email="carlos@ejemplo.com")
session.add_all([usuario1, usuario2])
session.commit()
print(f"Usuario creado con id={usuario1.id}")
# READ — Leer registros
with Session(engine) as session:
# Por clave primaria
usuario = session.get(Usuario, 1)
print(usuario)
# Todos los registros
usuarios = session.query(Usuario).all()
for u in usuarios:
print(f"{u.id}: {u.nombre} ({u.email})")
# UPDATE
with Session(engine) as session:
usuario = session.get(Usuario, 1)
usuario.nombre = "Ana Martínez"
session.commit()
# DELETE
with Session(engine) as session:
usuario = session.get(Usuario, 2)
session.delete(usuario)
session.commit()
```
## 4. Consultas avanzadas
```python
from sqlalchemy import select, func, and_, or_, desc
from sqlalchemy.orm import Session
with Session(engine) as session:
# SELECT con WHERE
stmt = select(Usuario).where(Usuario.activo == True)
usuarios_activos = session.scalars(stmt).all()
# Múltiples condiciones
stmt = select(Usuario).where(
and_(
Usuario.activo == True,
Usuario.nombre.like("Ana%"),
)
)
# OR
stmt = select(Usuario).where(
or_(Usuario.email.endswith("@gmail.com"), Usuario.activo == False)
)
# Ordenar y paginar
stmt = (
select(Usuario)
.where(Usuario.activo == True)
.order_by(desc(Usuario.creado_en))
.limit(10)
.offset(20)
)
pagina = session.scalars(stmt).all()
# Contar
total = session.scalar(select(func.count(Usuario.id)))
print(f"Total usuarios: {total}")
# Agregaciones
stmt = select(
func.count(Pedido.id).label("total_pedidos"),
func.sum(Pedido.total).label("suma_total"),
func.avg(Pedido.total).label("media_total"),
)
resultado = session.execute(stmt).one()
print(f"Pedidos: {resultado.total_pedidos}, Suma: {resultado.suma_total:.2f}")
```
## 5. Joins y relaciones
```python
from sqlalchemy import select, join
from sqlalchemy.orm import Session, joinedload, selectinload
with Session(engine) as session:
# JOIN implícito (carga eager)
stmt = (
select(Usuario)
.options(selectinload(Usuario.pedidos)) # Carga pedidos en la misma consulta
.where(Usuario.activo == True)
)
usuarios = session.scalars(stmt).all()
for u in usuarios:
print(f"{u.nombre}: {len(u.pedidos)} pedidos")
# JOIN explícito
stmt = (
select(Usuario, Pedido)
.join(Pedido, Pedido.usuario_id == Usuario.id)
.where(Pedido.estado == "pendiente")
)
for usuario, pedido in session.execute(stmt):
print(f"{usuario.nombre} → Pedido #{pedido.id}: {pedido.total:.2f}€")
# Subconsulta: usuarios sin pedidos
subq = select(Pedido.usuario_id).distinct()
stmt = select(Usuario).where(Usuario.id.not_in(subq))
sin_pedidos = session.scalars(stmt).all()
```
## 6. Relación muchos-a-muchos
```python
from sqlalchemy import Table, Column, Integer, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
# Tabla de asociación
etiqueta_producto = Table(
"etiqueta_producto",
Base.metadata,
Column("etiqueta_id", Integer, ForeignKey("etiquetas.id"), primary_key=True),
Column("producto_id", Integer, ForeignKey("productos.id"), primary_key=True),
)
class Etiqueta(Base):
__tablename__ = "etiquetas"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
nombre: Mapped[str] = mapped_column(String(50), unique=True)
productos: Mapped[list["Producto"]] = relationship(
"Producto", secondary=etiqueta_producto, back_populates="etiquetas"
)
class Producto(Base):
__tablename__ = "productos"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
nombre: Mapped[str] = mapped_column(String(200))
precio: Mapped[float] = mapped_column(Float)
etiquetas: Mapped[list["Etiqueta"]] = relationship(
"Etiqueta", secondary=etiqueta_producto, back_populates="productos"
)
# Uso
with Session(engine) as session:
python_tag = Etiqueta(nombre="python")
web_tag = Etiqueta(nombre="web")
prod = Producto(nombre="Libro FastAPI", precio=39.99, etiquetas=[python_tag, web_tag])
session.add(prod)
session.commit()
```
## 7. Patrón Repository
```python
from sqlalchemy.orm import Session
from sqlalchemy import select
from typing import Optional, List
class RepositorioUsuario:
def __init__(self, session: Session):
self.session = session
def crear(self, nombre: str, email: str) -> Usuario:
usuario = Usuario(nombre=nombre, email=email)
self.session.add(usuario)
self.session.flush() # Obtiene el id sin hacer commit
return usuario
def obtener_por_id(self, id: int) -> Optional[Usuario]:
return self.session.get(Usuario, id)
def obtener_por_email(self, email: str) -> Optional[Usuario]:
stmt = select(Usuario).where(Usuario.email == email)
return self.session.scalar(stmt)
def listar_activos(self, limit: int = 100) -> List[Usuario]:
stmt = select(Usuario).where(Usuario.activo == True).limit(limit)
return list(self.session.scalars(stmt))
def desactivar(self, id: int) -> bool:
usuario = self.obtener_por_id(id)
if usuario:
usuario.activo = False
return True
return False
# Uso con context manager
with Session(engine) as session:
repo = RepositorioUsuario(session)
nuevo = repo.crear("María", "maria@ejemplo.com")
session.commit()
print(f"Creado: {nuevo}")
```
## 8. Migraciones con Alembic
```bash
# Inicializar Alembic
alembic init alembic
# Editar alembic/env.py:
# from mi_app.models import Base
# target_metadata = Base.metadata
# Generar migración automática
alembic revision --autogenerate -m "crear tablas iniciales"
# Aplicar migraciones
alembic upgrade head
# Ver historial
alembic history
# Revertir última migración
alembic downgrade -1
```
## Resumen
| Característica | SQLAlchemy ORM |
|----------------|---------------|
| ORM moderno | `Mapped[tipo]` + `mapped_column()` |
| Sesión | `Session(engine)` como context manager |
| Consultas | `select(Modelo).where(condición)` |
| Relaciones | `relationship()` con `back_populates` |
| Migraciones | Alembic (`alembic revision --autogenerate`) |
| Async | `AsyncSession` + `create_async_engine` |
Guía