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"
38 lines
1.4 KiB
Python
38 lines
1.4 KiB
Python
from db.models.ch_experimentdb_experiment_data_model import ChExperimentDBExperimentData
|
|
from sqlalchemy import select,func
|
|
|
|
class ChExperimentDBExperimentDataRepository:
|
|
def __init__(self, session):
|
|
self.session = session
|
|
|
|
def get_all(self):
|
|
return self.session.query(ChExperimentDBExperimentData).all()
|
|
|
|
def get_by_id(self, id: int) -> ChExperimentDBExperimentData:
|
|
return self.session.query(ChExperimentDBExperimentData).filter(ChExperimentDBExperimentData.id == id).one_or_none()
|
|
|
|
def create(self, ch_experiment_data: ChExperimentDBExperimentData) -> ChExperimentDBExperimentData:
|
|
max_id_query = select(func.max(ChExperimentDBExperimentData.id))
|
|
max_id_result = self.session.execute(max_id_query).scalar()
|
|
|
|
max_id = max_id_result or 0
|
|
|
|
ch_experiment_data.id = max_id + 1
|
|
|
|
self.session.add(ch_experiment_data)
|
|
self.session.commit()
|
|
|
|
return ch_experiment_data
|
|
|
|
def update(self, id: int, updated_data: dict) -> ChExperimentDBExperimentData:
|
|
self.session.query(ChExperimentDBExperimentData).filter(ChExperimentDBExperimentData.id == id).update(updated_data)
|
|
self.session.commit()
|
|
return self.get_by_id(id)
|
|
|
|
def delete(self, id: int) -> bool:
|
|
item = self.get_by_id(id)
|
|
if item:
|
|
self.session.delete(item)
|
|
self.session.commit()
|
|
return True
|
|
return False |