orm

Ver en Git

---

Mapeo de la base de datos

La funcion start_mappers es la encagada de inicializar el mappeo de la db, se llama a esta funcion en archivo run.py encargado de inicializar la api.

[!NOTE] from features.activities.orm.mappers import start_mappers

Crear mappeo

from sqlalchemy import UUID, Column, Integer, String, Table

# Modelo de python
@dataclass
class Farm:
    id: int
    name: str
    farm_uuid: str

# Tabla sql
farm_table = Table(
    "estancias",
    metadata,
    Column("id", Integer, primary_key=True, autoincrement=True),
    Column("nombre", String, key='name'),
    Column("uuid_estancia", UUID(as_uuid=False), key='farm_uuid'),
)

# Defina que la relacion de las entidad con su tabla
mapper_registry.map_imperatively(Farm, farm_table)

Properties

Relaciones entre tables

Al momento de realizar el mappeo podemos agregar propiedades, una de estas pueden ser las relaciones entre tablas

from sqlalchemy.orm import relationship
mapper_registry.map_imperatively(
    Labour,
    labour_table,
    properties={
        #Agregamos una lista de LabourInput
        'inputs': relationship(
            LabourInput,
            cascade="save-update, merge",
            lazy='joined',
            primaryjoin=(labour_table.c.id_labour == input_labour_table.c.id_labour)
            & (input_labour_table.c.deleted_at.is_(None)),
            viewonly=False,
        )
    }

cascade

• 'save-update': Propaga las operaciones de persistencia de "save" y "update" desde el objeto principal a los objetos secundarios.
• 'merge': Propaga la operación de persistencia de "merge" desde el objeto principal a los objetos secundarios.
• 'refresh': Propaga la operación de persistencia de "refresh" desde el objeto principal a los objetos secundarios.
• 'expunge': Propaga la operación de persistencia de "expunge" desde el objeto principal a los objetos secundarios.
• 'delete': Propaga la operación de persistencia de "delete" desde el objeto principal a los objetos secundarios.
• all': Realiza todas las acciones de cascada posibles.
• None: no se propaga ninguna accion, este es el valor por defecto

lazy

• 'select' (predeterminado): Los objetos relacionados se recuperan de la base de datos cuando se accede por primera vez a la propiedad de la relación. Se realiza una consulta SELECT para cargar los objetos relacionados en ese momento.
• 'joined': Realiza una unión (JOIN) en la consulta SQL original para cargar los objetos relacionados junto con los objetos principales. Esto suele ser más eficiente que la carga perezosa select.
• 'subquery': Similar a 'joined', pero utiliza una subconsulta para cargar los objetos relacionados. Puede ser más eficiente que 'joined' en ciertos casos.

Esto se puede modificar tambien al momento se realizar las consultas

# set children to load lazily
session.query(Parent).options(lazyload(Parent.children)).all()

# set children to load eagerly with a join
session.query(Parent).options(joinedload(Parent.children)).all()

# Tambien se pueden encadenar
session.query(Parent).options(
    joinedload(Parent.children).subqueryload(Child.subelements)
).all()

primaryjoin

Por defecto utiliza la ForeignKey para realizar el join, en el ejemplo anterior cambiamos el comportamiento, para que ademas verifique que no este eliminado el objeto

primaryjoin=(labour_table.c.id_labour == input_labour_table.c.id_labour)
            & (input_labour_table.c.deleted_at.is_(None))

viewonly

Como el nombre lo indica sirve para que el objeto sea solo lectura, por lo tanto sqlalquemy no intentara actualizar el objeto de la relacion. Solo lo agrego en caso que sea True

Relacion de muchos a muchos

staff_rol_association_table = Table(
    'resources_staff_roles',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('person_rol_id', Integer, ForeignKey('resources_staff_rol.person_rol_id')),
    Column('person_id', Integer, ForeignKey('resources_staff.person_id')),
    Column('deleted_at', TIMESTAMP),
)

mapper_registry.map_imperatively(
        Staff,
        staff_table,
        properties={
            'roles': relationship(
                Role,
                secondary=staff_rol_association_table,
                secondaryjoin=(
                    (rol_table.c.person_rol_id == staff_rol_association_table.c.person_rol_id)
                    & (staff_rol_association_table.c.deleted_at.is_(None))
                ),
                viewonly=True,
            )
        },
    )

Para relacionar Staff con Role utilizamos la tabla intermedia secondary=staff_rol_association_table

En este caso tenemos 2 join:

• primaryjoin: dejo el default por lo tanto tomara ForeignKey('resources_staff.person_id')
• secondaryjoin: en este caso tomara por defecto ForeignKey('resources_staff.person_rol_id'), aunque estoy modificando su comportamiento para que tambien verifique que la relacion no haya sido eliminada

Agregar columnas

Se puede agregar una columna, ejemplo

mapper_registry.map_imperatively(
    WorkOrderInputRequired,
    work_order_input_retired_table,
    properties={
        "input_name": column_property(
            select(Input.name).where(Input.uuid == work_order_input_retired_table.c.input_uuid).scalar_subquery()
        )
    },
)

Se puede agregar deferred, para que se cargue unicamente cuando se acceda a la variable (esto implica realizar una consulta en ese momento)

mapper_registry.map_imperatively(
    WorkOrderInputRequired,
    work_order_input_retired_table,
    properties={
        "input_name": column_property(
            deferred(
                select(Input.name).where(Input.uuid == work_order_input_retired_table.c.input_uuid).scalar_subquery()
            )
        )
    },
)

Composite

Podemos separar las columnas de una tabla en objeros distintos ej:

seeding_table = Table(
    "siembras",
    metadata,
    ...
    # seed_rate
    Column("densidad", Float),
    Column("unidad_medida_densidad", String),
)

mapper_registry.map_imperatively(
    Seeding,
    seeding_table,
        properties={
            "seed_rate": composite(
                Measurement,
                seeding_table.c.densidad,
                seeding_table.c.unidad_medida_densidad,
            )
    }
)

te esta manera las columnas densidad y unidad_medida_densidad no estaran directamente en el obj Seeding sino en Seeding.seed_rate siendo un obj de Measurement

Polimorfismo

En ejemplo de esto es el caso de las labores que tenemos una tabla padre "labores" y luego una especifica para cada tipo de labor, esto lo podemos definir en el orm para que solo nos traiga el tipo correspondiente

# Clase padre especificamos polymorphic_on(columna que nos indica que tipo es), polymorphic_identity id del tipo (en este caso 0 no hay ninguna labor del tipo general)
mapper_registry.map_imperatively(
    Labour,
    labour_table,
    polymorphic_identity=0,
    polymorphic_on=labour_table.c.labour_type_id,
)
# En las clase hijas se especifica cual es el objeto padre (Labour) y el id del tipo polymorphic_identity (en este caso las labores del tipo 3 las construira como Seeding)
mapper_registry.map_imperatively(
    Seeding,
    seeding_table,
    inherits=Labour,
    polymorphic_identity=3,
)

Querys

ejemplo de como realizar una query

responsible = (
    # Objeto que quiero traer
    self.session.query(WorkOrderResponsible)
    # Quiero que la relacion staff se cargue con un join (pisa el valor cargado en el mappeo para esta consulta)
    .options(joinedload(WorkOrderResponsible.staff))
    # Todos los filtros que quiera separados por ',' es igual a 'and'
    .filter(WorkOrderResponsible.responsible_uuid == uuid, WorkOrderResponsible.deleted_at.is_(None))
    #obtener el primero que coincide, puede ser all() para traerse una lista de objetos
    .first()
)