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

from typing import Optional
from sqlalchemy import select
from db.db_connection import async_session
from db.models.load_parameters_model import LoadParameters


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

    @staticmethod
    async def get_by_id(id_: int) -> Optional[LoadParameters]:
        async with async_session() as session:
            result = await session.execute(select(LoadParameters).where(LoadParameters.id == id_))
            return result.scalars().first()

    @staticmethod
    async def create(load: int,
                     primary_air_consumption: float,
                     secondary_air_consumption: float,
                     gas_inlet_consumption: float):
        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(new_data: LoadParameters):
        async with async_session() as session:
            session.add(new_data)
            await session.commit()