PIbd-42_SSPR/db/repositories/load_parameters_repos.py

from typing import Optional
from sqlalchemy.future import select
from db.db_connection import async_session
from db.models.load_parameters_model import LoadParameters
from network.schemas import LoadParametersBody


class LoadParametersRepository:
    @staticmethod
    async def get_all() -> Optional[list[LoadParameters]]:
        async with async_session() as session:
            result = await session.execute(select(LoadParameters))
            return result.scalars().all()

    @staticmethod
    async def get_by_id(id_: int) -> Optional[LoadParameters]:
        async with async_session() as session:
            result = session.get(LoadParameters, id_)
            return result

    @staticmethod
    async def create(load: int,
                     primary_air_consumption: float,
                     secondary_air_consumption: float,
                     gas_inlet_consumption: float) -> None:
        new_data = LoadParameters(
            load=load,
            primary_air_consumption=primary_air_consumption,
            secondary_air_consumption=secondary_air_consumption,
            gas_inlet_consumption=gas_inlet_consumption
        )
        async with async_session() as session:
            session.add(new_data)
            await session.commit()

    @staticmethod
    async def create_from_pydantic(body: LoadParametersBody):
        new_data = LoadParameters(
            load=body.load,
            primary_air_consumption=body.primary_air_consumption,
            secondary_air_consumption=body.secondary_air_consumption,
            gas_inlet_consumption=body.gas_inlet_consumption
        )
        async with async_session() as session:
            session.add(new_data)
            await session.commit()