SQLAlchemy no es una librería, es dos sistemas distintos que comparten la misma base. SQLAlchemy Core es un constructor de SQL expresivo: escribes queries en Python pero el resultado es SQL real, con control total sobre cada join, índice o función específica de tu base de datos. SQLAlchemy ORM construye encima de Core y añade una capa de mapeo objeto-relacional: defines clases Python, y la librería se encarga de traducir operaciones sobre objetos a SQL y de mantener el estado sincronizado con la base de datos.
El Engine es el punto de entrada compartido entre los dos mundos. Encapsula el pool de conexiones y el dialecto específico de tu base de datos (PostgreSQL, SQLite, MySQL…). El ORM añade un concepto propio: la Session, que actúa como unidad de trabajo. Acumula objetos nuevos o modificados en memoria, y cuando llamas commit(), los persiste todos en una sola transacción coherente. Piénsalo como una cesta de la compra que vacías al llegar a caja.
La pregunta de cuándo usar cada uno tiene respuesta clara: usa ORM cuando tu dominio gira alrededor de objetos con ciclo de vida (usuarios, pedidos, productos) y priorizas velocidad de desarrollo. Usa Core cuando necesitas queries analíticas complejas, bulk inserts masivos, CTEs, window functions, o cuando el ORM estaría generando SQL que ni querrías ver. Mezclarlos en el mismo proyecto es perfectamente válido.
Lo que sí puede romperse con el ORM es el problema N+1 queries: accedes a una lista de objetos y luego, para cada uno, accedes a una relación — sin darte cuenta estás disparando una query por fila. La solución es eager loading, pero hay que configurarlo explícitamente.
# Requiere: pip install sqlalchemy
from sqlalchemy import (
create_engine, Column, Integer, String, ForeignKey, select, text
)
from sqlalchemy.orm import (
DeclarativeBase, relationship, Session, selectinload
)
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
# ── Definición del modelo (ORM) ──────────────────────────────────────────
class Base(DeclarativeBase):
pass
class Department(Base):
__tablename__ = "department"
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
# `lazy="select"` es el default: cada acceso a .employees dispara una query
employees = relationship("Employee", back_populates="department",
lazy="select")
class Employee(Base):
__tablename__ = "employee"
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
age = Column(Integer, nullable=False)
department_id = Column(Integer, ForeignKey("department.id"))
department = relationship("Department", back_populates="employees")
# ── Engine compartido ────────────────────────────────────────────────────
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
# ── ORM: insertar y consultar con Session ────────────────────────────────
def demo_orm() -> None:
with Session(engine) as session:
eng = Department(name="Engineering")
mkt = Department(name="Marketing")
eng.employees = [
Employee(name="Alice", age=31),
Employee(name="Bob", age=24),
]
mkt.employees = [
Employee(name="Carol", age=19),
]
session.add_all([eng, mkt])
session.commit()
# Query tipada: select() sobre la clase, no sobre una tabla raw
stmt = select(Employee).where(Employee.age > 25)
adults = session.scalars(stmt).all()
print("ORM — mayores de 25:", [e.name for e in adults])
# Eager loading explícito para evitar N+1
# selectinload emite UNA query extra con IN (...) en lugar de N queries
stmt_depts = select(Department).options(
selectinload(Department.employees)
)
for dept in session.scalars(stmt_depts):
names = [e.name for e in dept.employees] # sin query adicional
print(f" {dept.name}: {names}")
# ── Core: SQL expresivo con control total ────────────────────────────────
def demo_core() -> None:
emp_table = Employee.__table__ # Core puede reutilizar las tablas del ORM
dep_table = Department.__table__
with engine.connect() as conn:
# Join explícito — aquí decides exactamente qué columnas y cómo
stmt = (
select(emp_table.c.name, emp_table.c.age, dep_table.c.name.label("dept"))
.join(dep_table, emp_table.c.department_id == dep_table.c.id)
.where(emp_table.c.age > 18)
.order_by(emp_table.c.age.desc())
)
rows = conn.execute(stmt).fetchall()
for row in rows:
print(f"Core — {row.name} ({row.age}) → {row.dept}")
# Operación bulk con upsert específica de SQLite: Core es el camino correcto
# El ORM no tiene abstracción directa para INSERT OR REPLACE masivo
bulk_stmt = sqlite_insert(emp_table).values([
{"name": "Dave", "age": 28, "department_id": 1},
{"name": "Eve", "age": 22, "department_id": 2},
]).on_conflict_do_nothing()
conn.execute(bulk_stmt)
conn.commit()
if __name__ == "__main__":
demo_orm()
demo_core()
Qué está pasando realmente
El modelo hereda de DeclarativeBase, que es la forma moderna (SQLAlchemy 2.x) de declarar clases mapeadas. Al hacer Base.metadata.create_all(engine), Core recorre los metadatos y emite los CREATE TABLE correspondientes — el ORM solo mapea, Core ejecuta.
En demo_orm, la Session funciona como contexto: al salir del with, se cierra automáticamente. El session.add_all() no escribe nada todavía; marca los objetos como pending. El commit() serializa todo en una transacción. Si algo falla, un rollback() explícito (o la salida por excepción del bloque) deja la base de datos intacta.
La query select(Employee).where(Employee.age > 25) es ORM puro: trabajas con la clase, y SQLAlchemy sabe a qué tabla corresponde. session.scalars() devuelve directamente instancias de Employee, no tuplas crudas. Puedes acceder a e.name, e.age, etc., como atributos normales de Python.
El fragmento de eager loading es donde está la trampa más común. Si iteras departamentos y accedes a dept.employees con lazy="select" (el default), SQLAlchemy emite una SELECT por cada departamento. Con selectinload, emite exactamente una query adicional usando WHERE department_id IN (1, 2) y carga todas las relaciones de golpe. La diferencia entre 2 queries y 200 queries en producción puede ser la diferencia entre un endpoint que responde en 10 ms y uno que tarda 2 segundos.
En demo_core, usamos Employee.__table__ para acceder directamente al objeto Table subyacente que Core ya construyó cuando procesó el modelo ORM. No hace falta redefinir nada. El select() ahora opera sobre columnas concretas (emp_table.c.name), el join es explícito, y el resultado son filas nombradas — no objetos con ciclo de vida gestionado. Perfecto para reportes o queries de solo lectura.
El sqlite_insert(...).on_conflict_do_nothing() ilustra por qué Core existe: esa construcción es específica del dialecto SQLite. El ORM no tiene un método upsert() portable porque la semántica varía por base de datos. Core te da acceso directo a esas construcciones sin abandonar Python.
Errores que debes conocer
Error: Usar lazy loading por defecto en una ruta que itera una relación, generando N+1 queries sin que el stack trace lo muestre claramente.
# ❌ Wrong
with Session(engine) as session:
depts = session.scalars(select(Department)).all()
for d in depts:
print(d.employees) # query por cada iteración
# ✅ Right
with Session(engine) as session:
stmt = select(Department).options(selectinload(Department.employees))
depts = session.scalars(stmt).all()
for d in depts:
print(d.employees) # ya cargado, sin queries adicionales
selectinload (o joinedload si prefieres un JOIN en la misma query) carga la relación en el momento de la consulta principal, eliminando el problema N+1.
Error: Modificar objetos ORM fuera de una Session activa y esperar que los cambios se persistan.
# ❌ Wrong
with Session(engine) as session:
emp = session.get(Employee, 1)
emp.name = "Updated" # la session ya está cerrada; este cambio se pierde
# No hay commit, el objeto está detached
# ✅ Right
with Session(engine) as session:
emp = session.get(Employee, 1)
emp.name = "Updated" # dentro de la session, el ORM rastrea el cambio
session.commit()
Una Session cerrada deja los objetos en estado detached: SQLAlchemy ya no rastrea sus cambios. Cualquier modificación después del with simplemente no existe para la base de datos.
Error: Ejecutar operaciones Core con engine.connect() sin llamar conn.commit() cuando se necesita escritura.
# ❌ Wrong
with engine.connect() as conn:
conn.execute(text("INSERT INTO department (name) VALUES ('HR')"))
# el bloque termina sin commit → rollback implícito
# ✅ Right
with engine.connect() as conn:
conn.execute(text("INSERT INTO department (name) VALUES ('HR')"))
conn.commit()
En SQLAlchemy 2.x, engine.connect() abre una transacción pero no la confirma automáticamente al salir del contexto. Si quieres autocommit por operación, usa engine.begin() en su lugar, que sí hace commit al salir limpiamente del bloque.
N° 182