43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
from typing import Sequence, Optional
|
|
|
|
from sqlalchemy import update, delete
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.future import select
|
|
|
|
from db.models.experiment_data_model import ExperimentData
|
|
|
|
|
|
|
|
|
|
|
|
class ExperimentDataRepository:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def get_all(self) -> Sequence[ExperimentData]:
|
|
result = await self.session.execute(select(ExperimentData))
|
|
return result.scalars().all()
|
|
|
|
async def get_by_id(self, id: int) -> ExperimentData:
|
|
result = await self.session.execute(select(ExperimentData).where(ExperimentData.id == id))
|
|
return result.scalar_one_or_none()
|
|
|
|
async def create(self, new_data: ExperimentData):
|
|
self.session.add(new_data)
|
|
await self.session.commit()
|
|
|
|
async def update(self, id: int, updated_data: dict):
|
|
stmt = (
|
|
update(ExperimentData).
|
|
where(ExperimentData.id == id).
|
|
values(**updated_data)
|
|
)
|
|
await self.session.execute(stmt)
|
|
await self.session.commit()
|
|
|
|
async def delete(self, id: int):
|
|
stmt = delete(ExperimentData).where(ExperimentData.id == id)
|
|
await self.session.execute(stmt)
|
|
await self.session.commit()
|
|
|