54 lines
1.6 KiB
Python
54 lines
1.6 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
|
|
from network.schemas import ExperimentDataBody
|
|
|
|
|
|
class ExperimentDataRepository:
|
|
@staticmethod
|
|
async def get_all() -> Optional[list[ExperimentData]]:
|
|
async with async_session() as session:
|
|
result = await session.execute(select(ExperimentData))
|
|
return result.scalars().all()
|
|
|
|
@staticmethod
|
|
async def get_by_id(id_: int) -> Optional[ExperimentData]:
|
|
async with async_session() as session:
|
|
result = session.get(ExperimentData, id_)
|
|
return result
|
|
|
|
|
|
@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()
|
|
|
|
@staticmethod
|
|
async def create_from_pydantic(body: ExperimentDataBody):
|
|
new_data = ExperimentData(
|
|
direction=body.direction,
|
|
temperature=body.temperature,
|
|
nox=body.nox,
|
|
co2=body.co2,
|
|
co=body.co,
|
|
file_id=body.file_id
|
|
)
|
|
async with async_session() as session:
|
|
session.add(new_data)
|
|
await session.commit()
|