39 lines
1.2 KiB
Python
39 lines
1.2 KiB
Python
from typing import Optional
|
|
from sqlalchemy import select
|
|
from db.db_connection import async_session
|
|
from db.models.experiment_data_model import ExperimentData
|
|
|
|
|
|
class ExperimentDataRepository:
|
|
@staticmethod
|
|
async def get_all() -> Optional[list[ExperimentData]]:
|
|
async with async_session() as session:
|
|
result = await session.execute(select(ExperimentData))
|
|
return result.scalar.all()
|
|
|
|
@staticmethod
|
|
async def get_by_id(id_: int) -> Optional[ExperimentData]:
|
|
async with async_session() as session:
|
|
result = await session.execute(select(ExperimentData).where(ExperimentData.id == id_))
|
|
return result.scalars().first()
|
|
|
|
|
|
@staticmethod
|
|
async def create(direction: float,
|
|
temperature: float,
|
|
nox: float,
|
|
co2: float,
|
|
co: float,
|
|
file_id: str):
|
|
new_data = ExperimentData(
|
|
direction=direction,
|
|
temperature=temperature,
|
|
nox=nox,
|
|
co2=co2,
|
|
co=co,
|
|
file_id=file_id
|
|
)
|
|
async with async_session() as session:
|
|
session.add(new_data)
|
|
await session.commit()
|