Instalación y Configuración Inicial

Para comenzar con SQLAlchemy ORM en Python, primero instale la biblioteca mediante pip. Además, si utiliza bases de datos específicas como PostgreSQL o MySQL, se requieren controladores adicionales.

pip install sqlalchemy
# Para PostgreSQL, instalar psycopg2-binary
# Para MySQL, instalar mysql-connector-python
# SQLite está incluido en la biblioteca estándar de Python

Conceptos Fundamentales

SQLAlchemy ORM se basa en componentes clave como el motor de conexión (Engine), las sesiones de base de datos (Session), los modelos de datos (Model) y los objetos de consulta (Query). Estos facilitan la interacción con la base de datos de manera orientada a objetos.

Estableciendo Conexiones

Se configura un motor para la comunicación con la base de datos y una fábrica de sesiones para gestionar las operaciones. El siguiente ejemplo utiliza SQLite, pero se puede adaptar a otros sistemas de gestión de bases de datos.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Motor de conexión para SQLite
motor_db = create_engine('sqlite:///mi_base_de_datos.db', echo=False)

# Para PostgreSQL: create_engine('postgresql://usuario:contraseña@localhost:5432/mi_db')
# Para MySQL: create_engine('mysql+mysqlconnector://usuario:contraseña@localhost:3306/mi_db')

# Fábrica de sesiones
FabricaSesion = sessionmaker(autocommit=False, autoflush=False, bind=motor_db)

# Crear sesión
sesion = FabricaSesion()

Definición de Modelos de Datos

Los modelos representan las tablas de la base de datos. Se definen clases que heredan de una base declarativa, con columnas y relaciones. A continuación, se muestra un ejemplo con entidades de usuario, publicación y etiqueta.

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

BaseDeclarativa = declarative_base()

class Usuario(BaseDeclarativa):
    __tablename__ = 'usuarios'
    
    identificador = Column(Integer, primary_key=True, index=True)
    nombre = Column(String(50), nullable=False)
    correo = Column(String(100), unique=True, index=True)
    
    publicaciones = relationship("Publicacion", back_populates="autor")

class Publicacion(BaseDeclarativa):
    __tablename__ = 'publicaciones'
    
    identificador = Column(Integer, primary_key=True, index=True)
    titulo = Column(String(100), nullable=False)
    contenido = Column(String(500))
    id_autor = Column(Integer, ForeignKey('usuarios.identificador'))
    
    autor = relationship("Usuario", back_populates="publicaciones")
    etiquetas = relationship("Etiqueta", secondary="publicacion_etiqueta", back_populates="publicaciones")

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

class PublicacionEtiqueta(BaseDeclarativa):
    __tablename__ = 'publicacion_etiqueta'
    
    id_publicacion = Column(Integer, ForeignKey('publicaciones.identificador'), primary_key=True)
    id_etiqueta = Column(Integer, ForeignKey('etiquetas.identificador'), primary_key=True)

Creación de Estructuras de Base de Datos

Para generar las tablas en la base de datos según los modelos definidos, se utiliza el metadato de la base declarativa.

# Crear todas las tablas
BaseDeclarativa.metadata.create_all(bind=motor_db)

# Eliminar todas las tablas (usar con precaución)
# BaseDeclarativa.metadata.drop_all(bind=motor_db)

Operaciones CRUD Básicas

Inserción de Datos

Se pueden añadir registros individuales o en lote a la sesión y confirmarlos.

# Crear un usuario
nuevo_usuario = Usuario(nombre="Ana", correo="ana@ejemplo.com")
sesion.add(nuevo_usuario)
sesion.commit()

# Inserción múltiple
sesion.add_all([
    Usuario(nombre="Carlos", correo="carlos@ejemplo.com"),
    Usuario(nombre="Elena", correo="elena@ejemplo.com")
])
sesion.commit()

Lectura de Datos

Se recuperan registros mediante consultas, ya sea todos, el primero o por identificador.

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

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

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

Actualización de Datos

Los cambios en los objetos se persisten al confirmar la sesión. También se pueden realizar actualizaciones masivas.

