Coverage for api/invite_requests.py: 31.07%
141 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-13 11:37 +0000
1"""
2Invite Request API endpoints for Missing Table
4Handles public invite request submissions and admin management.
5"""
7import os
8import sys
9from datetime import datetime
11import structlog
12from fastapi import APIRouter, Depends, HTTPException, Query
13from pydantic import BaseModel, EmailStr, Field
15from supabase import create_client
17logger = structlog.get_logger(__name__)
19sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
21from auth import get_current_user_required
22from dao.match_dao import SupabaseConnection as DbConnectionHolder
24# Initialize database connection with service role for admin operations
25supabase_url = os.getenv("SUPABASE_URL", "")
26service_key = os.getenv("SUPABASE_SERVICE_KEY")
28# Create service role client for operations that bypass RLS
29if service_key: 29 ↛ 33line 29 didn't jump to line 33 because the condition on line 29 was always true
30 service_client = create_client(supabase_url, service_key)
31else:
32 # Fallback to regular connection if service key not available
33 db_conn_holder_obj = DbConnectionHolder()
34 service_client = db_conn_holder_obj.client
36router = APIRouter(prefix="/api/invite-requests", tags=["invite-requests"])
39# Pydantic models
40class InviteRequestCreate(BaseModel):
41 """Model for creating a new invite request"""
43 email: EmailStr = Field(..., description="Email address of the requester")
44 name: str = Field(..., min_length=1, max_length=255, description="Name of the requester")
45 team: str | None = Field(None, max_length=255, description="Team or club affiliation")
46 reason: str | None = Field(None, description="Reason for wanting to join")
47 website: str | None = Field(None, description="Honeypot field - should be empty")
50class InviteRequestResponse(BaseModel):
51 """Model for invite request response"""
53 id: str
54 email: str
55 name: str
56 team: str | None
57 reason: str | None
58 status: str
59 created_at: datetime
60 updated_at: datetime
61 reviewed_by: str | None
62 reviewed_at: datetime | None
63 admin_notes: str | None
66class InviteRequestStatusUpdate(BaseModel):
67 """Model for updating invite request status"""
69 status: str = Field(..., pattern="^(pending|approved|rejected)$")
70 admin_notes: str | None = Field(None, description="Admin notes about the decision")
73class InviteRequestStats(BaseModel):
74 """Statistics about invite requests"""
76 total: int
77 pending: int
78 approved: int
79 rejected: int
82# Public endpoint - no auth required
83@router.post("", status_code=201)
84async def create_invite_request(request: InviteRequestCreate):
85 """
86 Submit a new invite request (public endpoint).
88 Anyone can submit an invite request without authentication.
89 Duplicate email submissions are allowed (users can re-submit).
90 """
91 try:
92 # Honeypot check - if filled, it's a bot
93 if request.website:
94 # Return fake success to not alert the bot
95 return {
96 "success": True,
97 "message": "Thank you for your interest! We'll review your request and reach out soon.",
98 }
100 # Check for existing pending request with same email
101 existing = (
102 service_client.table("invite_requests")
103 .select("id, status")
104 .eq("email", request.email)
105 .eq("status", "pending")
106 .execute()
107 )
109 if existing.data:
110 # Return success anyway - we don't want to reveal if email exists
111 return {
112 "success": True,
113 "message": "Thank you for your interest! We'll review your request and reach out soon.",
114 }
116 # Insert new invite request
117 result = (
118 service_client.table("invite_requests")
119 .insert(
120 {
121 "email": request.email,
122 "name": request.name,
123 "team": request.team,
124 "reason": request.reason,
125 "status": "pending",
126 }
127 )
128 .execute()
129 )
131 if result.data:
132 return {
133 "success": True,
134 "message": "Thank you for your interest! We'll review your request and reach out soon.",
135 }
136 else:
137 raise HTTPException(status_code=500, detail="Failed to submit invite request")
139 except HTTPException:
140 raise
141 except Exception:
142 logger.exception("Error creating invite request")
143 raise HTTPException(status_code=500, detail="An error occurred while submitting your request") from None
146# Admin endpoints
147@router.get("", response_model=list[InviteRequestResponse])
148async def list_invite_requests(
149 current_user=Depends(get_current_user_required),
150 status: str | None = Query(None, pattern="^(pending|approved|rejected)$"),
151 limit: int = Query(50, ge=1, le=100),
152 offset: int = Query(0, ge=0),
153):
154 """
155 List all invite requests (admin only).
157 Supports filtering by status and pagination.
158 """
159 if current_user.get("role") not in ["admin", "club_manager"]:
160 raise HTTPException(status_code=403, detail="Only admins can view invite requests")
162 try:
163 query = (
164 service_client.table("invite_requests")
165 .select("*")
166 .order("created_at", desc=True)
167 .range(offset, offset + limit - 1)
168 )
170 if status:
171 query = query.eq("status", status)
173 result = query.execute()
175 return result.data or []
177 except Exception:
178 logger.exception("Error listing invite requests")
179 raise HTTPException(status_code=500, detail="Failed to retrieve invite requests") from None
182@router.get("/stats", response_model=InviteRequestStats)
183async def get_invite_request_stats(current_user=Depends(get_current_user_required)):
184 """
185 Get statistics about invite requests (admin only).
186 """
187 if current_user.get("role") not in ["admin", "club_manager"]:
188 raise HTTPException(status_code=403, detail="Only admins can view invite request stats")
190 try:
191 # Get counts by status
192 all_requests = service_client.table("invite_requests").select("status").execute()
194 stats = {"total": 0, "pending": 0, "approved": 0, "rejected": 0}
196 if all_requests.data:
197 stats["total"] = len(all_requests.data)
198 for req in all_requests.data:
199 status = req.get("status", "pending")
200 if status in stats:
201 stats[status] += 1
203 return stats
205 except Exception:
206 logger.exception("Error getting invite request stats")
207 raise HTTPException(status_code=500, detail="Failed to retrieve statistics") from None
210@router.get("/{request_id}", response_model=InviteRequestResponse)
211async def get_invite_request(request_id: str, current_user=Depends(get_current_user_required)):
212 """
213 Get a specific invite request by ID (admin only).
214 """
215 if current_user.get("role") not in ["admin", "club_manager"]:
216 raise HTTPException(status_code=403, detail="Only admins can view invite requests")
218 try:
219 result = service_client.table("invite_requests").select("*").eq("id", request_id).execute()
221 if not result.data:
222 raise HTTPException(status_code=404, detail="Invite request not found")
224 return result.data[0]
226 except HTTPException:
227 raise
228 except Exception:
229 logger.exception("Error getting invite request")
230 raise HTTPException(status_code=500, detail="Failed to retrieve invite request") from None
233@router.put("/{request_id}/status")
234async def update_invite_request_status(
235 request_id: str,
236 status_update: InviteRequestStatusUpdate,
237 current_user=Depends(get_current_user_required),
238):
239 """
240 Update the status of an invite request (admin only).
242 Used to approve or reject invite requests.
243 """
244 if current_user.get("role") not in ["admin", "club_manager"]:
245 raise HTTPException(status_code=403, detail="Only admins can update invite requests")
247 try:
248 # Get current request
249 existing = service_client.table("invite_requests").select("*").eq("id", request_id).execute()
251 if not existing.data:
252 raise HTTPException(status_code=404, detail="Invite request not found")
254 # Get user ID
255 user_id = current_user.get("id") or current_user.get("user_id") or current_user.get("sub")
257 # Update the request
258 update_data = {
259 "status": status_update.status,
260 "reviewed_by": user_id,
261 "reviewed_at": datetime.utcnow().isoformat(),
262 }
264 if status_update.admin_notes:
265 update_data["admin_notes"] = status_update.admin_notes
267 result = service_client.table("invite_requests").update(update_data).eq("id", request_id).execute()
269 if result.data:
270 return {
271 "success": True,
272 "message": f"Invite request {status_update.status}",
273 "request": result.data[0],
274 }
275 else:
276 raise HTTPException(status_code=500, detail="Failed to update invite request")
278 except HTTPException:
279 raise
280 except Exception:
281 logger.exception("Error updating invite request")
282 raise HTTPException(status_code=500, detail="Failed to update invite request") from None
285@router.delete("/{request_id}")
286async def delete_invite_request(request_id: str, current_user=Depends(get_current_user_required)):
287 """
288 Delete an invite request (admin only).
290 Use with caution - this permanently removes the request.
291 """
292 if current_user.get("role") not in ["admin", "club_manager"]:
293 raise HTTPException(status_code=403, detail="Only admins can delete invite requests")
295 try:
296 # Check if request exists
297 existing = service_client.table("invite_requests").select("id").eq("id", request_id).execute()
299 if not existing.data:
300 raise HTTPException(status_code=404, detail="Invite request not found")
302 # Delete the request
303 service_client.table("invite_requests").delete().eq("id", request_id).execute()
305 return {"success": True, "message": "Invite request deleted successfully"}
307 except HTTPException:
308 raise
309 except Exception:
310 logger.exception("Error deleting invite request")
311 raise HTTPException(status_code=500, detail="Failed to delete invite request") from None