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

38 lines
1.3 KiB
Python
Raw Normal View History

from typing import Optional
from sqlalchemy import select
from db.db_connection import async_session
from db.models.recycling_parameters_model import RecyclingParameters
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()