SQLAlchemy es uno de los marcos ORM (Object Relational Mapping) más populares en Python, ofreciendo una forma eficiente y flexible de operar bases de datos. Este artículo explica cómo usar SQLAlchemy ORM para manipular datos.

Índice

1. Instalación de SQLAlchemy
2. Conceptos fundamentales
3. Conexión a la base de datos
4. Definición del modelo de datos
5. Creación de tablas
6. Operaciones CRUD básicas
7. Consultas de datos
8. Manejo de relaciones
9. Administración de transacciones
10. Prácticas recomendadas

Instalación

bash

pip install sqlalchemy

Para conectar con bases de datos específicas, instale los controladores correspondientes:

bash

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (ya incluido en la biblioteca estándar de Python)

Conceptos fundamentales

• Motor: El moter de conexión a la base de datos, encargado de la comunicación
• Sesión: La sesión de base de datos, que gestiona todas las operaciones persistentes
• Modelo: Clase del modelo de datos, que corresponde a una tabla en la base de datos
• Consulta: Objeto de consulta, utilizado para construir y ejecutar consultas a la base de datos

Conexión a la base de datos

python

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Crear el motor de conexión a la base de datos
# Ejemplo con SQLite
motor = create_engine('sqlite:///ejemplo.db', echo=True)

# Ejemplo con PostgreSQL
# motor = create_engine('postgresql://usuario:contraseña@localhost:5432/mibase')

# Ejemplo con MySQL
# motor = create_engine('mysql+mysqlconnector://usuario:contraseña@localhost:3306/mibase')

# Crear la fábrica de sesiones
FabricaSesion = sessionmaker(autocommit=False, autoflush=False, bind=motor)

# Crear una instancia de sesión
sesion = FabricaSesion()

Definición del modelo de datos

python

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base

# Crear la clase base
Base = declarative_base()

class Usuario(Base):
    __tablename__ = 'usuarios'
    
    id = Column(Integer, primary_key=True, index=True)
    nombre = Column(String(50), nullable=False)
    correo = Column(String(100), unique=True, index=True)
    
    # Definir relación uno a muchos
    publicaciones = relationship("Publicacion", back_populates="autor")
    
class Publicacion(Base):
    __tablename__ = 'publicaciones'
    
    id = Column(Integer, primary_key=True, index=True)
    titulo = Column(String(100), nullable=False)
    contenido = Column(String(500))
    autor_id = Column(Integer, ForeignKey('usuarios.id'))
    
    # Definir relación muchos a uno
    autor = relationship("Usuario", back_populates="publicaciones")
    
    # Definir relación muchos a muchos (a través de una tabla intermedia)
    etiquetas = relationship("Etiqueta", secondary="publicacion_etiquetas", back_populates="publicaciones")

class Etiqueta(Base):
    __tablename__ = 'etiquetas'
    
    id = Column(Integer, primary_key=True, index=True)
    nombre = Column(String(30), unique=True, nullable=False)
    
    publicaciones = relationship("Publicacion", secondary="publicacion_etiquetas", back_populates="etiquetas")

# Tabla intermedia (para relaciones muchos a muchos)
class PublicacionEtiqueta(Base):
    __tablename__ = 'publicacion_etiquetas'
    
    publicacion_id = Column(Integer, ForeignKey('publicaciones.id'), primary_key=True)
    etiqueta_id = Column(Integer, ForeignKey('etiquetas.id'), primary_key=True)

Creación de tablas en la base de datos

python

# Crear todas las tablas
Base.metadata.create_all(bind=motor)

# Eliminar todas las tablas
# Base.metadata.drop_all(bind=motor)

Operaciones CRUD básicas

Crear datos

python

# Crear un nuevo usuario
nuevo_usuario = Usuario(nombre="Juan", correo="juan@example.com")
sesion.add(nuevo_usuario)
sesion.commit()

# Crear múltiples usuarios
sesion.add_all([
    Usuario(nombre="María", correo="maria@example.com"),
    Usuario(nombre="Pedro", correo="pedro@example.com")
])
sesion.commit()

Leer datos

python

# Obtener todos los usuarios
usuarios = sesion.query(Usuario).all()

# Obtener el primer usuario
primer_usuario = sesion.query(Usuario).first()

# Obtener un usuario por ID
usuario = sesion.query(Usuario).get(1)

Actualizar datos

python

# Consultar y actualizar
usuario = sesion.query(Usuario).get(1)
usuario.nombre = "Juan Pérez"
sesion.commit()

# Actualizar múltiples registros
sesion.query(Usuario).filter(Usuario.nombre.like("Juan%")).update({"nombre": "Apellido Juan"}, synchronize_session=False)
sesion.commit()

Eliminar datos

python

# Consultar y eliminar
usuario = sesion.query(Usuario).get(1)
sesion.delete(usuario)
sesion.commit()

# Eliminar múltiples registros
sesion.query(Usuario).filter(Usuario.nombre == "María").delete(synchronize_session=False)
sesion.commit()

Consultas de datos

Consultas básicas

python

# Obtener todos los registros
usuarios = sesion.query(Usuario).all()

# Obtener campos específicos
nombres = sesion.query(Usuario.nombre).all()

