SQLAlchemy es uno de los frameworks ORM (mapeo objeto-relacional) más populares en Python, proporcionando un método eficiente y flexible para interactuar con bases de datos. Este artículo explorará cómo utilizar el ORM de SQLAlchemy para operaciones con bases de datos.

Tabla de Contenidos

1. Instalación de SQLAlchemy
2. Conceptos Fundamentales
3. Conexión a Bases de Datos
4. Definición de Modelos de Datos
5. Creación de Tablas
6. Operaciones CRUD Básicas
7. Consultas de Datos
8. Operaciones con Relaciones
9. Gestión de Transacciones
10. Mejores Prácticas

Instalación

pip install sqlalchemy

Para conectar con bases de datos específicas, se requieren controladores adicionales:

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

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

Conceptos Fundamentales

• Engine: Motor de conexión a la base de datos, encargado de comunicarse con la misma
• Session: Sesión de base de datos, gestiona todas las operaciones de persistencia
• Model: Clase de modelo de datos, corresponde a una tabla en la base de datos
• Query: Objeto de consulta, utilizado para construir y ejecutar consultas a la base de datos

Conexión a Bases de Datos

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

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

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

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

# Crear fábrica de sesiones
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Crear instancia de sesión
session = SessionFactory()

Definición de Modelos de Datos

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

# Crear clase base
Base = declarative_base()

class Persona(Base):
    __tablename__ = 'personas'
    
    id = Column(Integer, primary_key=True, index=True)
    nombre = Column(String(50), nullable=False)
    email = 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('personas.id'))
    
    # Definir relación muchos a uno
    autor = relationship("Persona", back_populates="publicaciones")
    
    # Definir relación muchos a muchos (mediante 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

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

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

Operaciones CRUD Básicas

Creación de Datos

# Crear nueva persona
nueva_persona = Persona(nombre="Carlos García", email="carlos@example.com")
session.add(nueva_persona)
session.commit()

# Creación múltiple
session.add_all([
    Persona(nombre="María López", email="maria@example.com"),
    Persona(nombre="Pedro Martín", email="pedro@example.com")
])
session.commit()

Lectura de Datos

# Obtener todas las personas
personas = session.query(Persona).all()

# Obtener la primera persona
primera_persona = session.query(Persona).first()

# Obtener persona por ID
persona = session.query(Persona).get(1)

Actualización de Datos

# Consultar y actualizar
persona = session.query(Persona).get(1)
persona.nombre = "Carlos García Pérez"
session.commit()

# Actualización múltiple
session.query(Persona).filter(Persona.nombre.like("Carlos%")).update({"nombre": "Carlos"}, synchronize_session=False)
session.commit()

Eliminación de Datos

# Consultar y eliminar
persona = session.query(Persona).get(1)
session.delete(persona)
session.commit()

# Eliminación múltiple
session.query(Persona).filter(Persona.nombre == "María López").delete(synchronize_session=False)
session.commit()

Consultas de Datos

Consultas Básicas

# Obtener todos los registros
personas = session.query(Persona).all()

# Obtener campos específicos
nombres = session.query(Persona.nombre).all()

# Ordenamiento
personas = session.query(Persona).order_by(Persona.nombre.desc()).all()

# Limitar resultados
personas = session.query(Persona).limit(10).all()

# Desplazamiento
personas = session.query(Persona).offset(5).limit(10).all()

Consultas con Filtros

from sqlalchemy import or_

# Filtro de igualdad
persona = session.query(Persona).filter(Persona.nombre == "Carlos García").first()

# Búsqueda con comodines
personas = session.query(Persona).filter(Persona.nombre.like("Carlos%")).all()

# Consulta IN
personas = session.query(Persona).filter(Persona.nombre.in_(["Carlos García", "María López"])).all()

# Consulta con múltiples condiciones
personas = session.query(Persona).filter(
    Persona.nombre == "Carlos García", 
    Persona.email.like("%@example.com")
).all()

# Condiciones OR
personas = session.query(Persona).filter(
    or_(Persona.nombre == "Carlos García", Persona.nombre == "María López")
).all()

