Coverage for dao/club_dao.py: 14.17%
85 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-15 17:36 +0000
1"""
2Club Data Access Object.
4Handles all database operations related to clubs including:
5- Club CRUD operations
6- Club-team associations
7- Club queries and filters
8"""
10from collections import Counter
12import structlog
14from dao.base_dao import BaseDAO, dao_cache, invalidates_cache
16logger = structlog.get_logger()
18# Cache pattern for invalidation
19CLUBS_CACHE_PATTERN = "mt:dao:clubs:*"
22class ClubDAO(BaseDAO):
23 """Data access object for club operations."""
25 # === Club Query Methods ===
27 @dao_cache("clubs:all:{include_team_counts}")
28 def get_all_clubs(self, include_team_counts: bool = True) -> list[dict]:
29 """Get all clubs with their associated teams count.
31 Args:
32 include_team_counts: If True, include team_count field for each club
34 Returns:
35 List of clubs from the clubs table.
36 """
37 response = self.client.table("clubs").select("*").order("name").execute()
38 clubs = response.data
40 if include_team_counts:
41 # Enrich with team counts (single query instead of N+1)
42 teams_response = self.client.table("teams").select("club_id").not_.is_("club_id", "null").execute()
43 team_counts = Counter(team["club_id"] for team in teams_response.data if team.get("club_id"))
44 for club in clubs:
45 club["team_count"] = team_counts.get(club["id"], 0)
47 return clubs
49 @dao_cache("clubs:for_team:{team_id}")
50 def get_club_for_team(self, team_id: int) -> dict | None:
51 """Get the club for a team.
53 Args:
54 team_id: The team ID
56 Returns:
57 Club dict if team belongs to a club, None otherwise
58 """
59 # Get the team to find its club_id
60 team_response = self.client.table("teams").select("club_id").eq("id", team_id).execute()
61 if not team_response.data or len(team_response.data) == 0:
62 return None
64 club_id = team_response.data[0].get("club_id")
65 if not club_id:
66 return None
68 # Get the club details
69 club_response = self.client.table("clubs").select("*").eq("id", club_id).execute()
70 if club_response.data and len(club_response.data) > 0:
71 return club_response.data[0]
72 return None
74 # === Club CRUD Methods ===
76 @invalidates_cache(CLUBS_CACHE_PATTERN)
77 def create_club(
78 self,
79 name: str,
80 city: str,
81 website: str | None = None,
82 description: str | None = None,
83 logo_url: str | None = None,
84 primary_color: str | None = None,
85 secondary_color: str | None = None,
86 ) -> dict:
87 """Create a new club.
89 Args:
90 name: Club name
91 city: Club city/location
92 website: Optional website URL
93 description: Optional description
94 logo_url: Optional URL to club logo in Supabase Storage
95 primary_color: Optional primary brand color (hex code)
96 secondary_color: Optional secondary brand color (hex code)
98 Returns:
99 Created club dict
101 Raises:
102 ValueError: If club creation fails
103 """
104 club_data = {"name": name, "city": city}
105 if website:
106 club_data["website"] = website
107 if description:
108 club_data["description"] = description
109 if logo_url:
110 club_data["logo_url"] = logo_url
111 if primary_color:
112 club_data["primary_color"] = primary_color
113 if secondary_color:
114 club_data["secondary_color"] = secondary_color
116 result = self.client.table("clubs").insert(club_data).execute()
118 if not result.data or len(result.data) == 0:
119 raise ValueError("Failed to create club")
121 return result.data[0]
123 @invalidates_cache(CLUBS_CACHE_PATTERN)
124 def update_club(
125 self,
126 club_id: int,
127 name: str | None = None,
128 city: str | None = None,
129 website: str | None = None,
130 description: str | None = None,
131 logo_url: str | None = None,
132 primary_color: str | None = None,
133 secondary_color: str | None = None,
134 ) -> dict | None:
135 """Update an existing club.
137 Args:
138 club_id: ID of club to update
139 name: Optional new name
140 city: Optional new city/location
141 website: Optional new website URL
142 description: Optional new description
143 logo_url: Optional URL to club logo in Supabase Storage
144 primary_color: Optional primary brand color (hex code)
145 secondary_color: Optional secondary brand color (hex code)
147 Returns:
148 Updated club dict or None if not found
149 """
150 update_data = {}
151 if name is not None:
152 update_data["name"] = name
153 if city is not None:
154 update_data["city"] = city
155 if website is not None:
156 update_data["website"] = website
157 if description is not None:
158 update_data["description"] = description
159 if logo_url is not None:
160 update_data["logo_url"] = logo_url
161 if primary_color is not None:
162 update_data["primary_color"] = primary_color
163 if secondary_color is not None:
164 update_data["secondary_color"] = secondary_color
166 if not update_data:
167 return None
169 result = self.client.table("clubs").update(update_data).eq("id", club_id).execute()
171 if not result.data or len(result.data) == 0:
172 return None
174 return result.data[0]
176 @invalidates_cache(CLUBS_CACHE_PATTERN)
177 def delete_club(self, club_id: int) -> bool:
178 """Delete a club.
180 Args:
181 club_id: The club ID to delete
183 Returns:
184 True if deleted successfully
186 Raises:
187 Exception: If deletion fails
189 Note:
190 This will fail if there are teams still associated with this club
191 due to ON DELETE RESTRICT constraint.
192 Invitations referencing this club are deleted first.
193 """
194 # Delete invitations referencing this club first (FK constraint)
195 self.client.table("invitations").delete().eq("club_id", club_id).execute()
196 # Now delete the club
197 self.client.table("clubs").delete().eq("id", club_id).execute()
198 return True
200 # === Team-Club Association Methods ===
202 def update_team_club(self, team_id: int, club_id: int | None) -> dict:
203 """Update the club for a team.
205 Args:
206 team_id: The team ID to update
207 club_id: The club ID to assign (or None to remove club association)
209 Returns:
210 Updated team dict
212 Raises:
213 ValueError: If team update fails
214 """
215 result = self.client.table("teams").update({"club_id": club_id}).eq("id", team_id).execute()
216 if not result.data or len(result.data) == 0:
217 raise ValueError(f"Failed to update club for team {team_id}")
218 return result.data[0]
220 def get_all_parent_club_entities(self) -> list[dict]:
221 """Get all parent club entities (teams with no club_id).
223 This includes clubs that don't have children yet.
224 Used for dropdowns where users need to select a parent club.
225 """
226 try:
227 # Get all teams that could be parent clubs (no club_id)
228 response = self.client.table("teams").select("*").is_("club_id", "null").execute()
229 return response.data
230 except Exception:
231 logger.exception("Error querying parent club entities")
232 return []