# Ordenar resultados
usuarios = sesion.query(Usuario).order_by(Usuario.nombre.desc()).all()

# Limitar cantidad de resultados
usuarios = sesion.query(Usuario).limit(10).all()

# Paginación con offset
usuarios = sesion.query(Usuario).offset(5).limit(10).all()

Filtros en consultas

python

from sqlalchemy import or_

# Filtro por igualdad
usuario = sesion.query(Usuario).filter(Usuario.nombre == "Juan").first()

# Búsqueda con patrón
usuarios = sesion.query(Usuario).filter(Usuario.nombre.like("Juan%")).all()

# Filtro IN
usuarios = sesion.query(Usuario).filter(Usuario.nombre.in_(["Juan", "María"])).all()

# Múltiples condiciones
usuarios = sesion.query(Usuario).filter(
    Usuario.nombre == "Juan", 
    Usuario.correo.like("%@example.com")
).all()

# Condición OR
usuarios = sesion.query(Usuario).filter(
    or_(Usuario.nombre == "Juan", Usuario.nombre == "María")
).all()

# Condición NOT
usuarios = sesion.query(Usuario).filter(Usuario.nombre != "Juan").all()

Consultas agregadas

python

from sqlalchemy import func

# Contar registros
total = sesion.query(Usuario).count()

# Conteo por grupo
cuenta_publicaciones = sesion.query(
    Usuario.nombre, 
    func.count(Publicacion.id)
).join(Publicacion).group_by(Usuario.nombre).all()

# Promedio y sumas
promedio_id = sesion.query(func.avg(Usuario.id)).scalar()

Consultas con uniones

python

# Unión interna
resultados = sesion.query(Usuario, Publicacion).join(Publicacion).filter(Publicacion.titulo.like("%Python%")).all()

# Unión externa izquierda
resultados = sesion.query(Usuario, Publicacion).outerjoin(Publicacion).all()

# Especificar condición de unión
resultados = sesion.query(Usuario, Publicacion).join(Publicacion, Usuario.id == Publicacion.autor_id).all()

Manejo de relaciones

python

# Crear objetos con relaciones
usuario = Usuario(nombre="Luis", correo="luis@example.com")
publicacion = Publicacion(titulo="Mi primer blog", contenido="¡Hola Mundo!", autor=usuario)
sesion.add(publicacion)
sesion.commit()

# Acceder mediante relaciones
print(f"La publicación '{publicacion.titulo}' fue creada por {publicacion.autor.nombre}")
print(f"Todas las publicaciones de {usuario.nombre}:")
for p in usuario.publicaciones:
    print(f"  - {p.titulo}")

# Operaciones con relaciones muchos a muchos
etiqueta_python = Etiqueta(nombre="Python")
etiqueta_sqlalchemy = Etiqueta(nombre="SQLAlchemy")

publicacion.etiquetas.append(etiqueta_python)
publicacion.etiquetas.append(etiqueta_sqlalchemy)
sesion.commit()

print(f"Etiquetas de la publicación '{publicacion.titulo}':")
for etiqueta in publicacion.etiquetas:
    print(f"  - {etiqueta.nombre}")

Administración de transacciones

python

# Transacción con confirmación automática
try:
    usuario = Usuario(nombre="Usuario de prueba", correo="test@example.com")
    sesion.add(usuario)
    sesion.commit()
except Exception as e:
    sesion.rollback()
    print(f"Error ocurrido: {e}")

# Uso de administrador de contexto para transacciones
from sqlalchemy.orm import Session

def crear_usuario(sesion: Session, nombre: str, correo: str):
    try:
        usuario = Usuario(nombre=nombre, correo=correo)
        sesion.add(usuario)
        sesion.commit()
        return usuario
    except:
        sesion.rollback()
        raise

# Transacciones anidadas
with sesion.begin_nested():
    usuario = Usuario(nombre="Usuario transaccional", correo="transaccion@example.com")
    sesion.add(usuario)

# Puntos de guardado
punto_guardado = sesion.begin_nested()
try:
    usuario = Usuario(nombre="Usuario con punto de guardado", correo="guardado@example.com")
    sesion.add(usuario)
    punto_guardado.commit()
except:
    punto_guardado.rollback()

Prácticas recomendadas

1. Administración de sesiones: Crear una nueva sesión por solicitud y cerrarla tras finalizar
2. Manejo de excepciones: Siempre capturar errores y hacer rollback adecuado
3. Carga diferida: Prestar atención al problema N+1 y optimizar con carga anticpiada
4. Conexiones: Configurar correctamente el tamaño y tiempos de espera del pool
5. Validación de datos: Verificar la integridad de los datos en la capa de modelos o aplicación

python

# Usar administrador de contexto para sesiones
from contextlib import contextmanager

@contextmanager
def obtener_sesion():
    sesion = FabricaSesion()
    try:
        yield sesion
        sesion.commit()
    except Exception:
        sesion.rollback()
        raise
    finally:
        sesion.close()

# Ejemplo de uso
with obtener_sesion() as sesion:
    usuario = Usuario(nombre="Usuario de contexto", correo="contexto@example.com")
    sesion.add(usuario)