Gestión de Bases de Datos con SQLAlchemy ORM en Python

SQLAlchemy es uno de los ORM (Object-Relational Mapper) más utilizados en Python, ofreciendo una forma eficiente y flexible de interactura con bases de datos. Este artículo detalla cómo emplear SQLAlchemy ORM para diversas operaciones de base de datos.

Instalación

Para empezar, instale SQLAlchemy usando pip:


pip install sqlalchemy

Si necesita conectarse a una base de datos específica, también requerirá el controlador correspondiente:


# Para PostgreSQL
pip install psycopg2-binary

# Para MySQL
pip install mysql-connector-python

# Para SQLite (incluido en la biblioteca estándar de Python, no requiere instalación adicional)

Conceptos Fundamentales

  • Engine: Es el punto de partida para cualquier operación de base de datos. Gestiona la conexión con el motor de base de datos.
  • Session: Representa una "conversación" con la base de datos. Administra el ciclo de vida de los objetos y las transacciones.
  • Model: Clases de Python que se mapean a tablas de la base de datos.
  • Query: Objeto utilizado para construir y ejecutar consultas a la base de datos.

Conexión a la Base de Datos

Establezca una conexión utilizando el objeto Engine y cree una fábrica de sesiones.


from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Crear el motor de base de datos
# Ejemplo con SQLite:
db_url = 'sqlite:///mi_ejemplo.db'
engine = create_engine(db_url, echo=True) # echo=True muestra las sentencias SQL generadas

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

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

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

# Obtener una instancia de sesión
session = SessionLocal()

Definición de Modelos de Datos

Defina sus modelos como clases de Python que heredan de una base declarativa.


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

# Base declarativa
Base = declarative_base()

# Tabla de asociación para la relación muchos a muchos
post_tags_association = Table('post_tags', Base.metadata,
    Column('post_id', Integer, ForeignKey('posts.id'), primary_key=True),
    Column('tag_id', Integer, ForeignKey('tags.id'), primary_key=True)
)

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)
    
    # Relación uno a muchos con Post
    publicaciones = relationship("Publicacion", back_populates="autor")
    
class Publicacion(Base):
    __tablename__ = 'posts'
    
    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'))
    
    # Relación muchos a uno con Usuario
    autor = relationship("Usuario", back_populates="publicaciones")
    
    # Relación muchos a muchos con Tag
    etiquetas = relationship("Tag", secondary=post_tags_association, back_populates="publicaciones")

class Tag(Base):
    __tablename__ = 'tags'
    
    id = Column(Integer, primary_key=True, index=True)
    nombre = Column(String(30), unique=True, nullable=False)
    
    # Relación muchos a muchos con Publicacion
    publicaciones = relationship("Publicacion", secondary=post_tags_association, back_populates="etiquetas")

Creación de Tablas en la Base de Datos

Genere las tablas en la base de datos a partir de sus modelos definidos.


# Crear todas las tablas definidas en los modelos
Base.metadata.create_all(bind=engine)

# Para eliminar todas las tablas (usar con precaución)
# Base.metadata.drop_all(bind=engine)

Operaciones CRUD Básicas

Creación de Registros


# Crear un nuevo usuario
nuevo_usuario = Usuario(nombre="Ana García", correo="ana.garcia@ejemplo.com")
session.add(nuevo_usuario)
session.commit() # Confirma la transacción

# Crear múltiples registros a la vez
session.add_all([
    Usuario(nombre="Luis Pérez", correo="luis.perez@ejemplo.com"),
    Usuario(nombre="María López", correo="maria.lopez@ejemplo.com")
])
session.commit()

Lectura de Registros


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

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

# Obtener un usuario por su clave primaria (ID)
usuario_por_id = session.query(Usuario).get(1)

Actualización de Registros


# Consultar y luego actualizar un registro
usuario_a_actualizar = session.query(Usuario).get(1)
if usuario_a_actualizar:
    usuario_a_actualizar.nombre = "Ana García Update"
    session.commit()

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

Eliminación de Registros


# Consultar y luego eliminar un registro
usuario_a_eliminar = session.query(Usuario).get(2)
if usuario_a_eliminar:
    session.delete(usuario_a_eliminar)
    session.commit()

# Eliminación masiva
session.query(Usuario).filter(Usuario.nombre == "Luis Pérez").delete(synchronize_session=False)
session.commit()

Consultas Avanzadas

Filtrado de Datos


from sqlalchemy import or_

# Filtrar por igualdad
usuario_especifico = session.query(Usuario).filter(Usuario.nombre == "María López").first()

# Filtrar usando LIKE para búsquedas parciales
usuarios_por_nombre = session.query(Usuario).filter(Usuario.nombre.like("%López%")).all()

# Filtrar usando IN
usuarios_seleccionados = session.query(Usuario).filter(Usuario.nombre.in_(["Ana García Update", "María López"])).all()

# Múltiples condiciones de filtro (AND implícito)
usuarios_filtrados = session.query(Usuario).filter(
    Usuario.nombre.like("Mar%"), 
    Usuario.correo.endswith("@ejemplo.com")
).all()

# Usando OR
usuarios_o_condicion = session.query(Usuario).filter(
    or_(Usuario.nombre == "Ana García Update", Usuario.nombre == "Luis Pérez")
).all()

