42 lines
1.5 KiB
Python
42 lines
1.5 KiB
Python
from typing import Sequence
|
||
from typing_extensions import deprecated
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy import update, delete
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from db.models.recycling_parameters_model import RecyclingParameters
|
||
|
||
@deprecated(
|
||
"теперь есть параметризованный круд, а уже свои специфичные методы у каждой сущности в своем репозитории"
|
||
)
|
||
class RecyclingParametersRepository:
|
||
def __init__(self, session: AsyncSession):
|
||
self.session = session
|
||
|
||
async def get_all(self) -> Sequence[RecyclingParameters]:
|
||
result = await self.session.execute(select(RecyclingParameters))
|
||
return result.scalars().all()
|
||
|
||
async def get_by_id(self, id: int) -> RecyclingParameters:
|
||
result = await self.session.execute(select(RecyclingParameters).where(RecyclingParameters.id == id))
|
||
return result.scalar_one_or_none()
|
||
|
||
async def create(self, new_data: RecyclingParameters):
|
||
self.session.add(new_data)
|
||
await self.session.commit()
|
||
|
||
async def update(self, id: int, updated_data: dict):
|
||
stmt = (
|
||
update(RecyclingParameters).
|
||
where(RecyclingParameters.id == id).
|
||
values(**updated_data)
|
||
)
|
||
await self.session.execute(stmt)
|
||
await self.session.commit()
|
||
|
||
async def delete(self, id: int):
|
||
stmt = delete(RecyclingParameters).where(RecyclingParameters.id == id)
|
||
await self.session.execute(stmt)
|
||
await self.session.commit()
|