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

from typing import Optional
from sqlalchemy import select
from db.db_connection import async_session
from db.models.recycling_parameters_model import RecyclingParameters
from network.schemas import RecyclingParametersBody


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

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

    @staticmethod
    async def create(load_id: int,
                     recycling_level: int,
                     co2: float,
                     n2: float,
                     h2o: float,
                     o2: float):
        new_data = RecyclingParameters(
            load_id=load_id,
            recycling_level=recycling_level,
            co2=co2,
            n2=n2,
            h2o=h2o,
            o2=o2
        )
        async with async_session() as session:
            session.add(new_data)
            await session.commit()

    @staticmethod
    async def create_from_pydantic(body: RecyclingParametersBody):
        new_data = RecyclingParameters(
            load_id=body.load_id,
            recycling_level=body.recycling_level,
            co2=body.co2,
            n2=body.n2,
            h2o=body.h2o,
            o2=body.o2
        )
        async with async_session() as session:
            session.add(new_data)
            await session.commit()