Coverage for dao/playoff_dao.py: 7.58%
190 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-15 17:36 +0000
1"""
2Playoff Bracket Data Access Object.
4Handles all database operations for playoff brackets including:
5- Bracket retrieval with denormalized match data
6- Bracket generation from standings
7- Winner advancement between rounds
8- Bracket deletion (reset)
9"""
11from datetime import date, timedelta
13import structlog
15from dao.base_dao import BaseDAO, dao_cache, invalidates_cache
17logger = structlog.get_logger()
19PLAYOFF_CACHE_PATTERN = "mt:dao:playoffs:*"
20MATCHES_CACHE_PATTERN = "mt:dao:matches:*"
22PLAYOFF_MATCH_TYPE_ID = 4
24# Seeding: A1=1, B1=2, A2=3, B2=4, A3=5, B3=6, A4=7, B4=8
25# QF matchups: (1v8, 4v5, 3v6, 2v7)
26QF_MATCHUPS = [
27 {"position": 1, "home_seed": 1, "away_seed": 8}, # A1 vs B4
28 {"position": 2, "home_seed": 4, "away_seed": 5}, # B2 vs A3
29 {"position": 3, "home_seed": 3, "away_seed": 6}, # A2 vs B3
30 {"position": 4, "home_seed": 2, "away_seed": 7}, # B1 vs A4
31]
34class PlayoffDAO(BaseDAO):
35 """Data access object for playoff bracket operations."""
37 # === Bracket Query Methods ===
39 @dao_cache("playoffs:bracket:{league_id}:{season_id}:{age_group_id}")
40 def get_bracket(
41 self, league_id: int, season_id: int, age_group_id: int
42 ) -> list[dict]:
43 """Get all bracket slots with denormalized match data.
45 Returns slots with joined match and team information for display.
46 """
47 try:
48 response = (
49 self.client.table("playoff_bracket_slots")
50 .select(
51 "*, match:matches("
52 "id, match_date, scheduled_kickoff, match_status, home_score, away_score, "
53 "home_team:teams!matches_home_team_id_fkey(id, name, club_id), "
54 "away_team:teams!matches_away_team_id_fkey(id, name, club_id)"
55 ")"
56 )
57 .eq("league_id", league_id)
58 .eq("season_id", season_id)
59 .eq("age_group_id", age_group_id)
60 .order("round")
61 .order("bracket_position")
62 .execute()
63 )
65 slots = []
66 for row in response.data:
67 match = row.get("match")
68 slot = {
69 "id": row["id"],
70 "league_id": row["league_id"],
71 "season_id": row["season_id"],
72 "age_group_id": row["age_group_id"],
73 "round": row["round"],
74 "bracket_position": row["bracket_position"],
75 "bracket_tier": row.get("bracket_tier"),
76 "match_id": row["match_id"],
77 "home_seed": row["home_seed"],
78 "away_seed": row["away_seed"],
79 "home_source_slot_id": row["home_source_slot_id"],
80 "away_source_slot_id": row["away_source_slot_id"],
81 # Denormalized match data
82 "home_team_name": None,
83 "away_team_name": None,
84 "home_team_id": None,
85 "away_team_id": None,
86 "home_club_id": None,
87 "away_club_id": None,
88 "home_score": None,
89 "away_score": None,
90 "match_status": None,
91 "match_date": None,
92 "scheduled_kickoff": None,
93 }
94 if match:
95 slot["home_team_name"] = (
96 match["home_team"]["name"]
97 if match.get("home_team")
98 else None
99 )
100 slot["away_team_name"] = (
101 match["away_team"]["name"]
102 if match.get("away_team")
103 else None
104 )
105 slot["home_team_id"] = (
106 match["home_team"]["id"]
107 if match.get("home_team")
108 else None
109 )
110 slot["away_team_id"] = (
111 match["away_team"]["id"]
112 if match.get("away_team")
113 else None
114 )
115 slot["home_club_id"] = (
116 match["home_team"].get("club_id")
117 if match.get("home_team")
118 else None
119 )
120 slot["away_club_id"] = (
121 match["away_team"].get("club_id")
122 if match.get("away_team")
123 else None
124 )
125 slot["home_score"] = match.get("home_score")
126 slot["away_score"] = match.get("away_score")
127 slot["match_status"] = match.get("match_status")
128 slot["match_date"] = match.get("match_date")
129 slot["scheduled_kickoff"] = match.get("scheduled_kickoff")
130 slots.append(slot)
132 return slots
134 except Exception:
135 logger.exception(
136 "Error fetching playoff bracket",
137 league_id=league_id,
138 season_id=season_id,
139 age_group_id=age_group_id,
140 )
141 return []
143 # === Bracket Generation ===
145 @invalidates_cache(PLAYOFF_CACHE_PATTERN, MATCHES_CACHE_PATTERN)
146 def generate_bracket(
147 self,
148 league_id: int,
149 season_id: int,
150 age_group_id: int,
151 standings_a: list[dict],
152 standings_b: list[dict],
153 division_a_id: int,
154 division_b_id: int,
155 start_date: str,
156 tiers: list[dict],
157 ) -> list[dict]:
158 """Generate configurable multi-tier 8-team single elimination brackets.
160 Creates bracket slots and QF matches for each configured tier.
161 Each tier uses the same cross-division seeding pattern.
163 Args:
164 league_id: League ID
165 season_id: Season ID
166 age_group_id: Age group ID
167 standings_a: Full standings from division A (sorted by rank)
168 standings_b: Full standings from division B (sorted by rank)
169 division_a_id: Division A ID (for team lookups)
170 division_b_id: Division B ID (for team lookups)
171 start_date: ISO date string for QF matches (e.g., "2026-02-15")
172 tiers: List of tier configs, each with name, start_position, end_position
174 Returns:
175 List of created bracket slot dicts
177 Raises:
178 ValueError: If not enough teams for the configured tiers
179 """
180 # Determine required team count from tier configuration
181 max_position = max(t["end_position"] for t in tiers)
182 if len(standings_a) < max_position:
183 raise ValueError(
184 f"Division A needs at least {max_position} teams, has {len(standings_a)}"
185 )
186 if len(standings_b) < max_position:
187 raise ValueError(
188 f"Division B needs at least {max_position} teams, has {len(standings_b)}"
189 )
191 # Check for existing bracket
192 existing = (
193 self.client.table("playoff_bracket_slots")
194 .select("id")
195 .eq("league_id", league_id)
196 .eq("season_id", season_id)
197 .eq("age_group_id", age_group_id)
198 .limit(1)
199 .execute()
200 )
201 if existing.data:
202 raise ValueError("Bracket already exists for this league/season/age group")
204 # Build name→team_id mapping from both divisions
205 team_map = self._build_team_name_map(division_a_id, division_b_id)
207 # Process each configured tier
208 tier_names = []
209 for tier_config in tiers:
210 tier = tier_config["name"]
211 tier_names.append(tier)
212 start_pos = tier_config["start_position"] - 1 # Convert to 0-indexed
213 end_pos = tier_config["end_position"]
214 slice_a = standings_a[start_pos:end_pos]
215 slice_b = standings_b[start_pos:end_pos]
217 # Build seed→team mapping for this tier
218 # A1=seed1, B1=seed2, A2=seed3, B2=seed4, A3=seed5, B3=seed6, A4=seed7, B4=seed8
219 seed_teams = {}
220 for i, standing in enumerate(slice_a):
221 seed = (i * 2) + 1 # 1, 3, 5, 7
222 name = standing["team"]
223 if name not in team_map:
224 raise ValueError(f"Team '{name}' not found in division teams")
225 seed_teams[seed] = {"name": name, "id": team_map[name]}
227 for i, standing in enumerate(slice_b):
228 seed = (i * 2) + 2 # 2, 4, 6, 8
229 name = standing["team"]
230 if name not in team_map:
231 raise ValueError(f"Team '{name}' not found in division teams")
232 seed_teams[seed] = {"name": name, "id": team_map[name]}
234 logger.info(
235 "generating_playoff_bracket_tier",
236 league_id=league_id,
237 tier=tier,
238 seed_teams={s: t["name"] for s, t in seed_teams.items()},
239 )
241 # Create QF slots and matches for this tier
242 qf_slots = []
243 for matchup in QF_MATCHUPS:
244 home = seed_teams[matchup["home_seed"]]
245 away = seed_teams[matchup["away_seed"]]
247 # Create the match with configured start date
248 match_data = {
249 "match_date": start_date,
250 "home_team_id": home["id"],
251 "away_team_id": away["id"],
252 "season_id": season_id,
253 "age_group_id": age_group_id,
254 "match_type_id": PLAYOFF_MATCH_TYPE_ID,
255 "match_status": "scheduled",
256 "source": "playoff-generator",
257 }
258 match_response = (
259 self.client.table("matches").insert(match_data).execute()
260 )
261 match_id = match_response.data[0]["id"]
263 # Create the bracket slot
264 slot_data = {
265 "league_id": league_id,
266 "season_id": season_id,
267 "age_group_id": age_group_id,
268 "bracket_tier": tier,
269 "round": "quarterfinal",
270 "bracket_position": matchup["position"],
271 "match_id": match_id,
272 "home_seed": matchup["home_seed"],
273 "away_seed": matchup["away_seed"],
274 }
275 slot_response = (
276 self.client.table("playoff_bracket_slots")
277 .insert(slot_data)
278 .execute()
279 )
280 qf_slots.append(slot_response.data[0])
282 # Create SF slots (no matches yet — teams TBD)
283 sf_slots = []
284 # SF1: winner of QF1 vs winner of QF2
285 sf1_data = {
286 "league_id": league_id,
287 "season_id": season_id,
288 "age_group_id": age_group_id,
289 "bracket_tier": tier,
290 "round": "semifinal",
291 "bracket_position": 1,
292 "home_source_slot_id": qf_slots[0]["id"],
293 "away_source_slot_id": qf_slots[1]["id"],
294 }
295 sf1_response = (
296 self.client.table("playoff_bracket_slots")
297 .insert(sf1_data)
298 .execute()
299 )
300 sf_slots.append(sf1_response.data[0])
302 # SF2: winner of QF3 vs winner of QF4
303 sf2_data = {
304 "league_id": league_id,
305 "season_id": season_id,
306 "age_group_id": age_group_id,
307 "bracket_tier": tier,
308 "round": "semifinal",
309 "bracket_position": 2,
310 "home_source_slot_id": qf_slots[2]["id"],
311 "away_source_slot_id": qf_slots[3]["id"],
312 }
313 sf2_response = (
314 self.client.table("playoff_bracket_slots")
315 .insert(sf2_data)
316 .execute()
317 )
318 sf_slots.append(sf2_response.data[0])
320 # Create Final slot (no match yet)
321 final_data = {
322 "league_id": league_id,
323 "season_id": season_id,
324 "age_group_id": age_group_id,
325 "bracket_tier": tier,
326 "round": "final",
327 "bracket_position": 1,
328 "home_source_slot_id": sf_slots[0]["id"],
329 "away_source_slot_id": sf_slots[1]["id"],
330 }
331 self.client.table("playoff_bracket_slots").insert(final_data).execute()
333 logger.info(
334 "playoff_bracket_generated",
335 league_id=league_id,
336 season_id=season_id,
337 age_group_id=age_group_id,
338 tiers=tier_names,
339 )
341 # Return the full bracket
342 return self.get_bracket.__wrapped__(self, league_id, season_id, age_group_id)
344 # === Winner Advancement ===
346 @invalidates_cache(PLAYOFF_CACHE_PATTERN, MATCHES_CACHE_PATTERN)
347 def advance_winner(self, slot_id: int) -> dict | None:
348 """Advance the winner of a completed slot to the next round.
350 Determines the winner from the linked match scores, finds the
351 next-round slot where this slot feeds into, and updates it.
352 When both feeder slots for a next-round slot are complete,
353 creates the match for that round.
355 Args:
356 slot_id: The bracket slot ID whose winner should advance
358 Returns:
359 The updated next-round slot dict, or None on error
361 Raises:
362 ValueError: If match not completed, scores tied, or no next round
363 """
364 # Get the completed slot
365 slot_response = (
366 self.client.table("playoff_bracket_slots")
367 .select("*, match:matches(id, home_team_id, away_team_id, home_score, away_score, match_status)")
368 .eq("id", slot_id)
369 .execute()
370 )
371 if not slot_response.data:
372 raise ValueError(f"Slot {slot_id} not found")
374 slot = slot_response.data[0]
375 match = slot.get("match")
376 if not match:
377 raise ValueError(f"Slot {slot_id} has no linked match")
378 if match.get("match_status") not in ("completed", "forfeit"):
379 raise ValueError(f"Match for slot {slot_id} is not completed")
380 if match["home_score"] is None or match["away_score"] is None:
381 raise ValueError(f"Match for slot {slot_id} has no scores")
382 if match["home_score"] == match["away_score"]:
383 raise ValueError(
384 f"Match for slot {slot_id} is tied — admin must resolve before advancing"
385 )
387 # Determine winner
388 winner_team_id = (
389 match["home_team_id"]
390 if match["home_score"] > match["away_score"]
391 else match["away_team_id"]
392 )
394 # Find the next-round slot that this slot feeds into
395 next_slot_response = (
396 self.client.table("playoff_bracket_slots")
397 .select("*")
398 .or_(
399 f"home_source_slot_id.eq.{slot_id},"
400 f"away_source_slot_id.eq.{slot_id}"
401 )
402 .execute()
403 )
404 if not next_slot_response.data:
405 raise ValueError(
406 f"No next-round slot found for slot {slot_id} — may be the final"
407 )
409 next_slot = next_slot_response.data[0]
411 # Check if both feeder slots are now complete
412 other_source_slot_id = (
413 next_slot["away_source_slot_id"]
414 if next_slot["home_source_slot_id"] == slot_id
415 else next_slot["home_source_slot_id"]
416 )
418 other_winner_team_id = None
419 if other_source_slot_id:
420 other_slot_response = (
421 self.client.table("playoff_bracket_slots")
422 .select("match:matches(home_team_id, away_team_id, home_score, away_score, match_status)")
423 .eq("id", other_source_slot_id)
424 .execute()
425 )
426 if other_slot_response.data:
427 other_match = other_slot_response.data[0].get("match")
428 if other_match and other_match.get("match_status") in ("completed", "forfeit"):
429 if (
430 other_match["home_score"] is not None
431 and other_match["away_score"] is not None
432 and other_match["home_score"] != other_match["away_score"]
433 ):
434 if other_match["home_score"] > other_match["away_score"]:
435 other_winner_team_id = other_match["home_team_id"]
436 else:
437 other_winner_team_id = other_match["away_team_id"]
439 # If both feeders are complete, create the next-round match
440 if other_winner_team_id and not next_slot.get("match_id"):
441 # Determine home/away based on which source slot they came from
442 if next_slot["home_source_slot_id"] == slot_id:
443 home_team_id = winner_team_id
444 away_team_id = other_winner_team_id
445 else:
446 home_team_id = other_winner_team_id
447 away_team_id = winner_team_id
449 default_date = (date.today() + timedelta(days=5)).isoformat()
450 match_data = {
451 "match_date": default_date,
452 "home_team_id": home_team_id,
453 "away_team_id": away_team_id,
454 "season_id": next_slot["season_id"],
455 "age_group_id": next_slot["age_group_id"],
456 "match_type_id": PLAYOFF_MATCH_TYPE_ID,
457 "match_status": "scheduled",
458 "source": "playoff-generator",
459 }
460 match_response = (
461 self.client.table("matches").insert(match_data).execute()
462 )
463 new_match_id = match_response.data[0]["id"]
465 # Link match to the bracket slot
466 self.client.table("playoff_bracket_slots").update(
467 {"match_id": new_match_id}
468 ).eq("id", next_slot["id"]).execute()
470 logger.info(
471 "playoff_next_round_match_created",
472 next_slot_id=next_slot["id"],
473 round=next_slot["round"],
474 home_team_id=home_team_id,
475 away_team_id=away_team_id,
476 match_id=new_match_id,
477 )
479 logger.info(
480 "playoff_winner_advanced",
481 from_slot_id=slot_id,
482 to_slot_id=next_slot["id"],
483 winner_team_id=winner_team_id,
484 )
486 # Return the updated next-round slot
487 updated = (
488 self.client.table("playoff_bracket_slots")
489 .select("*")
490 .eq("id", next_slot["id"])
491 .execute()
492 )
493 return updated.data[0] if updated.data else None
495 # === Forfeit ===
497 @invalidates_cache(PLAYOFF_CACHE_PATTERN, MATCHES_CACHE_PATTERN)
498 def forfeit_match(self, slot_id: int, forfeit_team_id: int) -> dict | None:
499 """Declare a forfeit on a playoff match.
501 The forfeiting team loses 0-3, the match is marked as 'forfeit',
502 and the winning team automatically advances in the bracket.
504 Args:
505 slot_id: The bracket slot ID
506 forfeit_team_id: The team ID that is forfeiting
508 Returns:
509 The updated next-round slot dict from advance_winner, or None if final
511 Raises:
512 ValueError: If match not found, wrong status, or team not a participant
513 """
514 # Get the bracket slot with linked match
515 slot_response = (
516 self.client.table("playoff_bracket_slots")
517 .select(
518 "*, match:matches(id, home_team_id, away_team_id, "
519 "home_score, away_score, match_status)"
520 )
521 .eq("id", slot_id)
522 .execute()
523 )
524 if not slot_response.data:
525 raise ValueError(f"Slot {slot_id} not found")
527 slot = slot_response.data[0]
528 match = slot.get("match")
529 if not match:
530 raise ValueError(f"Slot {slot_id} has no linked match")
531 if match.get("match_status") not in ("scheduled", "live"):
532 raise ValueError(
533 f"Match for slot {slot_id} cannot be forfeited "
534 f"(status: {match.get('match_status')})"
535 )
537 home_team_id = match["home_team_id"]
538 away_team_id = match["away_team_id"]
540 if forfeit_team_id not in (home_team_id, away_team_id):
541 raise ValueError(
542 f"Team {forfeit_team_id} is not a participant in this match"
543 )
545 # Set scores: non-forfeiting team gets 3, forfeiting team gets 0
546 if forfeit_team_id == home_team_id:
547 home_score = 0
548 away_score = 3
549 else:
550 home_score = 3
551 away_score = 0
553 # Update the match
554 self.client.table("matches").update(
555 {
556 "match_status": "forfeit",
557 "home_score": home_score,
558 "away_score": away_score,
559 "forfeit_team_id": forfeit_team_id,
560 }
561 ).eq("id", match["id"]).execute()
563 logger.info(
564 "playoff_match_forfeited",
565 slot_id=slot_id,
566 match_id=match["id"],
567 forfeit_team_id=forfeit_team_id,
568 )
570 # Advance the winner to the next round (reuse existing logic)
571 if slot["round"] != "final":
572 return self.advance_winner(slot_id)
574 return None
576 # === Bracket Deletion ===
578 @invalidates_cache(PLAYOFF_CACHE_PATTERN, MATCHES_CACHE_PATTERN)
579 def delete_bracket(
580 self, league_id: int, season_id: int, age_group_id: int
581 ) -> int:
582 """Delete an entire playoff bracket and its associated matches.
584 Deletes in order: unlink matches from slots, delete slots
585 (final first due to self-referencing FKs), then delete orphaned
586 playoff matches.
588 Returns:
589 Number of slots deleted
590 """
591 try:
592 # Get all slots and their match IDs
593 slots_response = (
594 self.client.table("playoff_bracket_slots")
595 .select("id, match_id, round")
596 .eq("league_id", league_id)
597 .eq("season_id", season_id)
598 .eq("age_group_id", age_group_id)
599 .execute()
600 )
602 if not slots_response.data:
603 return 0
605 match_ids = [
606 s["match_id"] for s in slots_response.data if s["match_id"]
607 ]
609 # Delete slots in reverse round order to respect self-referencing FKs
610 round_order = ["final", "semifinal", "quarterfinal"]
611 total_deleted = 0
612 for round_name in round_order:
613 round_slots = [
614 s for s in slots_response.data if s["round"] == round_name
615 ]
616 for slot in round_slots:
617 self.client.table("playoff_bracket_slots").delete().eq(
618 "id", slot["id"]
619 ).execute()
620 total_deleted += 1
622 # Delete associated playoff matches
623 for match_id in match_ids:
624 self.client.table("matches").delete().eq("id", match_id).execute()
626 logger.info(
627 "playoff_bracket_deleted",
628 league_id=league_id,
629 season_id=season_id,
630 age_group_id=age_group_id,
631 slots_deleted=total_deleted,
632 matches_deleted=len(match_ids),
633 )
635 return total_deleted
637 except Exception:
638 logger.exception(
639 "Error deleting playoff bracket",
640 league_id=league_id,
641 season_id=season_id,
642 age_group_id=age_group_id,
643 )
644 raise
646 # === Internal Helpers ===
648 def _build_team_name_map(
649 self, division_a_id: int, division_b_id: int
650 ) -> dict[str, int]:
651 """Build a team name → team ID mapping from both divisions.
653 Queries team_mappings to find teams assigned to either division.
654 """
655 team_ids = set()
656 for div_id in [division_a_id, division_b_id]:
657 response = (
658 self.client.table("team_mappings")
659 .select("team_id")
660 .eq("division_id", div_id)
661 .execute()
662 )
663 for row in response.data:
664 team_ids.add(row["team_id"])
666 if not team_ids:
667 return {}
669 teams_response = (
670 self.client.table("teams")
671 .select("id, name")
672 .in_("id", list(team_ids))
673 .execute()
674 )
676 return {team["name"]: team["id"] for team in teams_response.data}