# Actualizar un usuario
usuario_a_modificar = sesion.query(Usuario).get(1)
usuario_a_modificar.nombre = "Ana Modificada"
sesion.commit()

# Actualización masiva
sesion.query(Usuario).filter(Usuario.nombre.like("Ana%")).update({"nombre": "Ana Actualizada"}, synchronize_session=False)
sesion.commit()

Eliminación de Datos

Los registros se eliminan individualmente o en bloque.

# Eliminar un usuario
usuario_a_eliminar = sesion.query(Usuario).get(1)
sesion.delete(usuario_a_eliminar)
sesion.commit()

# Eliminación masiva
sesion.query(Usuario).filter(Usuario.nombre == "Carlos").delete(synchronize_session=False)
sesion.commit()

Consultas Avanzadas

Filtrado y Ordenación

Se pueden aplicar filtros, ordenar resultados y limitar la cantidad de registros devueltos.

# Filtrado por condición
usuarios_filtrados = sesion.query(Usuario).filter(Usuario.nombre == "Ana").all()

# Filtrado con LIKE
usuarios_patron = sesion.query(Usuario).filter(Usuario.nombre.like("%li%")).all()

# Ordenación descendente
usuarios_ordenados = sesion.query(Usuario).order_by(Usuario.nombre.desc()).all()

# Limitar resultados
usuarios_limitados = sesion.query(Usuario).limit(5).offset(2).all()

Consultas con Operadores Lógicos

Se combinan condiciones usando operadores como OR.

from sqlalchemy import or_

# Condición OR
usuarios_or = sesion.query(Usuario).filter(
    or_(Usuario.nombre == "Ana", Usuario.nombre == "Elena")
).all()

Agregación y Agrupación

Funciones de agregación permiten calcular conteos, sumas o promedios.

from sqlalchemy import func

# Contar usuarios
total_usuarios = sesion.query(Usuario).count()

# Agrupar y contar publicaciones por usuario
conteo_publicaciones = sesion.query(
    Usuario.nombre,
    func.count(Publicacion.identificador)
).join(Publicacion).group_by(Usuario.nombre).all()

Consultas con Uniones

Se realizan uniones entre tablas para combinar datos relacionados.

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

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

Manejo de Relaciones

Las relaciones entre entidades se facilitan mediante artibutos de relación. Se pueden crear objetos relacionados y acceder a ellos directamente.

# Crear usuario y publicación con relación
usuario_rel = Usuario(nombre="Roberto", correo="roberto@ejemplo.com")
publicacion_rel = Publicacion(titulo="Introducción a ORM", contenido="Contenido de ejemplo", autor=usuario_rel)
sesion.add(publicacion_rel)
sesion.commit()

# Acceder a la relación
print(f"Autor: {publicacion_rel.autor.nombre}")
for pub in usuario_rel.publicaciones:
    print(f"Publicación: {pub.titulo}")

# Relación muchos a muchos
etiqueta_python = Etiqueta(nombre="Python Avanzado")
publicacion_rel.etiquetas.append(etiqueta_python)
sesion.commit()

for etiqueta in publicacion_rel.etiquetas:
    print(f"Etiqueta: {etiqueta.nombre}")

Gestión de Transacciones

Se controlan las transacciones mediante commit y rollback. También se pueden usar contextos para mayor seguridad.

# Transacción básica
try:
    usuario_trans = Usuario(nombre="Prueba", correo="prueba@ejemplo.com")
    sesion.add(usuario_trans)
    sesion.commit()
except Exception as error:
    sesion.rollback()
    print(f"Error en transacción: {error}")

# Contexto de sesión
from contextlib import contextmanager

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

# Uso del contexto
with obtener_sesion() as db:
    usuario_ctx = Usuario(nombre="Contexto", correo="ctx@ejemplo.com")
    db.add(usuario_ctx)

Recomendaciones de Implementación

Para optimizar el uso de SQLAlchemy ORM, se sugieren prácticas como gestionar sesionse por unidad de trabajo, manejar excepciones adecuadamente, utilizar carga ansiosa para evitar problemas N+1, configurar pools de conexiones y validar datos en la capa de aplicación.