SQLAlchemy - CRUD

sqlalchemy_crud.md

CRUD con SQLAlchemy

Tabla de contenidos

1. Agregando un modelo
2. Insert
3. Select
4. Delete
5. Update
6. Documentaci

Agregando un modelo

Para cada model tienes que declarar un class con las propiedades y el método serialize() que devolverá una representación en formato de diccionario de la clase.

from flask_sqlalchemy import SQLAlchemy


db = SQLAlchemy()


class Users(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password = db.Column(db.String(20), unique=False, nullable=False)
    name = db.Column(db.String(80), nullable=False)
    is_active = db.Column(db.Boolean)

    # Tell python how to print the class object on the console
    def __repr__(self):
        return f'<User: {self.email}>'

    # Tell python how convert the class object into a dictionary ready to jsonify
    def serialize(self):
        return {'id': self.id,
                'email': self.email,
                'name': self.name,
                'is_active': self.is_active}

INSERT: Insertando un registro en la base de datos

Para insertar un registro en la base de datos, es necesario:

1. Primero, contar con la instancia que se desea añadir.
2. A continuación, agregarlo a la sesión de la base de datos con db.session.add() y
3. Finalmente completar la acción con un commit.

Ejemplo:

user = Users(name = 'Joe', 
             email = 'joe@domain.com',
             password = '1234',
             is_active = True)
db.session.add(user)
db.session.commit()

# nota: email = data.get('email')

SELECT: Buscando o recuperando registros de la base de datos

Tres ejemplos para obtener información de la base de datos:

1. Obtener todos los registros de una tabla/modelo

   response_body = {}
   users = db.session.execute(db.select(Users)).scalars()
   results = [row.serialize() for row in users]
   response_body['results'] = results

2. Obtener un único registro en función de su clave principal mediante

   # Hay varias formas de obtener un registro
   user_id = 1  # Este dato lo obtienes en algún lugar anterior de tu código
   
   # La más sencilla: retorna el valor encontrado o `None` en caso de no encontrarlo
   user = db.session.get(Users, user_id)
   
   # Retorna el valor encontrado o retorna un error 404 con la `description` que definimos
   user = db.one_or_404(db.select(Users).filter_by(id=user_id), 
                        description=f"User not found , 404")
   
   # Retorna el primer valor encontrado que cumple la condición `where`
   user = db.session.execute(db.select(Users).where(Users.id == user_id)).scalar()

3. Obtener un grupo de registros en función de una consulta

   response_body = {}
   users = db.session.execute(db.select(Users).where(Users.name == 'Joe', Users.is_active == True)).scalar()
   results = [row.serialize() for row in users]
   response_body['results'] = results

DELETE: Eliminando un registro de la base de datos

Para eliminar un registro de la base de datos es necesario:

1. Seleccionar previamente la instancia que se desee borrar (a través de su clave primaria, por ejemplo)
2. Eliminarla utilizando db.session.delete(),
3. Finalmente, hacer commit.

Ejemplo:

user = db.session.execute(db.select(Users).where(Users.id == user_id)).scalar()
db.session.delete(user)
db.session.commit()

UDPATE: Actualizar un registro

Para modificar un registro, hay que realizar lo siguiente:

1. Seleccionar previamente el registro de la base de datos,
2. Luego puedes trabajar con él cambiando sus propiedades y
3. Finalmente, hacer commit.

Ejemplo:

user = db.session.execute(db.select(Users).where(Users.id == user_id)).scalar()
user.name = "Bob"
db.session.commit()

Documentación

Migracion SQLAlchemy 1.4 a SQLAlchemy 2.0

2.0 Migration - ORM Usage

Flask-SQLAlchemy

Flask SQALchemy