Coverage for api/invites.py: 27.74%
216 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-13 00:07 +0000
1"""
2Invite API endpoints for Missing Table
3"""
5import os
6import sys
7from datetime import datetime
9import structlog
10from fastapi import APIRouter, Depends, HTTPException, Query
11from pydantic import BaseModel, Field
13sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
15logger = structlog.get_logger(__name__)
17from auth import get_current_user_required
18from dao.match_dao import SupabaseConnection as DbConnectionHolder
19from services import InviteService, TeamManagerService
20from supabase import create_client
22# Initialize database connection with service role for admin operations
23supabase_url = os.getenv("SUPABASE_URL", "")
24service_key = os.getenv("SUPABASE_SERVICE_KEY")
26# Create service role client for admin operations that bypass RLS
27if service_key: 27 ↛ 31line 27 didn't jump to line 31 because the condition on line 27 was always true
28 service_client = create_client(supabase_url, service_key)
29else:
30 # Fallback to regular connection if service key not available
31 db_conn_holder_obj = DbConnectionHolder()
32 service_client = db_conn_holder_obj.client
34router = APIRouter(prefix="/api/invites", tags=["invites"])
37# Pydantic models
38class CreateInviteRequest(BaseModel):
39 invite_type: str = Field(..., pattern="^(team_manager|team_player|team_fan)$")
40 team_id: int
41 age_group_id: int
42 email: str | None = None
43 player_id: int | None = None # Links to existing roster entry (for team_player invites)
44 jersey_number: int | None = Field(None, ge=1, le=99) # Creates roster entry on redemption
47class CreateClubManagerInviteRequest(BaseModel):
48 club_id: int
49 email: str | None = None
52class ClubManagerInviteResponse(BaseModel):
53 id: str
54 invite_code: str
55 invite_type: str
56 club_id: int
57 club_name: str | None
58 email: str | None
59 status: str
60 expires_at: datetime
61 created_at: datetime
64class InviteCodeValidation(BaseModel):
65 invite_code: str = Field(..., min_length=12, max_length=12)
68class InviteResponse(BaseModel):
69 id: str
70 invite_code: str
71 invite_type: str
72 team_id: int
73 team_name: str | None
74 age_group_id: int
75 age_group_name: str | None
76 email: str | None
77 status: str
78 expires_at: datetime
79 created_at: datetime
82# Public endpoint - no auth required
83@router.get("/validate/{invite_code}")
84async def validate_invite_code(invite_code: str):
85 """Validate an invite code without authentication"""
86 # Use service client for validation to read any invite code
87 invite_service = InviteService(service_client)
89 validation = invite_service.validate_invite_code(invite_code)
91 if not validation:
92 raise HTTPException(status_code=404, detail="Invalid or expired invite code")
94 return validation
97# Admin endpoints
98@router.post("/admin/club-manager")
99async def create_club_manager_invite(
100 request: CreateClubManagerInviteRequest, current_user=Depends(get_current_user_required)
101):
102 """Create a club manager invitation (admin only)"""
103 if current_user["role"] != "admin":
104 raise HTTPException(status_code=403, detail="Only admins can create club manager invites")
106 # Use service role client for admin operations to bypass RLS
107 invite_service = InviteService(service_client)
109 try:
110 # Handle different user ID field names
111 user_id = current_user.get("id") or current_user.get("user_id") or current_user.get("sub")
112 if not user_id:
113 raise HTTPException(status_code=400, detail=f"User ID not found in current_user: {current_user}")
115 invitation = invite_service.create_invitation(
116 invited_by_user_id=user_id,
117 invite_type="club_manager",
118 club_id=request.club_id,
119 email=request.email,
120 )
122 return invitation
124 except Exception as e:
125 logger.exception("Club manager invite creation error")
126 raise HTTPException(status_code=400, detail=str(e)) from e
129@router.post("/admin/team-manager")
130async def create_team_manager_invite(request: CreateInviteRequest, current_user=Depends(get_current_user_required)):
131 """Create a team manager invitation (admin only)"""
132 if current_user["role"] != "admin":
133 raise HTTPException(status_code=403, detail="Only admins can create team manager invites")
135 # Use service role client for admin operations to bypass RLS
136 invite_service = InviteService(service_client)
138 try:
139 logger.debug(
140 "Creating invite",
141 current_user=current_user,
142 team_id=request.team_id,
143 age_group_id=request.age_group_id,
144 email=request.email,
145 )
147 # Handle different user ID field names
148 user_id = current_user.get("id") or current_user.get("user_id") or current_user.get("sub")
149 if not user_id:
150 raise HTTPException(status_code=400, detail=f"User ID not found in current_user: {current_user}")
152 invitation = invite_service.create_invitation(
153 invited_by_user_id=user_id,
154 invite_type="team_manager",
155 team_id=request.team_id,
156 age_group_id=request.age_group_id,
157 email=request.email,
158 )
160 return invitation
162 except Exception as e:
163 logger.exception("Invite creation error")
164 raise HTTPException(status_code=400, detail=str(e)) from e
167@router.post("/admin/club-fan")
168async def create_club_fan_invite_admin(
169 request: CreateClubManagerInviteRequest, current_user=Depends(get_current_user_required)
170):
171 """Create a club fan invitation (admin only)"""
172 if current_user["role"] != "admin":
173 raise HTTPException(status_code=403, detail="Only admins can create club fan invites")
175 # Use service role client for admin operations to bypass RLS
176 invite_service = InviteService(service_client)
178 try:
179 # Handle different user ID field names
180 user_id = current_user.get("id") or current_user.get("user_id") or current_user.get("sub")
181 if not user_id:
182 raise HTTPException(status_code=400, detail=f"User ID not found in current_user: {current_user}")
184 invitation = invite_service.create_invitation(
185 invited_by_user_id=user_id,
186 invite_type="club_fan",
187 club_id=request.club_id,
188 email=request.email,
189 )
191 return invitation
193 except Exception as e:
194 logger.exception("Club fan invite creation error")
195 raise HTTPException(status_code=400, detail=str(e)) from e
198@router.post("/admin/team-fan")
199async def create_team_fan_invite_admin(request: CreateInviteRequest, current_user=Depends(get_current_user_required)):
200 """Create a team fan invitation (admin) - DEPRECATED: Use club-fan instead"""
201 if current_user["role"] != "admin":
202 raise HTTPException(status_code=403, detail="Unauthorized")
204 # Use service role client for admin operations to bypass RLS
205 invite_service = InviteService(service_client)
207 try:
208 # Handle different user ID field names
209 user_id = current_user.get("id") or current_user.get("user_id") or current_user.get("sub")
210 if not user_id:
211 raise HTTPException(status_code=400, detail=f"User ID not found in current_user: {current_user}")
213 invitation = invite_service.create_invitation(
214 invited_by_user_id=user_id,
215 invite_type="team_fan",
216 team_id=request.team_id,
217 age_group_id=request.age_group_id,
218 email=request.email,
219 )
221 return invitation
223 except Exception as e:
224 raise HTTPException(status_code=400, detail=str(e)) from e
227@router.post(
228 "/admin/team-player",
229)
230async def create_team_player_invite_admin(
231 request: CreateInviteRequest, current_user=Depends(get_current_user_required)
232):
233 """Create a team player invitation (admin).
235 If player_id is provided, the invitation will be linked to an existing roster entry.
236 If jersey_number is provided (without player_id), a roster entry will be created
237 when the player accepts the invite.
238 """
239 if current_user["role"] != "admin":
240 raise HTTPException(status_code=403, detail="Unauthorized")
242 # Use service role client for admin operations to bypass RLS
243 invite_service = InviteService(service_client)
245 try:
246 # Handle different user ID field names
247 user_id = current_user.get("id") or current_user.get("user_id") or current_user.get("sub")
248 if not user_id:
249 raise HTTPException(status_code=400, detail=f"User ID not found in current_user: {current_user}")
251 invitation = invite_service.create_invitation(
252 invited_by_user_id=user_id,
253 invite_type="team_player",
254 team_id=request.team_id,
255 age_group_id=request.age_group_id,
256 email=request.email,
257 player_id=request.player_id,
258 jersey_number=request.jersey_number,
259 )
261 return invitation
263 except Exception as e:
264 raise HTTPException(status_code=400, detail=str(e)) from e
267# Club manager endpoints
268@router.post("/club-manager/club-fan")
269async def create_club_fan_invite_club_manager(
270 request: CreateClubManagerInviteRequest, current_user=Depends(get_current_user_required)
271):
272 """Create a club fan invitation (club manager or admin)"""
273 if current_user["role"] not in ["admin", "club_manager"]:
274 raise HTTPException(status_code=403, detail="Only club managers or admins can create club fan invites")
276 # Use service role client for operations to bypass RLS
277 invite_service = InviteService(service_client)
279 try:
280 # Handle different user ID field names
281 user_id = current_user.get("id") or current_user.get("user_id") or current_user.get("sub")
282 if not user_id:
283 raise HTTPException(status_code=400, detail=f"User ID not found in current_user: {current_user}")
285 # Club managers can only create invites for their own club
286 if current_user["role"] == "club_manager":
287 user_club_id = current_user.get("club_id")
288 if not user_club_id or user_club_id != request.club_id:
289 raise HTTPException(status_code=403, detail="You can only create fan invites for your own club")
291 invitation = invite_service.create_invitation(
292 invited_by_user_id=user_id,
293 invite_type="club_fan",
294 club_id=request.club_id,
295 email=request.email,
296 )
298 return invitation
300 except Exception as e:
301 logger.exception("Club fan invite creation error")
302 raise HTTPException(status_code=400, detail=str(e)) from e
305# Team manager endpoints
306@router.post("/team-manager/team-fan")
307async def create_team_fan_invite(request: CreateInviteRequest, current_user=Depends(get_current_user_required)):
308 """Create a team fan invitation (team manager) - DEPRECATED: Use club-fan instead"""
309 if current_user["role"] not in ["admin", "team-manager", "team_manager"]:
310 raise HTTPException(status_code=403, detail="Unauthorized")
312 supabase = service_client
313 team_manager_service = TeamManagerService(supabase)
315 # Handle different user ID field names
316 user_id = current_user.get("id") or current_user.get("user_id") or current_user.get("sub")
317 if not user_id:
318 raise HTTPException(status_code=400, detail=f"User ID not found in current_user: {current_user}")
320 # Check if team manager can manage this team
321 if current_user["role"] in ["team_manager", "team-manager"]:
322 can_manage = team_manager_service.can_manage_team(user_id, request.team_id, request.age_group_id)
324 if not can_manage:
325 raise HTTPException(status_code=403, detail="You can only create invites for teams you manage")
327 invite_service = InviteService(supabase)
329 try:
330 invitation = invite_service.create_invitation(
331 invited_by_user_id=user_id,
332 invite_type="team_fan",
333 team_id=request.team_id,
334 age_group_id=request.age_group_id,
335 email=request.email,
336 )
338 return invitation
340 except Exception as e:
341 raise HTTPException(status_code=400, detail=str(e)) from e
344@router.post(
345 "/team-manager/team-player",
346)
347async def create_team_player_invite(request: CreateInviteRequest, current_user=Depends(get_current_user_required)):
348 """Create a team player invitation (team manager).
350 If player_id is provided, the invitation will be linked to an existing roster entry.
351 If jersey_number is provided (without player_id), a roster entry will be created
352 when the player accepts the invite.
353 """
354 if current_user["role"] not in ["admin", "team-manager", "team_manager"]:
355 raise HTTPException(status_code=403, detail="Unauthorized")
357 supabase = service_client
358 team_manager_service = TeamManagerService(supabase)
360 # Handle different user ID field names
361 user_id = current_user.get("id") or current_user.get("user_id") or current_user.get("sub")
362 if not user_id:
363 raise HTTPException(status_code=400, detail=f"User ID not found in current_user: {current_user}")
365 # Check if team manager can manage this team
366 if current_user["role"] in ["team_manager", "team-manager"]:
367 can_manage = team_manager_service.can_manage_team(user_id, request.team_id, request.age_group_id)
369 if not can_manage:
370 raise HTTPException(status_code=403, detail="You can only create invites for teams you manage")
372 invite_service = InviteService(supabase)
374 try:
375 invitation = invite_service.create_invitation(
376 invited_by_user_id=user_id,
377 invite_type="team_player",
378 team_id=request.team_id,
379 age_group_id=request.age_group_id,
380 email=request.email,
381 player_id=request.player_id,
382 jersey_number=request.jersey_number,
383 )
385 return invitation
387 except Exception as e:
388 raise HTTPException(status_code=400, detail=str(e)) from e
391# List user's invitations
392@router.get(
393 "/my-invites",
394)
395async def get_my_invitations(
396 current_user=Depends(get_current_user_required),
397 status: str | None = Query(None, pattern="^(pending|used|expired)$"),
398):
399 """Get all invitations created by the current user"""
400 supabase = service_client
401 invite_service = InviteService(supabase)
403 # Handle different user ID field names
404 user_id = current_user.get("id") or current_user.get("user_id") or current_user.get("sub")
405 if not user_id:
406 raise HTTPException(status_code=400, detail=f"User ID not found in current_user: {current_user}")
408 invitations = invite_service.get_user_invitations(user_id)
410 # Filter by status if provided
411 if status:
412 invitations = [inv for inv in invitations if inv["status"] == status]
414 return invitations
417# Cancel invitation
418@router.delete(
419 "/{invite_id}",
420)
421async def cancel_invitation(invite_id: str, current_user=Depends(get_current_user_required)):
422 """Cancel a pending invitation"""
423 supabase = service_client
424 invite_service = InviteService(supabase)
426 # Check if user owns this invitation or is admin
427 invitations = invite_service.get_user_invitations(current_user["user_id"])
428 user_owns_invite = any(inv["id"] == invite_id for inv in invitations)
430 if not user_owns_invite and current_user["role"] != "admin":
431 raise HTTPException(status_code=403, detail="You can only cancel your own invitations")
433 success = invite_service.cancel_invitation(invite_id, current_user["user_id"])
435 if not success:
436 raise HTTPException(status_code=404, detail="Invitation not found or already cancelled")
438 return {"message": "Invitation cancelled successfully"}
441# Team manager assignments endpoint
442@router.get(
443 "/team-manager/assignments",
444)
445async def get_team_manager_assignments(current_user=Depends(get_current_user_required)):
446 """Get team assignments for the current user"""
447 if current_user["role"] not in ["admin", "team-manager", "team_manager"]:
448 raise HTTPException(status_code=403, detail="Unauthorized")
450 supabase = service_client
451 team_manager_service = TeamManagerService(supabase)
453 assignments = team_manager_service.get_user_team_assignments(current_user["id"])
455 return assignments