AnnZhimol
49e327005e
+ch_experiment_data in click_house +experiment_data and experiment_parameters in postgresql +experiment_data has foreign_key from experiment_parameters +new connections +realize routes for ch_experiment_data +new alg csv_to_db +new methods in repos +update dockerfile +update readme "how to init db"
50 lines
1.8 KiB
Python
50 lines
1.8 KiB
Python
from typing import Optional
|
|
from sqlalchemy.future import select
|
|
from db.postgres_db_connection import async_session_postgres
|
|
from db.models.load_parameters_model import LoadParameters
|
|
from network.schemas import LoadParametersBody
|
|
|
|
|
|
class LoadParametersRepository:
|
|
@staticmethod
|
|
async def get_all() -> Optional[list[LoadParameters]]:
|
|
async with async_session_postgres() as session:
|
|
result = await session.execute(select(LoadParameters))
|
|
return result.scalars().all()
|
|
|
|
@staticmethod
|
|
async def get_by_id(id_: int) -> Optional[LoadParameters]:
|
|
async with async_session_postgres() as session:
|
|
result = session.get(LoadParameters, id_)
|
|
return result
|
|
|
|
@staticmethod
|
|
async def create(load: int,
|
|
primary_air_consumption: float,
|
|
secondary_air_consumption: float,
|
|
gas_inlet_consumption: float) -> None:
|
|
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_postgres() as session:
|
|
session.add(new_data)
|
|
await session.commit()
|
|
|
|
@staticmethod
|
|
async def create_from_pydantic(body: LoadParametersBody):
|
|
new_data = LoadParameters(
|
|
load=body.load,
|
|
primary_air_consumption=body.primary_air_consumption,
|
|
secondary_air_consumption=body.secondary_air_consumption,
|
|
gas_inlet_consumption=body.gas_inlet_consumption
|
|
)
|
|
async with async_session_postgres() as session:
|
|
session.add(new_data)
|
|
await session.commit()
|
|
|
|
|
|
|