45 lines
1.4 KiB
Python
45 lines
1.4 KiB
Python
from sqlalchemy import select, func, Select
|
|
|
|
from models import Status, Deal
|
|
from repositories.base import BaseRepository
|
|
from repositories.mixins import (
|
|
RepDeleteMixin,
|
|
RepCreateMixin,
|
|
GetByIdMixin,
|
|
GetAllMixin,
|
|
RepUpdateMixin,
|
|
)
|
|
from schemas.status import UpdateStatusSchema, CreateStatusSchema
|
|
|
|
|
|
class StatusRepository(
|
|
BaseRepository,
|
|
GetAllMixin[Status],
|
|
RepDeleteMixin[Status],
|
|
RepCreateMixin[Status, CreateStatusSchema],
|
|
RepUpdateMixin[Status, UpdateStatusSchema],
|
|
GetByIdMixin[Status],
|
|
):
|
|
entity_class = Status
|
|
|
|
def _process_get_all_stmt_with_args(self, stmt: Select, *args) -> Select:
|
|
board_id = args[0]
|
|
return stmt.where(Status.board_id == board_id).order_by(Status.lexorank)
|
|
|
|
async def get_all(self, board_id: int) -> list[Status]:
|
|
stmt = (
|
|
select(Status)
|
|
.where(Status.is_deleted.is_(False), Status.board_id == board_id)
|
|
.order_by(Status.lexorank)
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_deals_count(self, status_id: int) -> int:
|
|
stmt = select(func.count(Deal.id)).where(Deal.status_id == status_id)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar()
|
|
|
|
async def update(self, status: Status, data: UpdateStatusSchema) -> Status:
|
|
return await self._apply_update_data_to_model(status, data, True)
|