54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
from fastapi import Depends, HTTPException, status
|
|
from typing import Annotated
|
|
from sqlalchemy.orm import Session
|
|
from uuid import UUID
|
|
|
|
from ...dependencies import get_db
|
|
from ...db import models
|
|
|
|
from ..auth import services as auth_services
|
|
from ..auth import schemas as auth_schemas
|
|
|
|
from . import schemas
|
|
|
|
|
|
async def get_user_queues(
|
|
current_user: Annotated[auth_schemas.User, Depends(auth_services.get_current_user)]
|
|
) -> list[schemas.QueueInDb]:
|
|
return [schemas.QueueInDb.model_validate(q) for q in current_user.owns_queues]
|
|
|
|
|
|
async def create_queue(
|
|
new_queue: schemas.Queue,
|
|
current_user: auth_schemas.UserInDB,
|
|
db: Session,
|
|
) -> schemas.QueueInDb:
|
|
q = models.Queue(
|
|
name=new_queue.name, description=new_queue.description, owner_id=current_user.id
|
|
)
|
|
db.add(q)
|
|
db.commit()
|
|
return schemas.QueueInDb.model_validate(q)
|
|
|
|
|
|
async def get_detailed_queue(
|
|
queue_id: UUID,
|
|
db: Annotated[Session, Depends(get_db)],
|
|
) -> schemas.QueueDetail:
|
|
q = db.query(models.Queue).filter(models.Queue.id == queue_id).first()
|
|
print("\n\n", queue_id, "\n\n", flush=True)
|
|
if q:
|
|
return schemas.QueueDetail(
|
|
id=q.id,
|
|
name=q.name,
|
|
description=q.description,
|
|
participants=schemas.ParticipantInfo(
|
|
total=q.users.count(),
|
|
remaining=q.users.filter(models.QueueUser.passed == False).count(),
|
|
),
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Not Found",
|
|
)
|