215 lines
6.9 KiB
Python
215 lines
6.9 KiB
Python
import logging
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from bson import ObjectId
|
|
|
|
from src.db.repositories.team_repository import team_repository
|
|
from src.schemas.team import TeamCreate, TeamUpdate, TeamResponse, TeamListResponse
|
|
from src.db.models.team import TeamModel
|
|
from src.api.v1.auth import get_current_user
|
|
from src.core.logging import log_request
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(tags=["Teams"], prefix="/teams")
|
|
|
|
@router.post("", response_model=TeamResponse, status_code=201)
|
|
async def create_team(team_data: TeamCreate, request: Request, current_user = Depends(get_current_user)):
|
|
"""
|
|
Create a new team
|
|
|
|
This endpoint requires admin privileges
|
|
"""
|
|
log_request(
|
|
{"path": request.url.path, "method": request.method, "team_data": team_data.dict()},
|
|
user_id=str(current_user.id),
|
|
team_id=str(current_user.team_id)
|
|
)
|
|
|
|
# Only admins can create teams
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Only admins can create teams")
|
|
|
|
# Create team
|
|
team = TeamModel(
|
|
name=team_data.name,
|
|
description=team_data.description
|
|
)
|
|
|
|
created_team = await team_repository.create(team)
|
|
|
|
# Convert to response model
|
|
response = TeamResponse(
|
|
id=str(created_team.id),
|
|
name=created_team.name,
|
|
description=created_team.description,
|
|
created_at=created_team.created_at,
|
|
updated_at=created_team.updated_at
|
|
)
|
|
|
|
return response
|
|
|
|
@router.get("", response_model=TeamListResponse)
|
|
async def list_teams(request: Request, current_user = Depends(get_current_user)):
|
|
"""
|
|
List all teams
|
|
|
|
This endpoint requires admin privileges
|
|
"""
|
|
log_request(
|
|
{"path": request.url.path, "method": request.method},
|
|
user_id=str(current_user.id),
|
|
team_id=str(current_user.team_id)
|
|
)
|
|
|
|
# Only admins can list all teams
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Only admins can list all teams")
|
|
|
|
# Get all teams
|
|
teams = await team_repository.get_all()
|
|
|
|
# Convert to response models
|
|
response_teams = []
|
|
for team in teams:
|
|
response_teams.append(TeamResponse(
|
|
id=str(team.id),
|
|
name=team.name,
|
|
description=team.description,
|
|
created_at=team.created_at,
|
|
updated_at=team.updated_at
|
|
))
|
|
|
|
return TeamListResponse(teams=response_teams, total=len(response_teams))
|
|
|
|
@router.get("/{team_id}", response_model=TeamResponse)
|
|
async def get_team(team_id: str, request: Request, current_user = Depends(get_current_user)):
|
|
"""
|
|
Get a team by ID
|
|
|
|
Users can only access their own team unless they are an admin
|
|
"""
|
|
log_request(
|
|
{"path": request.url.path, "method": request.method, "team_id": team_id},
|
|
user_id=str(current_user.id),
|
|
team_id=str(current_user.team_id)
|
|
)
|
|
|
|
try:
|
|
# Convert string ID to ObjectId
|
|
obj_id = ObjectId(team_id)
|
|
except:
|
|
raise HTTPException(status_code=400, detail="Invalid team ID")
|
|
|
|
# Get the team
|
|
team = await team_repository.get_by_id(obj_id)
|
|
if not team:
|
|
raise HTTPException(status_code=404, detail="Team not found")
|
|
|
|
# Check if user can access this team
|
|
if str(team.id) != str(current_user.team_id) and not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Not authorized to access this team")
|
|
|
|
# Convert to response model
|
|
response = TeamResponse(
|
|
id=str(team.id),
|
|
name=team.name,
|
|
description=team.description,
|
|
created_at=team.created_at,
|
|
updated_at=team.updated_at
|
|
)
|
|
|
|
return response
|
|
|
|
@router.put("/{team_id}", response_model=TeamResponse)
|
|
async def update_team(team_id: str, team_data: TeamUpdate, request: Request, current_user = Depends(get_current_user)):
|
|
"""
|
|
Update a team
|
|
|
|
This endpoint requires admin privileges
|
|
"""
|
|
log_request(
|
|
{"path": request.url.path, "method": request.method, "team_id": team_id, "team_data": team_data.dict()},
|
|
user_id=str(current_user.id),
|
|
team_id=str(current_user.team_id)
|
|
)
|
|
|
|
# Only admins can update teams
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Only admins can update teams")
|
|
|
|
try:
|
|
# Convert string ID to ObjectId
|
|
obj_id = ObjectId(team_id)
|
|
except:
|
|
raise HTTPException(status_code=400, detail="Invalid team ID")
|
|
|
|
# Get the team
|
|
team = await team_repository.get_by_id(obj_id)
|
|
if not team:
|
|
raise HTTPException(status_code=404, detail="Team not found")
|
|
|
|
# Update the team
|
|
update_data = team_data.dict(exclude_unset=True)
|
|
if not update_data:
|
|
# No fields to update
|
|
return TeamResponse(
|
|
id=str(team.id),
|
|
name=team.name,
|
|
description=team.description,
|
|
created_at=team.created_at,
|
|
updated_at=team.updated_at
|
|
)
|
|
|
|
updated_team = await team_repository.update(obj_id, update_data)
|
|
if not updated_team:
|
|
raise HTTPException(status_code=500, detail="Failed to update team")
|
|
|
|
# Convert to response model
|
|
response = TeamResponse(
|
|
id=str(updated_team.id),
|
|
name=updated_team.name,
|
|
description=updated_team.description,
|
|
created_at=updated_team.created_at,
|
|
updated_at=updated_team.updated_at
|
|
)
|
|
|
|
return response
|
|
|
|
@router.delete("/{team_id}", status_code=204)
|
|
async def delete_team(team_id: str, request: Request, current_user = Depends(get_current_user)):
|
|
"""
|
|
Delete a team
|
|
|
|
This endpoint requires admin privileges
|
|
"""
|
|
log_request(
|
|
{"path": request.url.path, "method": request.method, "team_id": team_id},
|
|
user_id=str(current_user.id),
|
|
team_id=str(current_user.team_id)
|
|
)
|
|
|
|
# Only admins can delete teams
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Only admins can delete teams")
|
|
|
|
try:
|
|
# Convert string ID to ObjectId
|
|
obj_id = ObjectId(team_id)
|
|
except:
|
|
raise HTTPException(status_code=400, detail="Invalid team ID")
|
|
|
|
# Check if team exists
|
|
team = await team_repository.get_by_id(obj_id)
|
|
if not team:
|
|
raise HTTPException(status_code=404, detail="Team not found")
|
|
|
|
# Don't allow deleting a user's own team
|
|
if str(team.id) == str(current_user.team_id):
|
|
raise HTTPException(status_code=400, detail="Cannot delete your own team")
|
|
|
|
# Delete the team
|
|
result = await team_repository.delete(obj_id)
|
|
if not result:
|
|
raise HTTPException(status_code=500, detail="Failed to delete team")
|
|
|
|
return None |