261 lines
7.9 KiB
Python
261 lines
7.9 KiB
Python
from fastapi import APIRouter, HTTPException, Query
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.orm import selectinload
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.api.deps import DbSession, CurrentUser, require_admin
|
|
from app.models import User, UserRole, Marathon, Participant, Game
|
|
from app.schemas import UserPublic, MarathonListItem, MessageResponse
|
|
|
|
router = APIRouter(prefix="/admin", tags=["admin"])
|
|
|
|
|
|
class SetUserRole(BaseModel):
|
|
role: str = Field(..., pattern="^(user|admin)$")
|
|
|
|
|
|
class AdminUserResponse(BaseModel):
|
|
id: int
|
|
login: str
|
|
nickname: str
|
|
role: str
|
|
avatar_url: str | None = None
|
|
telegram_id: int | None = None
|
|
telegram_username: str | None = None
|
|
marathons_count: int = 0
|
|
created_at: str
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class AdminMarathonResponse(BaseModel):
|
|
id: int
|
|
title: str
|
|
status: str
|
|
creator: UserPublic
|
|
participants_count: int
|
|
games_count: int
|
|
start_date: str | None
|
|
end_date: str | None
|
|
created_at: str
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
@router.get("/users", response_model=list[AdminUserResponse])
|
|
async def list_users(
|
|
current_user: CurrentUser,
|
|
db: DbSession,
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(50, ge=1, le=100),
|
|
search: str | None = None,
|
|
):
|
|
"""List all users. Admin only."""
|
|
require_admin(current_user)
|
|
|
|
query = select(User).order_by(User.created_at.desc())
|
|
|
|
if search:
|
|
query = query.where(
|
|
(User.login.ilike(f"%{search}%")) |
|
|
(User.nickname.ilike(f"%{search}%"))
|
|
)
|
|
|
|
query = query.offset(skip).limit(limit)
|
|
result = await db.execute(query)
|
|
users = result.scalars().all()
|
|
|
|
response = []
|
|
for user in users:
|
|
# Count marathons user participates in
|
|
marathons_count = await db.scalar(
|
|
select(func.count()).select_from(Participant).where(Participant.user_id == user.id)
|
|
)
|
|
response.append(AdminUserResponse(
|
|
id=user.id,
|
|
login=user.login,
|
|
nickname=user.nickname,
|
|
role=user.role,
|
|
avatar_url=user.avatar_url,
|
|
telegram_id=user.telegram_id,
|
|
telegram_username=user.telegram_username,
|
|
marathons_count=marathons_count,
|
|
created_at=user.created_at.isoformat(),
|
|
))
|
|
|
|
return response
|
|
|
|
|
|
@router.get("/users/{user_id}", response_model=AdminUserResponse)
|
|
async def get_user(user_id: int, current_user: CurrentUser, db: DbSession):
|
|
"""Get user details. Admin only."""
|
|
require_admin(current_user)
|
|
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
marathons_count = await db.scalar(
|
|
select(func.count()).select_from(Participant).where(Participant.user_id == user.id)
|
|
)
|
|
|
|
return AdminUserResponse(
|
|
id=user.id,
|
|
login=user.login,
|
|
nickname=user.nickname,
|
|
role=user.role,
|
|
avatar_url=user.avatar_url,
|
|
telegram_id=user.telegram_id,
|
|
telegram_username=user.telegram_username,
|
|
marathons_count=marathons_count,
|
|
created_at=user.created_at.isoformat(),
|
|
)
|
|
|
|
|
|
@router.patch("/users/{user_id}/role", response_model=AdminUserResponse)
|
|
async def set_user_role(
|
|
user_id: int,
|
|
data: SetUserRole,
|
|
current_user: CurrentUser,
|
|
db: DbSession,
|
|
):
|
|
"""Set user's global role. Admin only."""
|
|
require_admin(current_user)
|
|
|
|
# Cannot change own role
|
|
if user_id == current_user.id:
|
|
raise HTTPException(status_code=400, detail="Cannot change your own role")
|
|
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
user.role = data.role
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
|
|
marathons_count = await db.scalar(
|
|
select(func.count()).select_from(Participant).where(Participant.user_id == user.id)
|
|
)
|
|
|
|
return AdminUserResponse(
|
|
id=user.id,
|
|
login=user.login,
|
|
nickname=user.nickname,
|
|
role=user.role,
|
|
avatar_url=user.avatar_url,
|
|
telegram_id=user.telegram_id,
|
|
telegram_username=user.telegram_username,
|
|
marathons_count=marathons_count,
|
|
created_at=user.created_at.isoformat(),
|
|
)
|
|
|
|
|
|
@router.delete("/users/{user_id}", response_model=MessageResponse)
|
|
async def delete_user(user_id: int, current_user: CurrentUser, db: DbSession):
|
|
"""Delete a user. Admin only."""
|
|
require_admin(current_user)
|
|
|
|
# Cannot delete yourself
|
|
if user_id == current_user.id:
|
|
raise HTTPException(status_code=400, detail="Cannot delete yourself")
|
|
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
# Cannot delete another admin
|
|
if user.role == UserRole.ADMIN.value:
|
|
raise HTTPException(status_code=400, detail="Cannot delete another admin")
|
|
|
|
await db.delete(user)
|
|
await db.commit()
|
|
|
|
return MessageResponse(message="User deleted")
|
|
|
|
|
|
@router.get("/marathons", response_model=list[AdminMarathonResponse])
|
|
async def list_marathons(
|
|
current_user: CurrentUser,
|
|
db: DbSession,
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(50, ge=1, le=100),
|
|
search: str | None = None,
|
|
):
|
|
"""List all marathons. Admin only."""
|
|
require_admin(current_user)
|
|
|
|
query = (
|
|
select(Marathon)
|
|
.options(selectinload(Marathon.creator))
|
|
.order_by(Marathon.created_at.desc())
|
|
)
|
|
|
|
if search:
|
|
query = query.where(Marathon.title.ilike(f"%{search}%"))
|
|
|
|
query = query.offset(skip).limit(limit)
|
|
result = await db.execute(query)
|
|
marathons = result.scalars().all()
|
|
|
|
response = []
|
|
for marathon in marathons:
|
|
participants_count = await db.scalar(
|
|
select(func.count()).select_from(Participant).where(Participant.marathon_id == marathon.id)
|
|
)
|
|
games_count = await db.scalar(
|
|
select(func.count()).select_from(Game).where(Game.marathon_id == marathon.id)
|
|
)
|
|
response.append(AdminMarathonResponse(
|
|
id=marathon.id,
|
|
title=marathon.title,
|
|
status=marathon.status,
|
|
creator=UserPublic.model_validate(marathon.creator),
|
|
participants_count=participants_count,
|
|
games_count=games_count,
|
|
start_date=marathon.start_date.isoformat() if marathon.start_date else None,
|
|
end_date=marathon.end_date.isoformat() if marathon.end_date else None,
|
|
created_at=marathon.created_at.isoformat(),
|
|
))
|
|
|
|
return response
|
|
|
|
|
|
@router.delete("/marathons/{marathon_id}", response_model=MessageResponse)
|
|
async def delete_marathon(marathon_id: int, current_user: CurrentUser, db: DbSession):
|
|
"""Delete a marathon. Admin only."""
|
|
require_admin(current_user)
|
|
|
|
result = await db.execute(select(Marathon).where(Marathon.id == marathon_id))
|
|
marathon = result.scalar_one_or_none()
|
|
if not marathon:
|
|
raise HTTPException(status_code=404, detail="Marathon not found")
|
|
|
|
await db.delete(marathon)
|
|
await db.commit()
|
|
|
|
return MessageResponse(message="Marathon deleted")
|
|
|
|
|
|
@router.get("/stats")
|
|
async def get_stats(current_user: CurrentUser, db: DbSession):
|
|
"""Get platform statistics. Admin only."""
|
|
require_admin(current_user)
|
|
|
|
users_count = await db.scalar(select(func.count()).select_from(User))
|
|
marathons_count = await db.scalar(select(func.count()).select_from(Marathon))
|
|
games_count = await db.scalar(select(func.count()).select_from(Game))
|
|
participants_count = await db.scalar(select(func.count()).select_from(Participant))
|
|
|
|
return {
|
|
"users_count": users_count,
|
|
"marathons_count": marathons_count,
|
|
"games_count": games_count,
|
|
"total_participations": participants_count,
|
|
}
|