Coverage for dao/roster_dao.py: 11.46%
189 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-13 14:26 +0000
1"""
2Roster Data Access Object.
4Handles all database operations related to team rosters (players table):
5- Roster CRUD operations
6- Jersey number management
7- Account linking
8- Display name computation
9"""
11import structlog
13from dao.base_dao import BaseDAO, dao_cache, invalidates_cache
15logger = structlog.get_logger()
17# Cache patterns for invalidation
18ROSTER_CACHE_PATTERN = "mt:dao:roster:*"
21class RosterDAO(BaseDAO):
22 """Data access object for roster (players table) operations."""
24 # === Read Operations ===
26 @dao_cache("roster:team:{team_id}:season:{season_id}")
27 def get_team_roster(self, team_id: int, season_id: int) -> list[dict]:
28 """
29 Get all roster entries for a team in a specific season.
31 Args:
32 team_id: Team ID
33 season_id: Season ID
35 Returns:
36 List of player dicts with computed display_name
37 """
38 try:
39 # Use explicit FK relationship to avoid ambiguity with created_by FK
40 response = (
41 self.client.table("players")
42 .select("""
43 *,
44 user_profile:user_profiles!players_user_profile_id_fkey(
45 id, display_name, first_name, last_name,
46 photo_1_url, photo_2_url, photo_3_url, profile_photo_slot)
47 """)
48 .eq("team_id", team_id)
49 .eq("season_id", season_id)
50 .eq("is_active", True)
51 .order("jersey_number")
52 .execute()
53 )
55 players = response.data or []
56 # Add computed display_name to each player
57 return [self._add_display_name(p) for p in players]
59 except Exception as e:
60 logger.error("roster_get_team_error", team_id=team_id, season_id=season_id, error=str(e))
61 return []
63 def get_player_by_id(self, player_id: int) -> dict | None:
64 """
65 Get a single roster entry by ID.
67 Args:
68 player_id: Player ID
70 Returns:
71 Player dict with computed display_name, or None if not found
72 """
73 try:
74 response = (
75 self.client.table("players")
76 .select("""
77 *,
78 user_profile:user_profiles!players_user_profile_id_fkey(
79 id, display_name, first_name, last_name,
80 photo_1_url, photo_2_url, photo_3_url, profile_photo_slot)
81 """)
82 .eq("id", player_id)
83 .execute()
84 )
86 if response.data and len(response.data) > 0:
87 return self._add_display_name(response.data[0])
88 return None
90 except Exception as e:
91 logger.error("roster_get_player_error", player_id=player_id, error=str(e))
92 return None
94 def get_player_by_jersey(self, team_id: int, season_id: int, jersey_number: int) -> dict | None:
95 """
96 Get roster entry by jersey number (unique within team/season).
98 Args:
99 team_id: Team ID
100 season_id: Season ID
101 jersey_number: Jersey number to look up
103 Returns:
104 Player dict with computed display_name, or None if not found
105 """
106 try:
107 response = (
108 self.client.table("players")
109 .select("""
110 *,
111 user_profile:user_profiles!players_user_profile_id_fkey(
112 id, display_name, first_name, last_name,
113 photo_1_url, photo_2_url, photo_3_url, profile_photo_slot)
114 """)
115 .eq("team_id", team_id)
116 .eq("season_id", season_id)
117 .eq("jersey_number", jersey_number)
118 .execute()
119 )
121 if response.data and len(response.data) > 0:
122 return self._add_display_name(response.data[0])
123 return None
125 except Exception as e:
126 logger.error(
127 "roster_get_by_jersey_error",
128 team_id=team_id,
129 season_id=season_id,
130 jersey_number=jersey_number,
131 error=str(e),
132 )
133 return None
135 def get_player_by_user_profile_id(
136 self,
137 user_profile_id: str,
138 team_id: int | None = None,
139 season_id: int | None = None,
140 ) -> dict | None:
141 """
142 Get roster entry linked to a user profile.
144 When team_id and season_id are provided, returns the specific
145 player entry for that team/season. Otherwise, returns the most
146 recent active player entry.
148 Args:
149 user_profile_id: User profile ID (UUID)
150 team_id: Optional team ID filter
151 season_id: Optional season ID filter
153 Returns:
154 Player dict with computed display_name, or None if not found
155 """
156 try:
157 query = (
158 self.client.table("players")
159 .select("""
160 *,
161 user_profile:user_profiles!players_user_profile_id_fkey(
162 id, display_name, first_name, last_name,
163 photo_1_url, photo_2_url, photo_3_url, profile_photo_slot)
164 """)
165 .eq("user_profile_id", user_profile_id)
166 .eq("is_active", True)
167 )
169 if team_id is not None:
170 query = query.eq("team_id", team_id)
171 if season_id is not None:
172 query = query.eq("season_id", season_id)
174 # Order by created_at descending to get most recent
175 response = query.order("created_at", desc=True).limit(1).execute()
177 if response.data and len(response.data) > 0:
178 return self._add_display_name(response.data[0])
179 return None
181 except Exception as e:
182 logger.error(
183 "roster_get_by_user_profile_error",
184 user_profile_id=user_profile_id,
185 team_id=team_id,
186 season_id=season_id,
187 error=str(e),
188 )
189 return None
191 # === Create Operations ===
193 @invalidates_cache(ROSTER_CACHE_PATTERN)
194 def create_player(
195 self,
196 team_id: int,
197 season_id: int,
198 jersey_number: int,
199 first_name: str | None = None,
200 last_name: str | None = None,
201 positions: list[str] | None = None,
202 created_by: str | None = None,
203 ) -> dict | None:
204 """
205 Create a new roster entry.
207 Args:
208 team_id: Team ID
209 season_id: Season ID
210 jersey_number: Jersey number (1-99, unique per team/season)
211 first_name: Optional first name
212 last_name: Optional last name
213 positions: Optional list of position codes
214 created_by: Optional user ID of creator
216 Returns:
217 Created player dict, or None on error
218 """
219 try:
220 data = {
221 "team_id": team_id,
222 "season_id": season_id,
223 "jersey_number": jersey_number,
224 "is_active": True,
225 }
226 if first_name:
227 data["first_name"] = first_name
228 if last_name:
229 data["last_name"] = last_name
230 if positions:
231 data["positions"] = positions
232 if created_by:
233 data["created_by"] = created_by
235 response = self.client.table("players").insert(data).execute()
237 if response.data and len(response.data) > 0:
238 logger.info(
239 "roster_player_created",
240 player_id=response.data[0]["id"],
241 team_id=team_id,
242 jersey_number=jersey_number,
243 )
244 return self._add_display_name(response.data[0])
245 return None
247 except Exception as e:
248 logger.error("roster_create_error", team_id=team_id, jersey_number=jersey_number, error=str(e))
249 return None
251 @invalidates_cache(ROSTER_CACHE_PATTERN)
252 def bulk_create_players(
253 self,
254 team_id: int,
255 season_id: int,
256 players: list[dict],
257 created_by: str | None = None,
258 ) -> list[dict]:
259 """
260 Create multiple roster entries at once.
262 Args:
263 team_id: Team ID
264 season_id: Season ID
265 players: List of dicts with jersey_number, optional first_name, last_name, positions
266 created_by: Optional user ID of creator
268 Returns:
269 List of created player dicts
270 """
271 try:
272 data = []
273 for p in players:
274 entry = {
275 "team_id": team_id,
276 "season_id": season_id,
277 "jersey_number": p["jersey_number"],
278 "is_active": True,
279 }
280 if p.get("first_name"):
281 entry["first_name"] = p["first_name"]
282 if p.get("last_name"):
283 entry["last_name"] = p["last_name"]
284 if p.get("positions"):
285 entry["positions"] = p["positions"]
286 if created_by:
287 entry["created_by"] = created_by
288 data.append(entry)
290 response = self.client.table("players").insert(data).execute()
292 created = response.data or []
293 logger.info("roster_bulk_created", team_id=team_id, count=len(created))
294 return [self._add_display_name(p) for p in created]
296 except Exception as e:
297 logger.error("roster_bulk_create_error", team_id=team_id, count=len(players), error=str(e))
298 return []
300 # === Update Operations ===
302 @invalidates_cache(ROSTER_CACHE_PATTERN)
303 def update_player(
304 self,
305 player_id: int,
306 first_name: str | None = None,
307 last_name: str | None = None,
308 positions: list[str] | None = None,
309 ) -> dict | None:
310 """
311 Update a roster entry's name or positions.
313 Args:
314 player_id: Player ID
315 first_name: New first name (None to keep current)
316 last_name: New last name (None to keep current)
317 positions: New positions (None to keep current)
319 Returns:
320 Updated player dict, or None on error
321 """
322 try:
323 data = {}
324 if first_name is not None:
325 data["first_name"] = first_name or None
326 if last_name is not None:
327 data["last_name"] = last_name or None
328 if positions is not None:
329 data["positions"] = positions or []
331 if not data:
332 # Nothing to update
333 return self.get_player_by_id(player_id)
335 response = self.client.table("players").update(data).eq("id", player_id).execute()
337 if response.data and len(response.data) > 0:
338 logger.info("roster_player_updated", player_id=player_id)
339 return self._add_display_name(response.data[0])
340 return None
342 except Exception as e:
343 logger.error("roster_update_error", player_id=player_id, error=str(e))
344 return None
346 @invalidates_cache(ROSTER_CACHE_PATTERN)
347 def update_jersey_number(
348 self,
349 player_id: int,
350 new_number: int,
351 ) -> dict | None:
352 """
353 Change a player's jersey number.
355 Args:
356 player_id: Player ID
357 new_number: New jersey number (1-99)
359 Returns:
360 Updated player dict, or None on error (e.g., number already taken)
361 """
362 try:
363 response = self.client.table("players").update({"jersey_number": new_number}).eq("id", player_id).execute()
365 if response.data and len(response.data) > 0:
366 logger.info("roster_number_updated", player_id=player_id, new_number=new_number)
367 return self._add_display_name(response.data[0])
368 return None
370 except Exception as e:
371 logger.error(
372 "roster_number_update_error",
373 player_id=player_id,
374 new_number=new_number,
375 error=str(e),
376 )
377 return None
379 @invalidates_cache(ROSTER_CACHE_PATTERN)
380 def bulk_renumber(
381 self,
382 team_id: int,
383 season_id: int,
384 changes: list[dict],
385 ) -> bool:
386 """
387 Reassign multiple jersey numbers at once.
389 Uses negative numbers as temporary placeholders to avoid uniqueness
390 constraint violations during swaps.
392 Args:
393 team_id: Team ID
394 season_id: Season ID
395 changes: List of dicts with player_id and new_number
397 Returns:
398 True if successful, False on error
399 """
400 try:
401 # Step 1: Set all affected players to negative numbers (temporary)
402 for i, change in enumerate(changes):
403 self.client.table("players").update(
404 {
405 "jersey_number": -(i + 1000) # Negative temp value
406 }
407 ).eq("id", change["player_id"]).eq("team_id", team_id).eq("season_id", season_id).execute()
409 # Step 2: Set final numbers
410 for change in changes:
411 self.client.table("players").update({"jersey_number": change["new_number"]}).eq(
412 "id", change["player_id"]
413 ).eq("team_id", team_id).eq("season_id", season_id).execute()
415 logger.info("roster_bulk_renumbered", team_id=team_id, count=len(changes))
416 return True
418 except Exception as e:
419 logger.error("roster_bulk_renumber_error", team_id=team_id, error=str(e))
420 return False
422 @invalidates_cache(ROSTER_CACHE_PATTERN)
423 def link_user_to_player(
424 self,
425 player_id: int,
426 user_profile_id: str,
427 ) -> dict | None:
428 """
429 Link an MT account to a roster entry.
431 Called when a player accepts an invitation tied to a roster entry.
433 Args:
434 player_id: Player ID (roster entry)
435 user_profile_id: User profile ID to link
437 Returns:
438 Updated player dict, or None on error
439 """
440 try:
441 response = (
442 self.client.table("players").update({"user_profile_id": user_profile_id}).eq("id", player_id).execute()
443 )
445 if response.data and len(response.data) > 0:
446 logger.info("roster_user_linked", player_id=player_id, user_profile_id=user_profile_id)
447 # Fetch full player with user_profile data
448 return self.get_player_by_id(player_id)
449 return None
451 except Exception as e:
452 logger.error(
453 "roster_link_user_error",
454 player_id=player_id,
455 user_profile_id=user_profile_id,
456 error=str(e),
457 )
458 return None
460 # === Delete Operations ===
462 @invalidates_cache(ROSTER_CACHE_PATTERN)
463 def delete_player(self, player_id: int) -> bool:
464 """
465 Remove a roster entry (soft delete by setting is_active=false).
467 Args:
468 player_id: Player ID to remove
470 Returns:
471 True if successful, False on error
472 """
473 try:
474 response = self.client.table("players").update({"is_active": False}).eq("id", player_id).execute()
476 if response.data and len(response.data) > 0:
477 logger.info("roster_player_deleted", player_id=player_id)
478 return True
479 return False
481 except Exception as e:
482 logger.error("roster_delete_error", player_id=player_id, error=str(e))
483 return False
485 @invalidates_cache(ROSTER_CACHE_PATTERN)
486 def hard_delete_player(self, player_id: int) -> bool:
487 """
488 Permanently remove a roster entry.
490 Use sparingly - soft delete (delete_player) is preferred.
492 Args:
493 player_id: Player ID to remove
495 Returns:
496 True if successful, False on error
497 """
498 try:
499 self.client.table("players").delete().eq("id", player_id).execute()
501 logger.info("roster_player_hard_deleted", player_id=player_id)
502 return True
504 except Exception as e:
505 logger.error("roster_hard_delete_error", player_id=player_id, error=str(e))
506 return False
508 # === Helper Methods ===
510 def _add_display_name(self, player: dict) -> dict:
511 """
512 Add computed display_name to a player dict.
514 Priority:
515 1. Linked user's display_name or full name
516 2. Roster entry's first_name + last_name
517 3. Jersey number only: "#23"
519 Args:
520 player: Player dict from database
522 Returns:
523 Player dict with display_name and has_account added
524 """
525 user_profile = player.get("user_profile")
526 has_account = user_profile is not None and user_profile.get("id") is not None
528 display_name = None
530 # Try linked user's name first
531 if has_account:
532 if user_profile.get("display_name"):
533 display_name = user_profile["display_name"]
534 elif user_profile.get("first_name") or user_profile.get("last_name"):
535 first = user_profile.get("first_name", "")
536 last = user_profile.get("last_name", "")
537 display_name = f"{first} {last}".strip()
539 # Fall back to roster entry's name
540 if not display_name and (player.get("first_name") or player.get("last_name")):
541 first = player.get("first_name", "")
542 last = player.get("last_name", "")
543 display_name = f"{first} {last}".strip()
545 # Final fallback: jersey number only
546 if not display_name:
547 display_name = f"#{player['jersey_number']}"
549 player["display_name"] = display_name
550 player["has_account"] = has_account
552 return player
554 @staticmethod
555 def get_display_name(player: dict) -> str:
556 """
557 Static helper to get display name from a player dict.
559 Useful when you have a player dict and just need the name.
561 Args:
562 player: Player dict (may or may not have display_name computed)
564 Returns:
565 Display name string
566 """
567 if player.get("display_name"):
568 return player["display_name"]
570 # Try roster entry name
571 if player.get("first_name") or player.get("last_name"):
572 return f"{player.get('first_name', '')} {player.get('last_name', '')}".strip()
574 # Jersey number only
575 return f"#{player.get('jersey_number', '?')}"