56 lines
1.8 KiB
Python
56 lines
1.8 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from db.models.tickets import Ticket, TicketStatus
|
|
from schemas.tickets import NewTicketDTO
|
|
|
|
|
|
class TicketRepository:
|
|
async def get_ticket_by_user_id(
|
|
self, session: AsyncSession, user_id: int, status: TicketStatus | None = None
|
|
) -> list[Ticket]:
|
|
stmt = select(Ticket).where(Ticket.creator_id == user_id)
|
|
if status:
|
|
stmt = stmt.where(Ticket.status == status)
|
|
res = await session.execute(stmt)
|
|
|
|
return list(res.scalars().all())
|
|
|
|
async def get_available_ticket_by_user_id(
|
|
self, session: AsyncSession, user_id: int
|
|
) -> Ticket | None:
|
|
stmt = (
|
|
select(Ticket)
|
|
.where(Ticket.creator_id == user_id)
|
|
.where(Ticket.status != TicketStatus.CLOSED)
|
|
)
|
|
res = await session.execute(stmt)
|
|
|
|
tickets = list(res.scalars().all())
|
|
return tickets[0] if tickets else None
|
|
|
|
async def get_ticket_by_thread_id(self, session: AsyncSession, thread_id: int) -> Ticket | None:
|
|
stmt = select(Ticket).where(Ticket.thread_id == thread_id)
|
|
res = await session.execute(stmt)
|
|
|
|
return res.scalar_one_or_none()
|
|
|
|
async def create_ticket(self, session: AsyncSession, ticket_data: NewTicketDTO) -> Ticket:
|
|
ticket = Ticket(creator_id=ticket_data.creator_id)
|
|
session.add(ticket)
|
|
await session.commit()
|
|
|
|
return ticket
|
|
|
|
async def update_ticket_status(
|
|
self, session: AsyncSession, ticket: Ticket, status: TicketStatus
|
|
):
|
|
ticket.status = status
|
|
await session.commit()
|
|
return ticket
|
|
|
|
async def update_ticket_thread_id(self, session: AsyncSession, ticket: Ticket, thread_id: int):
|
|
ticket.thread_id = thread_id
|
|
await session.commit()
|
|
return ticket
|