159 lines
4.6 KiB
Python
159 lines
4.6 KiB
Python
"""Room management router endpoints."""
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, insert, update, delete
|
|
from typing import Optional, List
|
|
|
|
from app.core.database import get_db
|
|
from app.core.security import get_current_user
|
|
from app.models.models import Room
|
|
from app.schemas.schemas import RoomResponse
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/", response_model=List[RoomResponse])
|
|
async def list_rooms(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""List all available rooms."""
|
|
stmt = select(Room).order_by(Room.id)
|
|
result = await db.execute(stmt)
|
|
rooms = result.scalars().all()
|
|
|
|
return [RoomResponse.model_validate(room) for room in rooms]
|
|
|
|
|
|
@router.get("/{room_id}", response_model=RoomResponse)
|
|
async def get_room(
|
|
room_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Get room by ID."""
|
|
stmt = select(Room).where(Room.id == room_id)
|
|
result = await db.execute(stmt)
|
|
room = result.scalar_one_or_none()
|
|
|
|
if not room:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Room not found",
|
|
)
|
|
|
|
return RoomResponse.model_validate(room)
|
|
|
|
|
|
@router.post("/")
|
|
async def create_room(
|
|
name: str,
|
|
description: Optional[str] = None,
|
|
is_password_protected: bool = False,
|
|
password_hash: Optional[str] = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Create a new room (admin only)."""
|
|
# Check if current user is admin
|
|
if current_user.get("privlevel") not in ["Admin", "Judge"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin privileges required",
|
|
)
|
|
|
|
# Check if room name already exists
|
|
stmt = select(Room).where(Room.name == name)
|
|
result = await db.execute(stmt)
|
|
if result.scalar_one_or_none():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Room name already exists",
|
|
)
|
|
|
|
new_room = Room(
|
|
name=name,
|
|
description=description,
|
|
is_password_protected=is_password_protected,
|
|
password_hash=password_hash,
|
|
)
|
|
db.add(new_room)
|
|
await db.flush()
|
|
|
|
return {"message": f"Room '{name}' created successfully", "room_id": new_room.id}
|
|
|
|
|
|
@router.patch("/{room_id}")
|
|
async def update_room(
|
|
room_id: int,
|
|
name: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
is_password_protected: Optional[bool] = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Update room (admin only)."""
|
|
# Check if current user is admin
|
|
if current_user.get("privlevel") not in ["Admin", "Judge"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin privileges required",
|
|
)
|
|
|
|
stmt = select(Room).where(Room.id == room_id)
|
|
result = await db.execute(stmt)
|
|
room = result.scalar_one_or_none()
|
|
|
|
if not room:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Room not found",
|
|
)
|
|
|
|
# Update fields
|
|
update_data = {}
|
|
if name:
|
|
update_data["name"] = name
|
|
if description is not None:
|
|
update_data["description"] = description
|
|
if is_password_protected is not None:
|
|
update_data["is_password_protected"] = is_password_protected
|
|
|
|
stmt = (
|
|
update(Room)
|
|
.where(Room.id == room_id)
|
|
.values(**update_data)
|
|
)
|
|
await db.execute(stmt)
|
|
|
|
return {"message": f"Room '{name or room.name}' updated successfully"}
|
|
|
|
|
|
@router.delete("/{room_id}")
|
|
async def delete_room(
|
|
room_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""Delete room (admin only)."""
|
|
# Check if current user is admin
|
|
if current_user.get("privlevel") not in ["Admin", "Judge"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin privileges required",
|
|
)
|
|
|
|
stmt = select(Room).where(Room.id == room_id)
|
|
result = await db.execute(stmt)
|
|
room = result.scalar_one_or_none()
|
|
|
|
if not room:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Room not found",
|
|
)
|
|
|
|
await db.execute(delete(Room).where(Room.id == room_id))
|
|
|
|
return {"message": f"Room '{room.name}' deleted successfully"}
|