Creación de registros (Create)

Para crear nuevos registros en la base de datos utilizando Flask y SQLAlchemy, es fundamental entender cómo interactuar con el modelo y la sesión de base de datos. La creación de registros implica instanciar objetos de los modelos definidos y luego agregar esos objetos a la sesión para finalmente confirmar los cambios en la base de datos.

Por ejemplo, supongamos que tenemos un modelo llamado Usuario definido de la siguiente manera:

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class Usuario(db.Model):
    __tablename__ = 'usuarios'
    id = db.Column(db.Integer, primary_key=True)
    nombre = db.Column(db.String(100), nullable=False)
    correo = db.Column(db.String(100), unique=True, nullable=False)

Para crear un nuevo usuario, instanciamos el modelo Usuario con los datos necesarios:

nuevo_usuario = Usuario(nombre='Juan Pérez', correo='juan.perez@example.com')

Es importante asegurarse de que los campos marcados como nullable=False y unique=True cumplan con las restricciones definidas en el modelo para evitar errores.

Una vez instanciado el objeto, lo agregamos a la sesión de la base de datos y lo confirmamos:

db.session.add(nuevo_usuario)
db.session.commit()

El método add() agrega el objeto a la sesión actual, mientras que commit() persiste los cambios en la base de datos. Si se omite commit(), los cambios no se guardarán de forma permanente.

También es posible agregar múltiples registros a la vez utilizando add_all():

usuarios = [
    Usuario(nombre='María López', correo='maria.lopez@example.com'),
    Usuario(nombre='Carlos Sánchez', correo='carlos.sanchez@example.com')
]
db.session.add_all(usuarios)
db.session.commit()

Usar add_all() es eficiente al insertar múltiples registros, ya que permite agregarlos en una sola operación antes de confirmar los cambios.

Para manejar posibles excepciones, especialmente al trabajar con restricciones como unique, es recomendable utilizar bloques try-except:

from sqlalchemy.exc import IntegrityError

try:
    db.session.add(nuevo_usuario)
    db.session.commit()
except IntegrityError:
    db.session.rollback()
    # Manejar el error, por ejemplo, notificando que el correo ya existe

En este ejemplo, si se produce una violación de integridad, como intentar insertar un correo electrónico que ya existe, se hace un rollback() de la sesión para revertir los cambios pendientes.

Además, podemos crear registros dentro de una función de vista en Flask:

from flask import request, jsonify

@app.route('/usuarios', methods=['POST'])
def crear_usuario():
    datos = request.get_json()
    nuevo_usuario = Usuario(nombre=datos['nombre'], correo=datos['correo'])
    db.session.add(nuevo_usuario)
    db.session.commit()
    return jsonify({'mensaje': 'Usuario creado exitosamente'}), 201

Aquí, obtenemos los datos enviados en una solicitud POST, creamos un nuevo usuario y lo guardamos en la base de datos. Es crucial validar los datos recibidos para evitar errores y posibles vulnerabilidades.

Para asegurar que los datos cumplen con los formatos esperados, podemos utilizar validaciones adicionales o bibliotecas como Marshmallow, aunque su uso podría estar cubierto en otra sección.

Finalmente, al crear registros, es buena práctica proporcionar retroalimentación al usuario o al cliente que realiza la petición, indicando el resultado de la operación y cualquier información relevante.

Lectura y filtrado de información (Read)

La lectura y el filtrado de información de la base de datos son operaciones esenciales en cualquier aplicación Flask utilizando SQLAlchemy. Para acceder a los registros almacenados, usamos consultas al modelo definido, aprovechando las capacidades que ofrece el ORM de SQLAlchemy.

Para comenzar, supongamos que tenemos el siguiente modelo Usuario:

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class Usuario(db.Model):
    __tablename__ = 'usuarios'
    id = db.Column(db.Integer, primary_key=True)
    nombre = db.Column(db.String(100), nullable=False)
    correo = db.Column(db.String(100), unique=True, nullable=False)

Obtención de todos los registros

Para obtener todos los usuarios de la tabla usuarios, podemos utilizar el método query.all():

usuarios = Usuario.query.all()

Esto nos devuelve una lista de objetos Usuario. Podemos iterar sobre esta lista para acceder a los datos:

for usuario in usuarios:
    print(usuario.nombre, usuario.correo)

Si queremos mostrar esta información en una respuesta JSON, podemos convertir los objetos en un formato serializable:

from flask import jsonify