# Filtrar por desigualdad
usuarios_no_ana = session.query(Usuario).filter(Usuario.nombre != "Ana García Update").all()

Agregaciones y Funciones


from sqlalchemy import func

# Contar el número total de usuarios
total_usuarios = session.query(func.count(Usuario.id)).scalar()

# Contar publicaciones por usuario
publicaciones_por_usuario = session.query(
    Usuario.nombre, 
    func.count(Publicacion.id)
).join(Publicacion).group_by(Usuario.nombre).all()

# Calcular el promedio de IDs de usuario
promedio_id_usuarios = session.query(func.avg(Usuario.id)).scalar()

Consultas con JOIN


# INNER JOIN (implícito al usar join)
usuarios_y_publicaciones = session.query(Usuario, Publicacion).join(Publicacion).filter(Publicacion.titulo.like("%SQLAlchemy%")).all()

# LEFT OUTER JOIN
usuarios_con_o_sin_publicaciones = session.query(Usuario, Publicacion).outerjoin(Publicacion).all()

# Especificar la condición de JOIN explícitamente
usuarios_y_sus_posts = session.query(Usuario, Publicacion).join(Publicacion, Usuario.id == Publicacion.autor_id).all()

Mannejo de Relaciones

SQLAlchemy facilita la gestión de relaciones entre modelos.


# Crear una publicación asociada a un usuario existente
usuario_existente = session.query(Usuario).filter_by(nombre="Ana García Update").first()
if usuario_existente:
    nueva_publicacion = Publicacion(titulo="Mi primera entrada", contenido="Contenido de ejemplo.", autor=usuario_existente)
    session.add(nueva_publicacion)
    session.commit()

# Acceder a datos relacionados
print(f"Autor de '{nueva_publicacion.titulo}': {nueva_publicacion.autor.nombre}")
print(f"Publicaciones de {usuario_existente.nombre}:")
for pub in usuario_existente.publicaciones:
    print(f"  - {pub.titulo}")

# Gestionar relaciones Muchos a Muchos
tag_python = Tag.query.filter_by(nombre="Python").first()
if not tag_python:
    tag_python = Tag(nombre="Python")
    session.add(tag_python)

tag_sqlalchemy = Tag.query.filter_by(nombre="SQLAlchemy").first()
if not tag_sqlalchemy:
    tag_sqlalchemy = Tag(nombre="SQLAlchemy")
    session.add(tag_sqlalchemy)
session.commit() # Asegurarse que los tags existen

# Asociar tags a la publicación
nueva_publicacion.etiquetas.append(tag_python)
nueva_publicacion.etiquetas.append(tag_sqlalchemy)
session.commit()

print(f"Etiquetas para '{nueva_publicacion.titulo}':")
for tag in nueva_publicacion.etiquetas:
    print(f"  - {tag.nombre}")

Gestión de Transacciones

Es crucial manejar las transacciones correctamente para mantener la integridad de los datos.


# Manejo explícito de transacciones con commit y rollback
try:
    usuario_trans = Usuario(nombre="Usuario Transaccional", correo="trans@ejemplo.com")
    session.add(usuario_trans)
    
    # Podría ocurrir otra operación aquí
    # ...
    
    session.commit() # Confirma todas las operaciones si no hay errores
except Exception as e:
    session.rollback() # Revierte todas las operaciones en caso de error
    print(f"Error en la transacción: {e}")

# Uso del gestor de contexto para transacciones automáticas
from sqlalchemy.orm import Session

def crear_publicacion_segura(db_session: Session, usuario_id: int, titulo: str, contenido: str):
    try:
        usuario = db_session.query(Usuario).get(usuario_id)
        if not usuario:
            raise ValueError(f"Usuario con ID {usuario_id} no encontrado.")
        
        publicacion = Publicacion(titulo=titulo, contenido=contenido, autor=usuario)
        db_session.add(publicacion)
        db_session.commit()
        return publicacion
    except Exception as ex:
        db_session.rollback()
        raise ex

# Ejemplo de uso con el gestor de contexto
with SessionLocal() as db:
    try:
        publicacion_creada = crear_publicacion_segura(db, usuario_existente.id, "Publicación Segura", "Contenido seguro.")
        print("Publicación creada exitosamente.")
    except Exception as err:
        print(f"Fallo al crear publicación: {err}")

Buenas Prácticas

  • Gestión de Sesiones: Abra una nueva sesión para cada unidad de trabajo (ej. una petición web) y ciérrela al finalizar.
  • Manejo de Errores: Implemente bloques try...except para capturar excepciones y realizar rollback si es necesario.
  • Optimización de Consultas: Utilice joinedload o selectinload para evitar el probleam N+1 en cargas de relaciones (eager loading).
  • Pool de Conexiones: Configure adecuadamente el tamaño del pool de conexiones para optimizar el rendimiento.
  • Validación de Datos: Valide los datos antes de guardarlos en la base de datos, ya sea en el modelo o en la lógica de negocio.

SQLAlchemy ORM ofrece una solución robusta y flexible para la persistencia de datos en Python. Dominar sus características, desde la definición de modelos hasta la gestión de transacciones y optimización de consultas, es fundamental para el desarrollo de aplicaciones eficientes y escalables.

Etiquetas: SQLAlchemy ORM Python bases de datos CRUD

Publicado el 6-19 06:40

Etiquetas populares

©2026 Friki Work