Coverage for dao/player_dao.py: 21.09%
215 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-15 13:38 +0000
1"""
2Player Data Access Object.
4Handles all database operations related to players/user profiles including:
5- User profile CRUD operations
6- Team-player associations
7- Player team history
8- Admin player management
9"""
11import structlog
13from dao.base_dao import BaseDAO, dao_cache, invalidates_cache
15logger = structlog.get_logger()
17# Cache patterns for invalidation
18PLAYERS_CACHE_PATTERN = "mt:dao:players:*"
21class PlayerDAO(BaseDAO):
22 """Data access object for player/user profile operations."""
24 # === User Profile Query Methods ===
26 @dao_cache("players:profile:{user_id}")
27 def get_user_profile_with_relationships(self, user_id: str) -> dict | None:
28 """
29 Get user profile with team relationship.
31 Args:
32 user_id: User ID to fetch profile for
34 Returns:
35 User profile dict with team data, or None if not found
36 """
37 try:
38 response = (
39 self.client.table("user_profiles")
40 .select("""
41 *,
42 team:teams(id, name, city, club:clubs(id, name, primary_color, secondary_color, logo_url)),
43 club:clubs(id, name, primary_color, secondary_color, logo_url)
44 """)
45 .eq("id", user_id)
46 .execute()
47 )
49 if response.data and len(response.data) > 0:
50 profile = response.data[0]
51 if len(response.data) > 1:
52 logger.warning(f"Multiple profiles found for user {user_id}, using first one")
53 return profile
54 return None
55 except Exception as e:
56 logger.error(f"Error fetching user profile: {e}")
57 return None
59 def get_user_profile_by_email(self, email: str, exclude_user_id: str | None = None) -> dict | None:
60 """
61 Get user profile by email, optionally excluding a specific user ID.
63 Useful for checking if an email is already in use by another user.
64 No caching - used for auth/validation.
66 Args:
67 email: Email address to search for
68 exclude_user_id: Optional user ID to exclude from search
70 Returns:
71 User profile dict if found, None otherwise
72 """
73 try:
74 query = self.client.table("user_profiles").select("id").eq("email", email)
75 if exclude_user_id:
76 query = query.neq("id", exclude_user_id)
77 response = query.execute()
78 if response.data and len(response.data) > 0:
79 return response.data[0]
80 return None
81 except Exception as e:
82 logger.error(f"Error checking user profile by email: {e}")
83 return None
85 def get_user_profile_by_username(self, username: str, exclude_user_id: str | None = None) -> dict | None:
86 """
87 Get user profile by username, optionally excluding a specific user ID.
89 Useful for checking if a username is already taken.
90 No caching - used for auth/validation.
92 Args:
93 username: Username to search for (will be lowercased)
94 exclude_user_id: Optional user ID to exclude from search
96 Returns:
97 User profile dict if found, None otherwise
98 """
99 try:
100 query = self.client.table("user_profiles").select("id").eq("username", username.lower())
101 if exclude_user_id:
102 query = query.neq("id", exclude_user_id)
103 response = query.execute()
104 if response.data and len(response.data) > 0:
105 return response.data[0]
106 return None
107 except Exception as e:
108 logger.error(f"Error checking user profile by username: {e}")
109 return None
111 def get_user_for_password_reset(self, identifier: str) -> dict | None:
112 """
113 Look up a user by username or email for the password reset flow.
115 If ``@`` appears in the identifier it is treated as an email address;
116 otherwise it is treated as a username (lowercased).
118 Selects only ``id``, ``username``, and ``email`` — no caching because
119 this is used in an auth-sensitive path.
121 Args:
122 identifier: Username or email address supplied by the user
124 Returns:
125 Dict with ``id``, ``username``, ``email`` keys, or None if not found
126 """
127 try:
128 if "@" in identifier:
129 query = self.client.table("user_profiles").select("id, username, email").eq("email", identifier)
130 else:
131 query = (
132 self.client.table("user_profiles")
133 .select("id, username, email")
134 .eq("username", identifier.lower())
135 )
136 response = query.execute()
137 if response.data and len(response.data) > 0:
138 return response.data[0]
139 return None
140 except Exception as e:
141 logger.error(f"Error looking up user for password reset: {e}")
142 return None
144 @dao_cache("players:all")
145 def get_all_user_profiles(self) -> list[dict]:
146 """
147 Get all user profiles with team relationships.
149 Returns:
150 List of user profile dicts
151 """
152 try:
153 response = (
154 self.client.table("user_profiles")
155 .select("""
156 *,
157 team:teams(id, name, city)
158 """)
159 .order("created_at", desc=True)
160 .execute()
161 )
162 return response.data or []
163 except Exception as e:
164 logger.error(f"Error fetching all user profiles: {e}")
165 return []
167 # === User Profile CRUD Methods ===
169 @invalidates_cache(PLAYERS_CACHE_PATTERN)
170 def create_or_update_user_profile(self, profile_data: dict) -> dict | None:
171 """
172 Create or update a user profile.
174 Args:
175 profile_data: Dictionary containing user profile data
177 Returns:
178 Created/updated profile dict, or None on error
179 """
180 try:
181 response = self.client.table("user_profiles").upsert(profile_data).execute()
182 if response.data and len(response.data) > 0:
183 return response.data[0]
184 return None
185 except Exception as e:
186 logger.error(f"Error creating/updating user profile: {e}")
187 return None
189 @invalidates_cache(PLAYERS_CACHE_PATTERN)
190 def update_user_profile(self, user_id: str, update_data: dict) -> dict | None:
191 """
192 Update user profile fields.
194 Args:
195 user_id: User ID to update
196 update_data: Dictionary of fields to update
198 Returns:
199 Updated profile dict, or None on error
200 """
201 try:
202 response = self.client.table("user_profiles").update(update_data).eq("id", user_id).execute()
203 if response.data and len(response.data) > 0:
204 return response.data[0]
205 return None
206 except Exception as e:
207 logger.error(f"Error updating user profile: {e}")
208 return None
210 # === Team Players Methods ===
212 @dao_cache("players:by_team:{team_id}")
213 def get_team_players(self, team_id: int) -> list[dict]:
214 """
215 Get all players currently on a team for the team roster page.
217 Uses player_team_history to support multi-team players.
219 Returns player profiles with fields needed for player cards:
220 - id, display_name, player_number, positions
221 - photo_1_url, photo_2_url, photo_3_url, profile_photo_slot
222 - overlay_style, primary_color, text_color, accent_color
223 - instagram_handle, snapchat_handle, tiktok_handle
225 Args:
226 team_id: The team ID to get players for
228 Returns:
229 List of player profile dicts
230 """
231 try:
232 # Query player_team_history for current team members
233 response = (
234 self.client.table("player_team_history")
235 .select("""
236 player_id,
237 jersey_number,
238 positions,
239 user_profiles!player_team_history_player_id_fkey(
240 id,
241 display_name,
242 player_number,
243 positions,
244 photo_1_url,
245 photo_2_url,
246 photo_3_url,
247 profile_photo_slot,
248 overlay_style,
249 primary_color,
250 text_color,
251 accent_color,
252 instagram_handle,
253 snapchat_handle,
254 tiktok_handle
255 )
256 """)
257 .eq("team_id", team_id)
258 .eq("is_current", True)
259 .execute()
260 )
262 # Flatten the results - extract user_profiles and merge with history data
263 players = []
264 for entry in response.data or []:
265 profile = entry.get("user_profiles")
266 if profile:
267 # Use jersey_number from history if available, fallback to profile
268 player = {**profile}
269 if entry.get("jersey_number") is not None:
270 player["player_number"] = entry["jersey_number"]
271 # Use positions from history if available, fallback to profile
272 if entry.get("positions"):
273 player["positions"] = entry["positions"]
274 players.append(player)
276 # Sort by player_number
277 players.sort(key=lambda p: p.get("player_number") or 999)
278 return players
279 except Exception as e:
280 logger.error(f"Error fetching team players for team {team_id}: {e}")
281 return []
283 # === Player Team History Methods ===
285 @dao_cache("players:history:{player_id}")
286 def get_player_team_history(self, player_id: str) -> list[dict]:
287 """
288 Get complete team history for a player across all seasons.
290 Returns history entries ordered by season (most recent first),
291 with full team, season, age_group, league, and division details.
293 Args:
294 player_id: User ID of the player
296 Returns:
297 List of history entry dicts with related data
298 """
299 try:
300 response = (
301 self.client.table("player_team_history")
302 .select("""
303 *,
304 team:teams(id, name, city,
305 club:clubs(id, name, primary_color, secondary_color)
306 ),
307 season:seasons(id, name, start_date, end_date),
308 age_group:age_groups(id, name),
309 league:leagues(id, name),
310 division:divisions(id, name)
311 """)
312 .eq("player_id", player_id)
313 .order("season_id", desc=True)
314 .execute()
315 )
316 return response.data or []
317 except Exception as e:
318 logger.error(f"Error fetching player team history: {e}")
319 return []
321 def get_current_player_team_assignment(self, player_id: str) -> dict | None:
322 """
323 Get the current team assignment for a player (is_current=true).
325 No caching - this is used for real-time checks.
327 Args:
328 player_id: User ID of the player
330 Returns:
331 Current history entry dict with related data, or None if not found
332 """
333 try:
334 response = (
335 self.client.table("player_team_history")
336 .select("""
337 *,
338 team:teams(id, name, city,
339 club:clubs(id, name, primary_color, secondary_color)
340 ),
341 season:seasons(id, name, start_date, end_date),
342 age_group:age_groups(id, name),
343 league:leagues(id, name),
344 division:divisions(id, name)
345 """)
346 .eq("player_id", player_id)
347 .eq("is_current", True)
348 .limit(1)
349 .execute()
350 )
352 if response.data and len(response.data) > 0:
353 return response.data[0]
354 return None
355 except Exception as e:
356 logger.error(f"Error fetching current player team assignment: {e}")
357 return None
359 def get_all_current_player_teams(self, player_id: str) -> list[dict]:
360 """
361 Get ALL current team assignments for a player (is_current=true).
363 This supports players being on multiple teams simultaneously
364 (e.g., for futsal/soccer leagues).
365 No caching - this is used for real-time checks.
367 Args:
368 player_id: User ID of the player
370 Returns:
371 List of current history entries with related team and club data
372 """
373 try:
374 response = (
375 self.client.table("player_team_history")
376 .select("""
377 *,
378 team:teams(id, name, city,
379 club:clubs(id, name, logo_url, primary_color, secondary_color),
380 league:leagues(id, name),
381 division:divisions(id, name)
382 ),
383 season:seasons(id, name, start_date, end_date),
384 age_group:age_groups(id, name)
385 """)
386 .eq("player_id", player_id)
387 .eq("is_current", True)
388 .execute()
389 )
390 return response.data or []
391 except Exception as e:
392 logger.error(f"Error fetching all current player teams: {e}")
393 return []
395 @invalidates_cache(PLAYERS_CACHE_PATTERN)
396 def create_player_history_entry(
397 self,
398 player_id: str,
399 team_id: int,
400 season_id: int,
401 jersey_number: int | None = None,
402 positions: list[str] | None = None,
403 notes: str | None = None,
404 is_current: bool = False,
405 ) -> dict | None:
406 """
407 Create or update a player team history entry.
409 Uses UPSERT to handle the unique constraint on (player_id, team_id, season_id).
410 If an entry already exists for the same player/team/season, it updates that
411 entry instead of failing. This supports:
412 - Re-adding a player to a team they were previously on
413 - Updating is_current flag on existing entries
414 - Players on multiple teams simultaneously (futsal use case)
416 Args:
417 player_id: User ID of the player
418 team_id: Team ID
419 season_id: Season ID
420 jersey_number: Optional jersey number for that season
421 positions: Optional list of positions played
422 notes: Optional notes about the assignment
423 is_current: Whether this is the current assignment
425 Returns:
426 Created/updated history entry dict, or None on error
427 """
428 try:
429 # Get team details for league_id, division_id
430 team_response = (
431 self.client.table("teams").select("league_id, division_id").eq("id", team_id).execute()
432 )
434 team_data = team_response.data[0] if team_response.data else {}
436 # Get age_group_id from team_mappings (teams.age_group_id is deprecated/never populated)
437 mapping_response = (
438 self.client.table("team_mappings").select("age_group_id").eq("team_id", team_id).limit(1).execute()
439 )
440 age_group_id = mapping_response.data[0]["age_group_id"] if mapping_response.data else None
442 upsert_data = {
443 "player_id": player_id,
444 "team_id": team_id,
445 "season_id": season_id,
446 "age_group_id": age_group_id,
447 "league_id": team_data.get("league_id"),
448 "division_id": team_data.get("division_id"),
449 "jersey_number": jersey_number,
450 "positions": positions,
451 "is_current": is_current,
452 "notes": notes,
453 }
455 # Use upsert to handle existing entries - if (player_id, team_id, season_id)
456 # already exists, update it instead of failing
457 response = (
458 self.client.table("player_team_history")
459 .upsert(upsert_data, on_conflict="player_id,team_id,season_id")
460 .execute()
461 )
462 if response.data and len(response.data) > 0:
463 return self.get_player_history_entry_by_id(response.data[0]["id"])
464 return None
465 except Exception as e:
466 logger.error(f"Error creating/updating player history entry: {e}")
467 return None
469 def get_player_history_entry_by_id(self, history_id: int) -> dict | None:
470 """
471 Get a single player history entry by ID with related data.
473 Args:
474 history_id: History entry ID
476 Returns:
477 History entry dict with related data, or None if not found
478 """
479 try:
480 response = (
481 self.client.table("player_team_history")
482 .select("""
483 *,
484 team:teams(id, name, city,
485 club:clubs(id, name, primary_color, secondary_color)
486 ),
487 season:seasons(id, name, start_date, end_date),
488 age_group:age_groups(id, name),
489 league:leagues(id, name),
490 division:divisions(id, name)
491 """)
492 .eq("id", history_id)
493 .execute()
494 )
496 if response.data and len(response.data) > 0:
497 return response.data[0]
498 return None
499 except Exception as e:
500 logger.error(f"Error fetching player history entry: {e}")
501 return None
503 @invalidates_cache(PLAYERS_CACHE_PATTERN)
504 def update_player_history_entry(
505 self,
506 history_id: int,
507 jersey_number: int | None = None,
508 positions: list[str] | None = None,
509 notes: str | None = None,
510 is_current: bool | None = None,
511 ) -> dict | None:
512 """
513 Update a player team history entry.
515 Args:
516 history_id: History entry ID to update
517 jersey_number: Optional new jersey number
518 positions: Optional new positions list
519 notes: Optional new notes
520 is_current: Optional new is_current flag
522 Returns:
523 Updated history entry dict, or None on error
524 """
525 try:
526 update_data = {"updated_at": "now()"}
528 if jersey_number is not None:
529 update_data["jersey_number"] = jersey_number
530 if positions is not None:
531 update_data["positions"] = positions
532 if notes is not None:
533 update_data["notes"] = notes
534 if is_current is not None:
535 update_data["is_current"] = is_current
536 # Note: We allow multiple current teams (for futsal/multi-league players)
537 # So we don't automatically unset is_current on other entries
539 response = self.client.table("player_team_history").update(update_data).eq("id", history_id).execute()
541 if response.data and len(response.data) > 0:
542 return self.get_player_history_entry_by_id(history_id)
543 return None
544 except Exception as e:
545 logger.error(f"Error updating player history entry: {e}")
546 return None
548 @invalidates_cache(PLAYERS_CACHE_PATTERN)
549 def delete_player_history_entry(self, history_id: int) -> bool:
550 """
551 Delete a player team history entry.
553 Args:
554 history_id: History entry ID to delete
556 Returns:
557 True if deleted successfully, False otherwise
558 """
559 try:
560 self.client.table("player_team_history").delete().eq("id", history_id).execute()
561 return True
562 except Exception as e:
563 logger.error(f"Error deleting player history entry: {e}")
564 return False
566 # === Admin Player Management Methods ===
568 def get_all_players_admin(
569 self,
570 search: str | None = None,
571 club_id: int | None = None,
572 team_id: int | None = None,
573 limit: int = 50,
574 offset: int = 0,
575 ) -> dict:
576 """
577 Get all players with their current team assignments for admin management.
578 No caching - admin queries should always be fresh.
580 Args:
581 search: Optional text search on display_name or email
582 club_id: Optional filter by club ID
583 team_id: Optional filter by team ID
584 limit: Max number of results (default 50)
585 offset: Offset for pagination (default 0)
587 Returns:
588 Dict with 'players' list and 'total' count
589 """
590 try:
591 # Build the base query for players (team-player role)
592 query = (
593 self.client.table("user_profiles")
594 .select(
595 """
596 id,
597 email,
598 display_name,
599 player_number,
600 positions,
601 photo_1_url,
602 profile_photo_slot,
603 team_id,
604 created_at,
605 team:teams(id, name, club_id)
606 """,
607 count="exact",
608 )
609 .eq("role", "team-player")
610 )
612 # Apply text search filter
613 if search:
614 query = query.or_(f"display_name.ilike.%{search}%,email.ilike.%{search}%")
616 # Apply team filter
617 if team_id:
618 query = query.eq("team_id", team_id)
619 # Apply club filter (via team)
620 elif club_id:
621 # Need to get team IDs for this club first
622 teams_response = self.client.table("teams").select("id").eq("club_id", club_id).execute()
623 if teams_response.data:
624 team_ids = [t["id"] for t in teams_response.data]
625 query = query.in_("team_id", team_ids)
627 # Apply pagination and ordering
628 query = query.order("display_name").range(offset, offset + limit - 1)
630 response = query.execute()
632 players = response.data or []
633 total = response.count or 0
635 # Enrich with current team assignments from player_team_history
636 for player in players:
637 history_response = (
638 self.client.table("player_team_history")
639 .select(
640 """
641 id,
642 team_id,
643 season_id,
644 jersey_number,
645 is_current,
646 created_at,
647 team:teams(id, name, club:clubs(id, name)),
648 season:seasons(id, name)
649 """
650 )
651 .eq("player_id", player["id"])
652 .eq("is_current", True)
653 .execute()
654 )
655 player["current_teams"] = history_response.data or []
657 return {"players": players, "total": total}
659 except Exception as e:
660 logger.error(f"Error fetching players for admin: {e}")
661 return {"players": [], "total": 0}
663 @invalidates_cache(PLAYERS_CACHE_PATTERN)
664 def update_player_admin(
665 self,
666 player_id: str,
667 display_name: str | None = None,
668 player_number: int | None = None,
669 positions: list[str] | None = None,
670 ) -> dict | None:
671 """
672 Update player profile info (admin/manager operation).
674 Args:
675 player_id: User ID of the player
676 display_name: Optional new display name
677 player_number: Optional new jersey number
678 positions: Optional new positions list
680 Returns:
681 Updated player profile dict, or None on error
682 """
683 try:
684 update_data = {"updated_at": "now()"}
686 if display_name is not None:
687 update_data["display_name"] = display_name
688 if player_number is not None:
689 update_data["player_number"] = player_number
690 if positions is not None:
691 update_data["positions"] = positions
693 response = self.client.table("user_profiles").update(update_data).eq("id", player_id).execute()
695 if response.data and len(response.data) > 0:
696 return response.data[0]
697 return None
699 except Exception as e:
700 logger.error(f"Error updating player admin: {e}")
701 return None
703 @invalidates_cache(PLAYERS_CACHE_PATTERN)
704 def end_player_team_assignment(self, history_id: int) -> dict | None:
705 """
706 End a player's team assignment by setting is_current=false.
708 Args:
709 history_id: Player team history ID
711 Returns:
712 Updated history entry dict, or None on error
713 """
714 try:
715 update_data = {"is_current": False, "updated_at": "now()"}
717 response = self.client.table("player_team_history").update(update_data).eq("id", history_id).execute()
719 if response.data and len(response.data) > 0:
720 return self.get_player_history_entry_by_id(history_id)
721 return None
723 except Exception as e:
724 logger.error(f"Error ending player team assignment: {e}")
725 return None