@app.route('/usuarios', methods=['GET'])
def obtener_usuarios():
    usuarios = Usuario.query.all()
    lista_usuarios = []
    for usuario in usuarios:
        usuario_data = {
            'id': usuario.id,
            'nombre': usuario.nombre,
            'correo': usuario.correo
        }
        lista_usuarios.append(usuario_data)
    return jsonify(lista_usuarios)

En este ejemplo, estamos creando una ruta que devuelve todos los usuarios en formato JSON. Cada objeto Usuario se convierte en un diccionario que se agrega a una lista, la cual se devuelve como respuesta.

Filtrado de registros

Para filtrar registros según ciertos criterios, utilizamos los métodos filter_by() y filter().

Uso de filter_by()

El método filter_by() permite filtrar por igualdad. Por ejemplo, para obtener un usuario por su correo electrónico:

usuario = Usuario.query.filter_by(correo='juan.perez@example.com').first()

El método first() devuelve el primer resultado de la consulta o None si no se encuentra ningún registro.

Uso de filter()

El método filter() es más flexible y permite utilizar expresiones. Por ejemplo, para encontrar usuarios cuyo nombre contiene la palabra "María":

usuarios = Usuario.query.filter(Usuario.nombre.like('%María%')).all()

La expresión Usuario.nombre.like('%María%') busca coincidencias que contengan la cadena "María".

También podemos combinar filtros usando operadores lógicos:

from sqlalchemy import or_

usuarios = Usuario.query.filter(
    or_(
        Usuario.nombre.like('%María%'),
        Usuario.nombre.like('%Carlos%')
    )
).all()

En este ejemplo, obtenemos usuarios cuyo nombre contiene "María" o "Carlos".

Obtener un registro por su clave primaria

Para obtener un registro específico por su clave primaria, podemos usar el método get():

usuario = Usuario.query.get(1)

Esto devuelve el usuario con id igual a 1 o None si no existe.

Ordenamiento y limitación de resultados

Podemos ordenar los resultados usando order_by() y limitar la cantidad de registros con limit():

usuarios = Usuario.query.order_by(Usuario.nombre.desc()).limit(10).all()

Aquí obtenemos los primeros 10 usuarios ordenados por nombre en orden descendente.

Paginación de resultados

La paginación es útil para manejar grandes cantidades de datos. Podemos implementar paginación usando offset() y limit():

pagina = 2
tamaño_pagina = 10
usuarios = Usuario.query.offset((pagina - 1) * tamaño_pagina).limit(tamaño_pagina).all()

Esto obtiene la segunda página de usuarios, mostrando 10 usuarios por página.

Acceso a columnas específicas

Si solo necesitamos ciertas columnas, podemos especificarlas en la consulta:

nombres = db.session.query(Usuario.nombre).all()

Esto devuelve una lista de tuplas con los nombres de los usuarios.

Consultas avanzadas

SQLAlchemy permite realizar consultas más complejas, como agregaciones o subconsultas.

Conteo de registros

Para contar el número de usuarios:

from sqlalchemy import func

total_usuarios = db.session.query(func.count(Usuario.id)).scalar()

El método scalar() devuelve el valor escalar resultante de la consulta.

Uso de subconsultas

Podemos utilizar subconsultas para realizar operaciones más avanzadas:

subconsulta = db.session.query(Usuario.id).filter(Usuario.nombre.like('%María%')).subquery()
usuarios = Usuario.query.filter(Usuario.id.in_(subconsulta)).all()

Manejo de relaciones

Si el modelo Usuario tiene una relación con otro modelo, por ejemplo, Publicacion, podemos acceder a las publicaciones de un usuario:

class Publicacion(db.Model):
    __tablename__ = 'publicaciones'
    id = db.Column(db.Integer, primary_key=True)
    titulo = db.Column(db.String(200), nullable=False)
    contenido = db.Column(db.Text, nullable=False)
    usuario_id = db.Column(db.Integer, db.ForeignKey('usuarios.id'), nullable=False)
    usuario = db.relationship('Usuario', back_populates='publicaciones')

Usuario.publicaciones = db.relationship('Publicacion', back_populates='usuario', lazy='dynamic')

Para obtener las publicaciones de un usuario específico:

usuario = Usuario.query.get(1)
for publicacion in usuario.publicaciones:
    print(publicacion.titulo)

El parámetro lazy='dynamic' en relationship permite realizar consultas adicionales sobre la relación.

Manejo de errores en consultas

Al realizar consultas, es importante manejar posibles excepciones:

from sqlalchemy.orm.exc import NoResultFound, MultipleResultsFound

try:
    usuario = Usuario.query.filter_by(correo='no.existe@example.com').one()
