132 lines
4.6 KiB
Python
132 lines
4.6 KiB
Python
from fastapi import APIRouter, Body, HTTPException, Request
|
|
|
|
from models.user import CreateUser
|
|
from models.team import TeamBase, Team
|
|
from lib.db import get_connection
|
|
from lib.helpers import hash_password
|
|
from lib.logger import logger
|
|
|
|
router = APIRouter(prefix="/admin", tags=["Admin"])
|
|
|
|
|
|
# @router.patch("/contest/status")
|
|
# async def set_contest_status(is_active: bool, session: Session = Depends(get_session)):
|
|
# """
|
|
# Set the contest status (active/inactive)
|
|
# Only accessible to admin users
|
|
# """
|
|
# # Implementation will be added with authentication logic
|
|
# return {"is_active": is_active}
|
|
|
|
|
|
@router.post("/teams")
|
|
async def create_team(team: TeamBase, request: Request):
|
|
logger.info(
|
|
f"Admin action: Creating new team '{team.name}' from IP: {request.client.host}"
|
|
)
|
|
try:
|
|
with get_connection() as conn:
|
|
conn.execute("INSERT INTO Team (name) VALUES (?)", (team.name,))
|
|
logger.info(f"Team '{team.name}' created successfully")
|
|
return {"message": "Team created successfully"}
|
|
except Exception as e:
|
|
logger.error(f"Failed to create team '{team.name}': {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Failed to create team")
|
|
|
|
|
|
@router.get("/teams", response_model=list[Team])
|
|
async def list_teams(request: Request):
|
|
logger.debug(f"Admin action: Listing all teams from IP: {request.client.host}")
|
|
with get_connection() as conn:
|
|
teams = conn.execute("SELECT * FROM Team").fetchdf()
|
|
return teams.to_dict(orient="records")
|
|
|
|
|
|
@router.post("/users")
|
|
async def create_user(user: CreateUser, request: Request):
|
|
logger.info(
|
|
f"Admin action: Creating new user '{user.username}' from IP: {request.client.host}"
|
|
)
|
|
try:
|
|
with get_connection() as conn:
|
|
# Check if username already exists
|
|
existing = conn.execute(
|
|
"SELECT COUNT(*) as count FROM User WHERE username = ?",
|
|
(user.username,),
|
|
).fetchdf()
|
|
if existing["count"][0] > 0:
|
|
logger.warning(
|
|
f"Failed to create user: Username '{user.username}' already exists"
|
|
)
|
|
raise HTTPException(status_code=400, detail="Username already exists")
|
|
|
|
conn.execute(
|
|
"INSERT INTO User (username, hashed_password, team_id, is_admin) VALUES (?, ?, ?, ?)",
|
|
(
|
|
user.username,
|
|
hash_password(user.password),
|
|
user.team_id,
|
|
user.is_admin,
|
|
),
|
|
)
|
|
logger.info(
|
|
f"User '{user.username}' created successfully with team_id: {user.team_id}, admin status: {user.is_admin}"
|
|
)
|
|
return {"message": "User created successfully"}
|
|
except HTTPException:
|
|
# Re-raise HTTP exceptions
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Failed to create user '{user.username}': {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Failed to create user")
|
|
|
|
|
|
@router.post("/users/disable")
|
|
async def disable_user(user_id: int = Body(..., embed=True), request: Request = None):
|
|
logger.info(
|
|
f"Admin action: Attempting to disable user with ID: {user_id} from IP: {request.client.host}"
|
|
)
|
|
logger.warning(f"Disable user functionality not implemented for user_id: {user_id}")
|
|
raise HTTPException(status_code=405, detail="Method not yet implemented")
|
|
|
|
|
|
# """
|
|
# Add a new user
|
|
# Only accessible to admin users
|
|
# """
|
|
# # Implementation will be added with proper password hashing
|
|
# return {"username": username, "group_id": group_id}
|
|
|
|
# @router.get("/users")
|
|
# async def list_users(group_id: int = None, is_active: bool = None, session: Session = Depends(get_session)):
|
|
# """
|
|
# List users with optional filters
|
|
# Only accessible to admin users
|
|
# """
|
|
# # Implementation for filtering users
|
|
# return {"users": []}
|
|
|
|
# @router.get("/users/{user_id}")
|
|
# async def get_user(user_id: int, session: Session = Depends(get_session)):
|
|
# """
|
|
# Get details for a specific user
|
|
# Only accessible to admin users
|
|
# """
|
|
# # Implementation for getting user details
|
|
# return {"user_id": user_id}
|
|
|
|
# @router.patch("/users/{user_id}")
|
|
# async def update_user(
|
|
# user_id: int,
|
|
# group_id: int = None,
|
|
# password: str = None,
|
|
# is_active: bool = None,
|
|
# session: Session = Depends(get_session)
|
|
# ):
|
|
# """
|
|
# Update a user's details
|
|
# Only accessible to admin users
|
|
# """
|
|
# # Implementation for updating user details
|
|
# return {"user_id": user_id}
|