# Diferente de
personas = session.query(Persona).filter(Persona.nombre != "Carlos García").all()

Consultas de Agregación

from sqlalchemy import func

# Conteo
conteo = session.query(Persona).count()

# Conteo por grupo
conteo_publicaciones = session.query(
    Persona.nombre, 
    func.count(Publicacion.id)
).join(Publicacion).group_by(Persona.nombre).all()

# Suma, promedio, etc.
promedio_id = session.query(func.avg(Persona.id)).scalar()

Consultas con Uniones

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

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

# Especificar condiciones de unión
resultados = session.query(Persona, Publicacion).join(Publicacion, Persona.id == Publicacion.autor_id).all()

Operaciones con Relaciones

# Crear objetos con relaciones
persona = Persona(nombre="Ana Torres", email="ana@example.com")
publicacion = Publicacion(titulo="Mi primer artículo", contenido="¡Hola mundo!", autor=persona)
session.add(publicacion)
session.commit()

# Acceder mediante relaciones
print(f"El autor de '{publicacion.titulo}' es {publicacion.autor.nombre}")
print(f"Las publicaciones de {persona.nombre}:")
for p in persona.publicaciones:
    print(f"  - {p.titulo}")

# Operaciones con relaciones muchos a muchos
python_tag = Etiqueta(nombre="Python")
sqlalchemy_tag = Etiqueta(nombre="SQLAlchemy")

publicacion.etiquetas.append(python_tag)
publicacion.etiquetas.append(sqlalchemy_tag)
session.commit()

print(f"Etiquetas del artículo '{publicacion.titulo}':")
for etiqueta in publicacion.etiquetas:
    print(f"  - {etiqueta.nombre}")

Gestión de Transacciones

# Transacción con commit automático
try:
    persona = Persona(nombre="Usuario de prueba", email="test@example.com")
    session.add(persona)
    session.commit()
except Exception as e:
    session.rollback()
    print(f"Ocurrió un error: {e}")

# Usando gestor de contexto para transacciones
from sqlalchemy.orm import Session

def crear_persona(session: Session, nombre: str, email: str):
    try:
        persona = Persona(nombre=nombre, email=email)
        session.add(persona)
        session.commit()
        return persona
    except:
        session.rollback()
        raise

# Transacciones anidadas
with session.begin_nested():
    persona = Persona(nombre="Usuario transaccional", email="transaction@example.com")
    session.add(persona)

# Puntos de guardado
punto_guardado = session.begin_nested()
try:
    persona = Persona(nombre="Usuario con punto de guardado", email="savepoint@example.com")
    session.add(persona)
    punto_guardado.commit()
except:
    punto_guardado.rollback()

Mejores Prácticas

1. Gestión de Sesiones: Crear una nueva sesión para cada solicitud y cerrarla al finalizar
2. Manejo de Excepciones: Siempre manejar excepciones y hacer rollback de transacciones apropiadamente
3. Carga Diferida: Prestar atención al problema N+1 en consultas, usar carga anticipada para optimizar
4. Pool de Conexiones: Configurar adecuadamente el tamaño del pool y tiempos de espera
5. Validación de Datos: Verificar integridad de datos en la capa de modelos o aplicación

# Usar gestor de contexto para manejar sesiones
from contextlib import contextmanager

@contextmanager
def obtener_db():
    db = SessionFactory()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

# Ejemplo de uso
with obtener_db() as db:
    persona = Persona(nombre="Usuario contexto", email="context@example.com")
    db.add(persona)

Conclusión

El ORM de SQLAlchemy proporciona un método poderoso y flexible para operar con bases de datos. A través de esta introducción, debería ser capaz de:

1. Instalar y configurar SQLAlchemy
2. Definir modelos de datos y relaciones
3. Realizar operaciones CRUD básicas
4. Construir consultas complejas
5. Gestionar transacciones de base de datos
6. Seguir las mejores prácticas

SQLAlchemy tiene muchas características avanzadas, como propiedades híbridas, escucha de eventos, consultas personalizadas, entre otras, que vale la pena explorar y aprender.