except NoResultFound:
    # Manejar el caso en que no se encuentre el usuario
except MultipleResultsFound:
    # Manejar el caso en que se encuentren múltiples usuarios

Consultas en funciones de vista

Podemos integrar estas consultas en las funciones de vista de Flask:

@app.route('/usuarios/<int:id>', methods=['GET'])
def obtener_usuario_por_id(id):
    usuario = Usuario.query.get_or_404(id)
    usuario_data = {
        'id': usuario.id,
        'nombre': usuario.nombre,
        'correo': usuario.correo
    }
    return jsonify(usuario_data)

En este ejemplo, utilizamos get_or_404(id) para obtener el usuario o devolver un error 404 si no existe.

Actualización y guardado de cambios (Update)

La actualización de registros en la base de datos es una operación fundamental al trabajar con Flask y SQLAlchemy. Para modificar los datos existentes, primero debemos recuperar el registro que deseamos actualizar, realizar los cambios necesarios en sus atributos y luego guardar los cambios confirmando la sesión.

Supongamos que tenemos el modelo Usuario definido anteriormente:

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class Usuario(db.Model):
    __tablename__ = 'usuarios'
    id = db.Column(db.Integer, primary_key=True)
    nombre = db.Column(db.String(100), nullable=False)
    correo = db.Column(db.String(100), unique=True, nullable=False)

Recuperación del registro a actualizar

Para actualizar un usuario específico, primero debemos localizarlo en la base de datos. Esto se logra mediante una consulta al modelo:

usuario = Usuario.query.get(1)

En este caso, estamos obteniendo el usuario con id igual a 1. Si el usuario no existe, usuario será None, por lo que es buena práctica verificar su existencia antes de continuar:

if usuario is None:
    # Manejar el caso en que el usuario no existe

Modificación de los atributos

Una vez que hemos recuperado el objeto usuario, podemos modificar sus atributos de la siguiente manera:

usuario.nombre = 'Juan Carlos Pérez'
usuario.correo = 'juancarlos.perez@example.com'

Es importante asignar valores que cumplan con las restricciones definidas en el modelo, como nullable=False y unique=True, para evitar violaciones de integridad.

Confirmación de los cambios en la sesión

Después de modificar los atributos, debemos guardar los cambios en la base de datos. Esto se hace confirmando la sesión:

db.session.commit()

El método commit() persiste todos los cambios pendientes en la sesión actual a la base de datos. Si se producen errores, como violaciones de restricciones de unicidad, se lanzará una excepción.

Manejo de excepciones y rollback

Es recomendable envolver las operaciones de actualización en un bloque try-except para manejar excepciones y realizar un rollback() en caso de errores:

from sqlalchemy.exc import IntegrityError

try:
    db.session.commit()
except IntegrityError:
    db.session.rollback()
    # Manejar el error, por ejemplo, notificar que el correo ya está en uso

El método rollback() revierte todos los cambios pendientes en la sesión, dejando la sesión en un estado limpio para futuras operaciones.

Actualización masiva de registros

Para actualizar múltiples registros a la vez, podemos utilizar el método update() en una consulta. Por ejemplo, para incrementar el valor de un campo en todos los registros:

Usuario.query.filter(Usuario.activo == True).update({Usuario.activo: False})
db.session.commit()

En este ejemplo, estamos desactivando todos los usuarios que están activos. Es importante tener en cuenta que update() ejecuta la operación directamente en la base de datos y no devuelve objetos Usuario.

Uso de transacciones

Al realizar operaciones que involucran múltiples pasos, es recomendable utilizar transacciones para asegurar la integridad de los datos. SQLAlchemy proporciona el contexto session.begin() para manejar transacciones:

with db.session.begin():
    usuario = Usuario.query.get(1)
    usuario.nombre = 'Juan Carlos Pérez'
    usuario.correo = 'juancarlos.perez@example.com'
    # Otros cambios adicionales

Dentro del bloque with, si se produce una excepción, se realizará automáticamente un rollback(). Si todo va bien, se realizará un commit() al finalizar el bloque.

Actualización en funciones de vista

Podemos integrar la actualización de registros en las funciones de vista de Flask. Por ejemplo:

from flask import request, jsonify

@app.route('/usuarios/<int:id>', methods=['PUT'])
def actualizar_usuario(id):
    datos = request.get_json()
    usuario = Usuario.query.get_or_404(id)
    usuario.nombre = datos.get('nombre', usuario.nombre)
    usuario.correo = datos.get('correo', usuario.correo)
    try:
        db.session.commit()
        return jsonify({'mensaje': 'Usuario actualizado exitosamente'}), 200
    except IntegrityError:
        db.session.rollback()
        return jsonify({'error': 'El correo electrónico ya está en uso'}), 400

En este ejemplo:

• Usamos get_or_404(id) para obtener el usuario o devolver un error 404 si no existe.
• Actualizamos los campos nombre y correo si se proporcionan en los datos recibidos.
• Manejamos posibles violaciones de integridad, como correos duplicados.

Validación de datos

Antes de actualizar los registros, es fundamental validar los datos recibidos para asegurar que cumplen con los requisitos de la aplicación y evitar errores o vulnerabilidades.

Podemos utilizar bibliotecas como Marshmallow para la validación y deserialización de datos, aunque su uso específico puede ser tratado en otra sección.

Optimización de consultas

Cuando trabajamos con un gran número de registros, es conveniente optimizar las consultas y operaciones de actualización para mejorar el rendimiento. Por ejemplo, al actualizar múltiples registros, podemos utilizar subconsultas o condiciones más específicas para limitar el alcance de la operación.

Ejemplo de actualización condicional

Supongamos que necesitamos actualizar el dominio de los correos electrónicos de todos los usuarios que tienen un dominio antiguo:

usuarios_actualizados = Usuario.query.filter(Usuario.correo.like('%@antiguo-dominio.com')).update(
    {Usuario.correo: Usuario.correo.op('REPLACE')('@antiguo-dominio.com', '@nuevo-dominio.com')},
    synchronize_session=False
)
db.session.commit()

En este caso:

• Estamos reemplazando el dominio antiguo por el nuevo en los correos electrónicos.
• Usamos synchronize_session=False para mejorar el rendimiento, aunque debemos tener precaución al utilizarlo.

Actualización parcial de registros

En ocasiones, es necesario permitir la actualización parcial de un registro, modificando solo algunos campos. Para ello, podemos utilizar el método update() en combinación con filtros:

Usuario.query.filter_by(id=1).update({'nombre': 'Juan C. Pérez'})
db.session.commit()

Este enfoque es útil para actualizar campos específicos sin necesidad de cargar el objeto completo en memoria.

Advertencia sobre cargas diferidas

Al utilizar update() directamente en la consulta, debemos tener cuidado con las cargas diferidas y las relaciones, ya que los cambios realizados de esta manera no desencadenan eventos ORM ni actualizan objetos en memoria.

Eliminación de registros (Delete)

La eliminación de registros es una operación crítica en el manejo de bases de datos, ya que implica la remoción permanente de datos. En Flask, utilizando SQLAlchemy como ORM, este proceso se realiza interactuando con el modelo y la sesión de la base de datos. A continuación, se detalla cómo llevar a cabo esta operación de manera segura y eficiente.

Eliminación de un registro individual

Para eliminar un registro específico, primero es necesario recuperar el objeto que se desea borrar. Esto se hace mediante una consulta al modelo:

usuario = Usuario.query.get(1)

En este ejemplo, obtenemos el usuario con id igual a 1. Es fundamental verificar que el objeto existe antes de proceder:

if usuario is None:
    # Manejar el caso en que el usuario no existe

Una vez que hemos confirmado la existencia del objeto, podemos eliminarlo utilizando el método delete() de la sesión:

db.session.delete(usuario)
db.session.commit()

El método delete() marca el objeto para eliminación, y commit() persiste el cambio en la base de datos. Tras esta operación, el registro se elimina de manera permanente.

Manejo de errores y transacciones

Al eliminar registros, es crucial manejar posibles excepciones y asegurar la integridad de la base de datos. Por ejemplo, si existen claves foráneas que dependen del registro a eliminar, puede producirse una violación de restricciones de integridad referencial. Para manejar estos casos, se recomienda el uso de bloques try-except:

from sqlalchemy.exc import IntegrityError

try:
    db.session.delete(usuario)
    db.session.commit()
except IntegrityError:
    db.session.rollback()
    # Manejar el error, como notificar que el usuario tiene registros relacionados

El método rollback() revierte cualquier cambio pendiente en la sesión, manteniendo la base de datos en un estado consistente.

Eliminación en cascada y restricciones

Si el modelo Usuario tiene relaciones con otros modelos, como Publicacion, es importante definir el comportamiento al eliminar un usuario. Una opción es configurar la eliminación en cascada, de manera que al borrar un usuario, también se eliminen sus publicaciones asociadas:

class Publicacion(db.Model):
    __tablename__ = 'publicaciones'
    id = db.Column(db.Integer, primary_key=True)
    titulo = db.Column(db.String(200), nullable=False)
    contenido = db.Column(db.Text, nullable=False)
    usuario_id = db.Column(db.Integer, db.ForeignKey('usuarios.id', ondelete='CASCADE'), nullable=False)
    usuario = db.relationship('Usuario', back_populates='publicaciones')

Usuario.publicaciones = db.relationship(
    'Publicacion',
    back_populates='usuario',
    cascade='all, delete-orphan',
    passive_deletes=True
)

En este ejemplo:

• ondelete='CASCADE' en la clave foránea define el comportamiento a nivel de base de datos.
• El parámetro cascade='all, delete-orphan' en relationship especifica que las publicaciones relacionadas se eliminarán junto con el usuario.

Esta configuración garantiza que no queden registros huérfanos y que la integridad referencial se mantenga.

Eliminación de múltiples registros

Para eliminar varios registros que cumplen cierta condición, podemos utilizar el método delete() directamente sobre una consulta:

usuarios_inactivos = Usuario.query.filter(Usuario.activo == False)
usuarios_inactivos.delete(synchronize_session=False)
db.session.commit()

En este caso, se eliminan todos los usuarios que no están activos. El parámetro synchronize_session=False mejora el rendimiento al omitir la sincronización de objetos en la sesión, pero debe usarse con precaución.

Eliminación lógica (borrado suave)

En lugar de eliminar físicamente los registros, en algunas aplicaciones es preferible realizar una eliminación lógica, también conocida como borrado suave. Esto implica marcar los registros como eliminados sin borrarlos de la base de datos:

class Usuario(db.Model):
    __tablename__ = 'usuarios'
    id = db.Column(db.Integer, primary_key=True)
    nombre = db.Column(db.String(100), nullable=False)
    correo = db.Column(db.String(100), unique=True, nullable=False)
    eliminado = db.Column(db.Boolean, default=False)

Para "eliminar" un usuario, actualizamos el campo eliminado a True:

usuario = Usuario.query.get(1)
usuario.eliminado = True
db.session.commit()

Al realizar consultas, es necesario filtrar los registros eliminados:

usuarios_activos = Usuario.query.filter(Usuario.eliminado == False).all()

El borrado suave permite conservar el historial de datos y proporciona la posibilidad de restaurar registros si es necesario.

Eliminación desde funciones de vista

Podemos integrar la eliminación de registros en las funciones de vista de Flask:

from flask import jsonify

@app.route('/usuarios/<int:id>', methods=['DELETE'])
def eliminar_usuario(id):
    usuario = Usuario.query.get_or_404(id)
    db.session.delete(usuario)
    try:
        db.session.commit()
        return jsonify({'mensaje': 'Usuario eliminado exitosamente'}), 200
    except IntegrityError:
        db.session.rollback()
        return jsonify({'error': 'No se puede eliminar el usuario debido a referencias existentes'}), 400

En este ejemplo:

• Usamos get_or_404(id) para obtener el usuario o devolver un error 404 si no existe.
• Manejamos excepciones para notificar si la eliminación no es posible por restricciones.

Consideraciones de seguridad

La eliminación de registros debe manejarse con precaución para evitar pérdida de datos y garantizar la seguridad de la aplicación:

• Validación de permisos: Asegurarse de que el usuario tiene autorización para eliminar el registro.
• Confirmaciones: En aplicaciones con interfaces de usuario, solicitar confirmación antes de eliminar.
• Auditoría: Mantener registros de quién y cuándo se eliminó un registro para fines de auditoría.

Uso de transacciones para operaciones críticas

Al realizar operaciones críticas, es recomendable utilizar transacciones para garantizar la consistencia de los datos:

with db.session.begin():
    db.session.delete(usuario)
    # Realizar otras operaciones si es necesario

Si ocurre una excepción dentro del bloque with, la sesión se revertirá automáticamente.

Eliminación condicional basada en subconsultas

Podemos eliminar registros basados en condiciones más complejas utilizando subconsultas:

from sqlalchemy import exists

subconsulta = db.session.query(Compra.usuario_id).filter(Compra.monto > 1000).subquery()
Usuario.query.filter(Usuario.id.in_(subconsulta)).delete(synchronize_session=False)
db.session.commit()

En este ejemplo, eliminamos usuarios que han realizado compras con montos superiores a 1000.

Advertencia sobre la eliminación en cascada

La eliminación en cascada puede tener consecuencias significativas si no se configura correctamente. Es vital entender cómo afecta a las relaciones entre modelos y garantizar que solo se eliminen los registros deseados.