cal/major-domo-database
1
0
Fork 0
Code Issues 25 Pull Requests 25 Actions Packages Projects Releases Wiki Activity

Fix linting and formatting issues

Browse Source 
- Add missing imports: json, csv, io, model_to_dict
- Remove unused imports: Any, Dict, Type, invalidate_cache
- Remove redundant f-string prefixes from static error messages
- Format code with black
- All ruff and black checks pass
- All 76 unit tests pass (9 skipped)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Cal Corum 2026-02-04 08:44:12 -06:00
parent 408b187305
commit 2189aea8da
3 changed files with 299 additions and 242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
app/services/base.py
View File

@ -3,20 +3,25 @@ Base Service Class - Dependency Injection Version
Provides common functionality with configurable dependencies. Provides common functionality with configurable dependencies.
""" """
import json
import logging import logging
from typing import Optional, Any, Dict, TypeVar, Type from typing import Optional, TypeVar
from .interfaces import AbstractPlayerRepository, AbstractTeamRepository, AbstractCacheService from .interfaces import (
AbstractPlayerRepository,
AbstractTeamRepository,
AbstractCacheService,
)
from .mocks import MockCacheService from .mocks import MockCacheService
logger = logging.getLogger('discord_app') logger = logging.getLogger("discord_app")
T = TypeVar('T') T = TypeVar("T")
class ServiceConfig: class ServiceConfig:
"""Configuration for service dependencies.""" """Configuration for service dependencies."""
def __init__( def __init__(
self, self,
player_repo: Optional[AbstractPlayerRepository] = None, player_repo: Optional[AbstractPlayerRepository] = None,
@ -34,10 +39,10 @@ _default_config = ServiceConfig()
class BaseService: class BaseService:
"""Base class for all services with dependency injection support.""" """Base class for all services with dependency injection support."""
# Subclasses should override these # Subclasses should override these
cache_patterns = [] cache_patterns = []
def __init__( def __init__(
self, self,
config: Optional[ServiceConfig] = None, config: Optional[ServiceConfig] = None,
@ -47,7 +52,7 @@ class BaseService:
): ):
""" """
Initialize service with dependencies. Initialize service with dependencies.
Args: Args:
config: Optional ServiceConfig containing all dependencies config: Optional ServiceConfig containing all dependencies
player_repo: Override for player repository player_repo: Override for player repository
@ -63,163 +68,170 @@ class BaseService:
self._player_repo = player_repo self._player_repo = player_repo
self._team_repo = team_repo self._team_repo = team_repo
self._cache = cache self._cache = cache
# Lazy imports for defaults (avoids circular imports) # Lazy imports for defaults (avoids circular imports)
self._using_defaults = ( self._using_defaults = (
self._player_repo is None and self._player_repo is None
self._team_repo is None and and self._team_repo is None
self._cache is None and self._cache is None
) )
@property @property
def player_repo(self) -> AbstractPlayerRepository: def player_repo(self) -> AbstractPlayerRepository:
"""Get player repository, importing from db_engine if not set.""" """Get player repository, importing from db_engine if not set."""
if self._player_repo is None: if self._player_repo is None:
from ..db_engine import Player from ..db_engine import Player
class DefaultPlayerRepo: class DefaultPlayerRepo:
def select_season(self, season): def select_season(self, season):
return Player.select_season(season) return Player.select_season(season)
def get_by_id(self, player_id): def get_by_id(self, player_id):
return Player.get_or_none(Player.id == player_id) return Player.get_or_none(Player.id == player_id)
def get_or_none(self, *conditions): def get_or_none(self, *conditions):
return Player.get_or_none(*conditions) return Player.get_or_none(*conditions)
def update(self, data, *conditions): def update(self, data, *conditions):
return Player.update(data).where(*conditions).execute() return Player.update(data).where(*conditions).execute()
def insert_many(self, data): def insert_many(self, data):
return Player.insert_many(data).execute() return Player.insert_many(data).execute()
def delete_by_id(self, player_id): def delete_by_id(self, player_id):
player = Player.get_by_id(player_id) player = Player.get_by_id(player_id)
if player: if player:
return player.delete_instance() return player.delete_instance()
return 0 return 0
self._player_repo = DefaultPlayerRepo() self._player_repo = DefaultPlayerRepo()
return self._player_repo return self._player_repo
@property @property
def team_repo(self) -> AbstractTeamRepository: def team_repo(self) -> AbstractTeamRepository:
"""Get team repository, importing from db_engine if not set.""" """Get team repository, importing from db_engine if not set."""
if self._team_repo is None: if self._team_repo is None:
from ..db_engine import Team from ..db_engine import Team
class DefaultTeamRepo: class DefaultTeamRepo:
def select_season(self, season): def select_season(self, season):
return Team.select_season(season) return Team.select_season(season)
def get_by_id(self, team_id): def get_by_id(self, team_id):
return Team.get_by_id(team_id) return Team.get_by_id(team_id)
def get_or_none(self, *conditions): def get_or_none(self, *conditions):
return Team.get_or_none(*conditions) return Team.get_or_none(*conditions)
def update(self, data, *conditions): def update(self, data, *conditions):
return Team.update(data).where(*conditions).execute() return Team.update(data).where(*conditions).execute()
def insert_many(self, data): def insert_many(self, data):
return Team.insert_many(data).execute() return Team.insert_many(data).execute()
def delete_by_id(self, team_id): def delete_by_id(self, team_id):
team = Team.get_by_id(team_id) team = Team.get_by_id(team_id)
if team: if team:
return team.delete_instance() return team.delete_instance()
return 0 return 0
self._team_repo = DefaultTeamRepo() self._team_repo = DefaultTeamRepo()
return self._team_repo return self._team_repo
@property @property
def cache(self) -> AbstractCacheService: def cache(self) -> AbstractCacheService:
"""Get cache service, importing from dependencies if not set.""" """Get cache service, importing from dependencies if not set."""
if self._cache is None: if self._cache is None:
try: try:
from ..dependencies import redis_client, invalidate_cache from ..dependencies import redis_client
class DefaultCache: class DefaultCache:
def get(self, key: str): def get(self, key: str):
if redis_client is None: if redis_client is None:
return None return None
return redis_client.get(key) return redis_client.get(key)
def set(self, key: str, value: str, ttl: int = 300): def set(self, key: str, value: str, ttl: int = 300):
if redis_client is None: if redis_client is None:
return False return False
redis_client.setex(key, ttl, value) redis_client.setex(key, ttl, value)
return True return True
def setex(self, key: str, ttl: int, value: str): def setex(self, key: str, ttl: int, value: str):
return self.set(key, value, ttl) return self.set(key, value, ttl)
def keys(self, pattern: str): def keys(self, pattern: str):
if redis_client is None: if redis_client is None:
return [] return []
return redis_client.keys(pattern) return redis_client.keys(pattern)
def delete(self, *keys: str): def delete(self, *keys: str):
if redis_client is None: if redis_client is None:
return 0 return 0
return redis_client.delete(*keys) return redis_client.delete(*keys)
def invalidate_pattern(self, pattern: str): def invalidate_pattern(self, pattern: str):
if redis_client is None: if redis_client is None:
return 0 return 0
keys = self.keys(pattern) keys = self.keys(pattern)
return self.delete(*keys) return self.delete(*keys)
def exists(self, key: str): def exists(self, key: str):
if redis_client is None: if redis_client is None:
return False return False
return redis_client.exists(key) return redis_client.exists(key)
self._cache = DefaultCache() self._cache = DefaultCache()
except ImportError: except ImportError:
# Fall back to mock if dependencies not available # Fall back to mock if dependencies not available
self._cache = MockCacheService() self._cache = MockCacheService()
return self._cache return self._cache
def close_db(self): def close_db(self):
"""Safely close database connection (for non-injected repos).""" """Safely close database connection (for non-injected repos)."""
if self._using_defaults: if self._using_defaults:
try: try:
from ..db_engine import db from ..db_engine import db
db.close() db.close()
except Exception: except Exception:
pass # Connection may already be closed pass # Connection may already be closed
def invalidate_cache_for(self, entity_type: str, entity_id: Optional[int] = None): def invalidate_cache_for(self, entity_type: str, entity_id: Optional[int] = None):
"""Invalidate cache entries for an entity.""" """Invalidate cache entries for an entity."""
if entity_id: if entity_id:
self.cache.invalidate_pattern(f"{entity_type}*{entity_id}*") self.cache.invalidate_pattern(f"{entity_type}*{entity_id}*")
else: else:
self.cache.invalidate_pattern(f"{entity_type}*") self.cache.invalidate_pattern(f"{entity_type}*")
def invalidate_related_cache(self, patterns: list): def invalidate_related_cache(self, patterns: list):
"""Invalidate multiple cache patterns.""" """Invalidate multiple cache patterns."""
for pattern in patterns: for pattern in patterns:
self.cache.invalidate_pattern(pattern) self.cache.invalidate_pattern(pattern)
def handle_error(self, operation: str, error: Exception, rethrow: bool = True) -> dict: def handle_error(
self, operation: str, error: Exception, rethrow: bool = True
) -> dict:
"""Handle errors consistently.""" """Handle errors consistently."""
logger.error(f"{operation}: {error}") logger.error(f"{operation}: {error}")
if rethrow: if rethrow:
try: try:
from fastapi import HTTPException from fastapi import HTTPException
raise HTTPException(status_code=500, detail=f"{operation}: {str(error)}")
raise HTTPException(
status_code=500, detail=f"{operation}: {str(error)}"
)
except ImportError: except ImportError:
# For testing without FastAPI # For testing without FastAPI
raise RuntimeError(f"{operation}: {str(error)}") raise RuntimeError(f"{operation}: {str(error)}")
return {"error": operation, "detail": str(error)} return {"error": operation, "detail": str(error)}
@classmethod @classmethod
def log_error(cls, operation: str, error: Exception) -> None: def log_error(cls, operation: str, error: Exception) -> None:
"""Class method for logging errors without needing an instance.""" """Class method for logging errors without needing an instance."""
logger.error(f"{operation}: {error}") logger.error(f"{operation}: {error}")
def require_auth(self, token: str) -> bool: def require_auth(self, token: str) -> bool:
"""Validate authentication token.""" """Validate authentication token."""
try: try:
@ -227,56 +239,60 @@ class BaseService:
from ..dependencies import valid_token from ..dependencies import valid_token
if not valid_token(token): if not valid_token(token):
logger.warning(f"Unauthorized access attempt with token: {token[:10]}...") logger.warning(
f"Unauthorized access attempt with token: {token[:10]}..."
)
raise HTTPException(status_code=401, detail="Unauthorized") raise HTTPException(status_code=401, detail="Unauthorized")
except ImportError: except ImportError:
# For testing without FastAPI - accept "valid_token" as test token # For testing without FastAPI - accept "valid_token" as test token
if token != "valid_token": if token != "valid_token":
logger.warning(f"Unauthorized access attempt with token: {token[:10] if len(token) >= 10 else token}...") logger.warning(
f"Unauthorized access attempt with token: {token[:10] if len(token) >= 10 else token}..."
)
error = RuntimeError("Unauthorized") error = RuntimeError("Unauthorized")
error.status_code = 401 # Add status_code for test compatibility error.status_code = 401 # Add status_code for test compatibility
raise error raise error
return True return True
def format_csv_response(self, headers: list, rows: list) -> str: def format_csv_response(self, headers: list, rows: list) -> str:
"""Format data as CSV.""" """Format data as CSV."""
from pandas import DataFrame from pandas import DataFrame
all_data = [headers] + rows all_data = [headers] + rows
return DataFrame(all_data).to_csv(header=False, index=False) return DataFrame(all_data).to_csv(header=False, index=False)
def parse_query_params(self, params: dict, remove_none: bool = True) -> dict: def parse_query_params(self, params: dict, remove_none: bool = True) -> dict:
"""Parse and clean query parameters.""" """Parse and clean query parameters."""
if remove_none: if remove_none:
return {k: v for k, v in params.items() if v is not None and v != [] and v != ""} return {
k: v for k, v in params.items() if v is not None and v != [] and v != ""
}
return params return params
def with_cache( def with_cache(self, key: str, ttl: int = 300, fallback: Optional[callable] = None):
self,
key: str,
ttl: int = 300,
fallback: Optional[callable] = None
):
""" """
Decorator-style cache wrapper for methods. Decorator-style cache wrapper for methods.
Usage: Usage:
@service.with_cache("player:123", ttl=600) @service.with_cache("player:123", ttl=600)
def get_player(self): def get_player(self):
... ...
""" """
def decorator(func): def decorator(func):
async def wrapper(*args, **kwargs): async def wrapper(*args, **kwargs):
# Try cache first # Try cache first
cached = self.cache.get(key) cached = self.cache.get(key)
if cached: if cached:
return json.loads(cached) return json.loads(cached)
# Execute and cache result # Execute and cache result
result = func(*args, **kwargs) result = func(*args, **kwargs)
if result is not None: if result is not None:
import json
self.cache.set(key, json.dumps(result, default=str), ttl) self.cache.set(key, json.dumps(result, default=str), ttl)
return result return result
return wrapper return wrapper
return decorator return decorator

262
app/services/player_service.py
View File

@ -3,6 +3,8 @@ Player Service - Dependency Injection Version
Business logic for player operations with injectable dependencies. Business logic for player operations with injectable dependencies.
""" """
import csv
import io
import logging import logging
from typing import List, Optional, Dict, Any, TYPE_CHECKING from typing import List, Optional, Dict, Any, TYPE_CHECKING
@ -27,18 +29,14 @@ except ImportError:
self.detail = detail self.detail = detail
super().__init__(detail) super().__init__(detail)
logger = logging.getLogger('discord_app')
logger = logging.getLogger("discord_app")
class PlayerService(BaseService): class PlayerService(BaseService):
"""Service for player-related operations with dependency injection.""" """Service for player-related operations with dependency injection."""
cache_patterns = [ cache_patterns = ["players*", "players-search*", "player*", "team-roster*"]
"players*",
"players-search*",
"player*",
"team-roster*"
]
# Class-level repository for dependency injection # Class-level repository for dependency injection
_injected_repo: Optional[AbstractPlayerRepository] = None _injected_repo: Optional[AbstractPlayerRepository] = None
@ -46,8 +44,8 @@ class PlayerService(BaseService):
def __init__( def __init__(
self, self,
player_repo: Optional[AbstractPlayerRepository] = None, player_repo: Optional[AbstractPlayerRepository] = None,
config: Optional['ServiceConfig'] = None, config: Optional["ServiceConfig"] = None,
**kwargs **kwargs,
): ):
""" """
Initialize PlayerService with optional repository. Initialize PlayerService with optional repository.
@ -75,10 +73,10 @@ class PlayerService(BaseService):
return cls._get_real_repo() return cls._get_real_repo()
@classmethod @classmethod
def _get_real_repo(cls) -> 'RealPlayerRepository': def _get_real_repo(cls) -> "RealPlayerRepository":
"""Get a real DB repository for production use.""" """Get a real DB repository for production use."""
return RealPlayerRepository(Player) return RealPlayerRepository(Player)
@classmethod @classmethod
def get_players( def get_players(
cls, cls,
@ -90,7 +88,7 @@ class PlayerService(BaseService):
is_injured: Optional[bool] = None, is_injured: Optional[bool] = None,
sort: Optional[str] = None, sort: Optional[str] = None,
short_output: bool = False, short_output: bool = False,
as_csv: bool = False as_csv: bool = False,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
""" """
Get players with filtering and sorting. Get players with filtering and sorting.
@ -124,7 +122,7 @@ class PlayerService(BaseService):
pos=pos, pos=pos,
strat_code=strat_code, strat_code=strat_code,
name=name, name=name,
is_injured=is_injured is_injured=is_injured,
) )
# Apply sorting # Apply sorting
@ -137,18 +135,17 @@ class PlayerService(BaseService):
if as_csv: if as_csv:
return cls._format_player_csv(players_data) return cls._format_player_csv(players_data)
else: else:
return { return {"count": len(players_data), "players": players_data}
"count": len(players_data),
"players": players_data
}
except Exception as e: except Exception as e:
logger.error(f"Error fetching players: {e}") logger.error(f"Error fetching players: {e}")
try: try:
raise HTTPException(status_code=500, detail=f"Error fetching players: {str(e)}") raise HTTPException(
status_code=500, detail=f"Error fetching players: {str(e)}"
)
except ImportError: except ImportError:
raise RuntimeError(f"Error fetching players: {str(e)}") raise RuntimeError(f"Error fetching players: {str(e)}")
@classmethod @classmethod
def _apply_player_filters( def _apply_player_filters(
cls, cls,
@ -157,10 +154,10 @@ class PlayerService(BaseService):
pos: Optional[List[str]] = None, pos: Optional[List[str]] = None,
strat_code: Optional[List[str]] = None, strat_code: Optional[List[str]] = None,
name: Optional[str] = None, name: Optional[str] = None,
is_injured: Optional[bool] = None is_injured: Optional[bool] = None,
) -> QueryResult: ) -> QueryResult:
"""Apply player filters in a repo-agnostic way.""" """Apply player filters in a repo-agnostic way."""
# Check if repo supports where() method (real DB) # Check if repo supports where() method (real DB)
# Only use DB-native filtering if: # Only use DB-native filtering if:
# 1. Query has where() method # 1. Query has where() method
@ -169,34 +166,34 @@ class PlayerService(BaseService):
for item in query: for item in query:
first_item = item first_item = item
break break
# Use DB-native filtering only for real Peewee models # Use DB-native filtering only for real Peewee models
if first_item is not None and not isinstance(first_item, dict): if first_item is not None and not isinstance(first_item, dict):
try: try:
if team_id: if team_id:
query = query.where(Player.team_id << team_id) query = query.where(Player.team_id << team_id)
if strat_code: if strat_code:
code_list = [x.lower() for x in strat_code] code_list = [x.lower() for x in strat_code]
query = query.where(peewee_fn.Lower(Player.strat_code) << code_list) query = query.where(peewee_fn.Lower(Player.strat_code) << code_list)
if name: if name:
query = query.where(peewee_fn.lower(Player.name) == name.lower()) query = query.where(peewee_fn.lower(Player.name) == name.lower())
if pos: if pos:
p_list = [x.upper() for x in pos] p_list = [x.upper() for x in pos]
pos_conditions = ( pos_conditions = (
(Player.pos_1 << p_list) | (Player.pos_1 << p_list)
(Player.pos_2 << p_list) | | (Player.pos_2 << p_list)
(Player.pos_3 << p_list) | | (Player.pos_3 << p_list)
(Player.pos_4 << p_list) | | (Player.pos_4 << p_list)
(Player.pos_5 << p_list) | | (Player.pos_5 << p_list)
(Player.pos_6 << p_list) | | (Player.pos_6 << p_list)
(Player.pos_7 << p_list) | | (Player.pos_7 << p_list)
(Player.pos_8 << p_list) | (Player.pos_8 << p_list)
) )
query = query.where(pos_conditions) query = query.where(pos_conditions)
if is_injured is not None: if is_injured is not None:
if is_injured: if is_injured:
query = query.where(Player.il_return.is_null(False)) query = query.where(Player.il_return.is_null(False))
@ -208,55 +205,54 @@ class PlayerService(BaseService):
else: else:
# Use Python filtering for mocks # Use Python filtering for mocks
def matches(player): def matches(player):
if team_id and player.get('team_id') not in team_id: if team_id and player.get("team_id") not in team_id:
return False return False
if strat_code: if strat_code:
code_list = [s.lower() for s in strat_code] code_list = [s.lower() for s in strat_code]
player_code = (player.get('strat_code') or '').lower() player_code = (player.get("strat_code") or "").lower()
if player_code not in code_list: if player_code not in code_list:
return False return False
if name and (player.get('name') or '').lower() != name.lower(): if name and (player.get("name") or "").lower() != name.lower():
return False return False
if pos: if pos:
p_list = [p.upper() for p in pos] p_list = [p.upper() for p in pos]
player_pos = [ player_pos = [
player.get(f'pos_{i}') for i in range(1, 9) player.get(f"pos_{i}")
if player.get(f'pos_{i}') for i in range(1, 9)
if player.get(f"pos_{i}")
] ]
if not any(p in p_list for p in player_pos): if not any(p in p_list for p in player_pos):
return False return False
if is_injured is not None: if is_injured is not None:
has_injury = player.get('il_return') is not None has_injury = player.get("il_return") is not None
if is_injured and not has_injury: if is_injured and not has_injury:
return False return False
if not is_injured and has_injury: if not is_injured and has_injury:
return False return False
return True return True
# Filter in memory # Filter in memory
filtered = [p for p in query if matches(p)] filtered = [p for p in query if matches(p)]
query = InMemoryQueryResult(filtered) query = InMemoryQueryResult(filtered)
return query return query
@classmethod @classmethod
def _apply_player_sort( def _apply_player_sort(
cls, cls, query: QueryResult, sort: Optional[str] = None
query: QueryResult,
sort: Optional[str] = None
) -> QueryResult: ) -> QueryResult:
"""Apply player sorting in a repo-agnostic way.""" """Apply player sorting in a repo-agnostic way."""
# Check if items are Peewee models (not dicts) # Check if items are Peewee models (not dicts)
first_item = None first_item = None
for item in query: for item in query:
first_item = item first_item = item
break break
# Use DB-native sorting only for real Peewee models # Use DB-native sorting only for real Peewee models
if first_item is not None and not isinstance(first_item, dict): if first_item is not None and not isinstance(first_item, dict):
try: try:
if sort == "cost-asc": if sort == "cost-asc":
query = query.order_by(Player.wara) query = query.order_by(Player.wara)
elif sort == "cost-desc": elif sort == "cost-desc":
@ -270,13 +266,14 @@ class PlayerService(BaseService):
except ImportError: except ImportError:
# Fall back to Python sorting if DB not available # Fall back to Python sorting if DB not available
pass pass
# Use Python sorting for mocks or if DB sort failed # Use Python sorting for mocks or if DB sort failed
if not hasattr(query, 'order_by') or isinstance(query, InMemoryQueryResult): if not hasattr(query, "order_by") or isinstance(query, InMemoryQueryResult):
def get_sort_key(player): def get_sort_key(player):
name = player.get('name', '') name = player.get("name", "")
wara = player.get('wara', 0) wara = player.get("wara", 0)
player_id = player.get('id', 0) player_id = player.get("id", 0)
if sort == "cost-asc": if sort == "cost-asc":
return (wara, name, player_id) return (wara, name, player_id)
@ -290,29 +287,27 @@ class PlayerService(BaseService):
return (player_id,) return (player_id,)
# Use reverse for descending name sort # Use reverse for descending name sort
reverse_sort = (sort == "name-desc") reverse_sort = sort == "name-desc"
sorted_list = sorted(list(query), key=get_sort_key, reverse=reverse_sort) sorted_list = sorted(list(query), key=get_sort_key, reverse=reverse_sort)
query = InMemoryQueryResult(sorted_list) query = InMemoryQueryResult(sorted_list)
return query return query
@classmethod @classmethod
def _query_to_player_dicts( def _query_to_player_dicts(
cls, cls, query: QueryResult, short_output: bool = False
query: QueryResult,
short_output: bool = False
) -> List[Dict[str, Any]]: ) -> List[Dict[str, Any]]:
"""Convert query results to list of player dicts.""" """Convert query results to list of player dicts."""
# Check if we have DB models or dicts # Check if we have DB models or dicts
first_item = None first_item = None
for item in query: for item in query:
first_item = item first_item = item
break break
if first_item is None: if first_item is None:
return [] return []
# If items are already dicts (from mock) # If items are already dicts (from mock)
if isinstance(first_item, dict): if isinstance(first_item, dict):
players_data = list(query) players_data = list(query)
@ -320,33 +315,33 @@ class PlayerService(BaseService):
return players_data return players_data
# Add computed fields if needed # Add computed fields if needed
return players_data return players_data
# If items are DB models (from real repo) # If items are DB models (from real repo)
players_data = [] players_data = []
for player in query: for player in query:
player_dict = model_to_dict(player, recurse=not short_output) player_dict = model_to_dict(player, recurse=not short_output)
players_data.append(player_dict) players_data.append(player_dict)
return players_data return players_data
@classmethod @classmethod
def search_players( def search_players(
cls, cls,
query_str: str, query_str: str,
season: Optional[int] = None, season: Optional[int] = None,
limit: int = 10, limit: int = 10,
short_output: bool = False short_output: bool = False,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
""" """
Search players by name with fuzzy matching. Search players by name with fuzzy matching.
Args: Args:
query_str: Search query query_str: Search query
season: Season to search (None/0 for all) season: Season to search (None/0 for all)
limit: Maximum results limit: Maximum results
short_output: Exclude related data short_output: Exclude related data
Returns: Returns:
Dict with count and matching players Dict with count and matching players
""" """
@ -363,46 +358,49 @@ class PlayerService(BaseService):
# Convert to dicts if needed # Convert to dicts if needed
all_player_dicts = cls._query_to_player_dicts( all_player_dicts = cls._query_to_player_dicts(
InMemoryQueryResult(all_players), InMemoryQueryResult(all_players), short_output=True
short_output=True
) )
# Sort by relevance (exact matches first) # Sort by relevance (exact matches first)
exact_matches = [] exact_matches = []
partial_matches = [] partial_matches = []
for player in all_player_dicts: for player in all_player_dicts:
name_lower = player.get('name', '').lower() name_lower = player.get("name", "").lower()
if name_lower == query_lower: if name_lower == query_lower:
exact_matches.append(player) exact_matches.append(player)
elif query_lower in name_lower: elif query_lower in name_lower:
partial_matches.append(player) partial_matches.append(player)
# Sort by season within each group (newest first) # Sort by season within each group (newest first)
if search_all_seasons: if search_all_seasons:
exact_matches.sort(key=lambda p: p.get('season', 0), reverse=True) exact_matches.sort(key=lambda p: p.get("season", 0), reverse=True)
partial_matches.sort(key=lambda p: p.get('season', 0), reverse=True) partial_matches.sort(key=lambda p: p.get("season", 0), reverse=True)
# Combine and limit # Combine and limit
results = (exact_matches + partial_matches)[:limit] results = (exact_matches + partial_matches)[:limit]
return { return {
"count": len(results), "count": len(results),
"total_matches": len(exact_matches + partial_matches), "total_matches": len(exact_matches + partial_matches),
"all_seasons": search_all_seasons, "all_seasons": search_all_seasons,
"players": results "players": results,
} }
except Exception as e: except Exception as e:
cls.log_error("Error searching players", e) cls.log_error("Error searching players", e)
try: try:
raise HTTPException(status_code=500, detail=f"Error searching players: {str(e)}") raise HTTPException(
status_code=500, detail=f"Error searching players: {str(e)}"
)
except ImportError: except ImportError:
raise RuntimeError(f"Error searching players: {str(e)}") raise RuntimeError(f"Error searching players: {str(e)}")
@classmethod @classmethod
def get_player(cls, player_id: int, short_output: bool = False) -> Optional[Dict[str, Any]]: def get_player(
cls, player_id: int, short_output: bool = False
) -> Optional[Dict[str, Any]]:
"""Get a single player by ID.""" """Get a single player by ID."""
try: try:
repo = cls._get_player_repo() repo = cls._get_player_repo()
@ -413,26 +411,31 @@ class PlayerService(BaseService):
except Exception as e: except Exception as e:
cls.log_error(f"Error fetching player {player_id}", e) cls.log_error(f"Error fetching player {player_id}", e)
try: try:
raise HTTPException(status_code=500, detail=f"Error fetching player {player_id}: {str(e)}") raise HTTPException(
status_code=500,
detail=f"Error fetching player {player_id}: {str(e)}",
)
except ImportError: except ImportError:
raise RuntimeError(f"Error fetching player {player_id}: {str(e)}") raise RuntimeError(f"Error fetching player {player_id}: {str(e)}")
@classmethod @classmethod
def _player_to_dict(cls, player, recurse: bool = True) -> Dict[str, Any]: def _player_to_dict(cls, player, recurse: bool = True) -> Dict[str, Any]:
"""Convert player to dict.""" """Convert player to dict."""
# If already a dict, return as-is # If already a dict, return as-is
if isinstance(player, dict): if isinstance(player, dict):
return player return player
# Try to convert Peewee model # Try to convert Peewee model
try: try:
return model_to_dict(player, recurse=recurse) return model_to_dict(player, recurse=recurse)
except ImportError: except ImportError:
# Fall back to basic dict conversion # Fall back to basic dict conversion
return dict(player) return dict(player)
@classmethod @classmethod
def update_player(cls, player_id: int, data: Dict[str, Any], token: str) -> Dict[str, Any]: def update_player(
cls, player_id: int, data: Dict[str, Any], token: str
) -> Dict[str, Any]:
"""Update a player (full update via PUT).""" """Update a player (full update via PUT)."""
temp_service = cls() temp_service = cls()
temp_service.require_auth(token) temp_service.require_auth(token)
@ -441,7 +444,9 @@ class PlayerService(BaseService):
# Verify player exists # Verify player exists
repo = cls._get_player_repo() repo = cls._get_player_repo()
if not repo.get_by_id(player_id): if not repo.get_by_id(player_id):
raise HTTPException(status_code=404, detail=f"Player ID {player_id} not found") raise HTTPException(
status_code=404, detail=f"Player ID {player_id} not found"
)
# Execute update # Execute update
repo.update(data, player_id=player_id) repo.update(data, player_id=player_id)
@ -450,10 +455,14 @@ class PlayerService(BaseService):
except Exception as e: except Exception as e:
cls.log_error(f"Error updating player {player_id}", e) cls.log_error(f"Error updating player {player_id}", e)
raise HTTPException(status_code=500, detail=f"Error updating player {player_id}: {str(e)}") raise HTTPException(
status_code=500, detail=f"Error updating player {player_id}: {str(e)}"
)
@classmethod @classmethod
def patch_player(cls, player_id: int, data: Dict[str, Any], token: str) -> Dict[str, Any]: def patch_player(
cls, player_id: int, data: Dict[str, Any], token: str
) -> Dict[str, Any]:
"""Patch a player (partial update).""" """Patch a player (partial update)."""
temp_service = cls() temp_service = cls()
temp_service.require_auth(token) temp_service.require_auth(token)
@ -462,7 +471,9 @@ class PlayerService(BaseService):
repo = cls._get_player_repo() repo = cls._get_player_repo()
player = repo.get_by_id(player_id) player = repo.get_by_id(player_id)
if not player: if not player:
raise HTTPException(status_code=404, detail=f"Player ID {player_id} not found") raise HTTPException(
status_code=404, detail=f"Player ID {player_id} not found"
)
# Apply updates using repo # Apply updates using repo
repo.update(data, player_id=player_id) repo.update(data, player_id=player_id)
@ -471,10 +482,14 @@ class PlayerService(BaseService):
except Exception as e: except Exception as e:
cls.log_error(f"Error patching player {player_id}", e) cls.log_error(f"Error patching player {player_id}", e)
raise HTTPException(status_code=500, detail=f"Error patching player {player_id}: {str(e)}") raise HTTPException(
status_code=500, detail=f"Error patching player {player_id}: {str(e)}"
)
@classmethod @classmethod
def create_players(cls, players_data: List[Dict[str, Any]], token: str) -> Dict[str, Any]: def create_players(
cls, players_data: List[Dict[str, Any]], token: str
) -> Dict[str, Any]:
"""Create multiple players.""" """Create multiple players."""
temp_service = cls() temp_service = cls()
temp_service.require_auth(token) temp_service.require_auth(token)
@ -484,13 +499,12 @@ class PlayerService(BaseService):
repo = cls._get_player_repo() repo = cls._get_player_repo()
for player in players_data: for player in players_data:
dupe = repo.get_or_none( dupe = repo.get_or_none(
season=player.get("season"), season=player.get("season"), name=player.get("name")
name=player.get("name")
) )
if dupe: if dupe:
raise HTTPException( raise HTTPException(
status_code=500, status_code=500,
detail=f"Player {player.get('name')} already exists in Season {player.get('season')}" detail=f"Player {player.get('name')} already exists in Season {player.get('season')}",
) )
# Insert in batches # Insert in batches
@ -499,9 +513,11 @@ class PlayerService(BaseService):
return {"message": f"Inserted {len(players_data)} players"} return {"message": f"Inserted {len(players_data)} players"}
except Exception as e: except Exception as e:
cls.log_error(f"Error creating players", e) cls.log_error("Error creating players", e)
raise HTTPException(status_code=500, detail=f"Error creating players: {str(e)}") raise HTTPException(
status_code=500, detail=f"Error creating players: {str(e)}"
)
@classmethod @classmethod
def delete_player(cls, player_id: int, token: str) -> Dict[str, str]: def delete_player(cls, player_id: int, token: str) -> Dict[str, str]:
"""Delete a player.""" """Delete a player."""
@ -511,7 +527,9 @@ class PlayerService(BaseService):
try: try:
repo = cls._get_player_repo() repo = cls._get_player_repo()
if not repo.get_by_id(player_id): if not repo.get_by_id(player_id):
raise HTTPException(status_code=404, detail=f"Player ID {player_id} not found") raise HTTPException(
status_code=404, detail=f"Player ID {player_id} not found"
)
repo.delete_by_id(player_id) repo.delete_by_id(player_id)
@ -519,8 +537,10 @@ class PlayerService(BaseService):
except Exception as e: except Exception as e:
cls.log_error(f"Error deleting player {player_id}", e) cls.log_error(f"Error deleting player {player_id}", e)
raise HTTPException(status_code=500, detail=f"Error deleting player {player_id}: {str(e)}") raise HTTPException(
status_code=500, detail=f"Error deleting player {player_id}: {str(e)}"
)
@classmethod @classmethod
def _format_player_csv(cls, players: List[Dict]) -> str: def _format_player_csv(cls, players: List[Dict]) -> str:
"""Format player list as CSV - works with both real DB and mocks.""" """Format player list as CSV - works with both real DB and mocks."""
@ -543,52 +563,52 @@ class InMemoryQueryResult:
In-memory query result for mock repositories. In-memory query result for mock repositories.
Supports filtering, sorting, and iteration. Supports filtering, sorting, and iteration.
""" """
def __init__(self, items: List[Dict[str, Any]]): def __init__(self, items: List[Dict[str, Any]]):
self._items = list(items) self._items = list(items)
def where(self, *conditions) -> 'InMemoryQueryResult': def where(self, *conditions) -> "InMemoryQueryResult":
"""Apply filter conditions (no-op for compatibility).""" """Apply filter conditions (no-op for compatibility)."""
return self return self
def order_by(self, *fields) -> 'InMemoryQueryResult': def order_by(self, *fields) -> "InMemoryQueryResult":
"""Apply sort (no-op, sorting done by service).""" """Apply sort (no-op, sorting done by service)."""
return self return self
def count(self) -> int: def count(self) -> int:
return len(self._items) return len(self._items)
def __iter__(self): def __iter__(self):
return iter(self._items) return iter(self._items)
def __len__(self): def __len__(self):
return len(self._items) return len(self._items)
def __getitem__(self, index): def __getitem__(self, index):
return self._items[index] return self._items[index]
class RealPlayerRepository: class RealPlayerRepository:
"""Real database repository implementation.""" """Real database repository implementation."""
def __init__(self, model_class): def __init__(self, model_class):
self._model = model_class self._model = model_class
def select_season(self, season: int): def select_season(self, season: int):
"""Return query for season.""" """Return query for season."""
return self._model.select().where(self._model.season == season) return self._model.select().where(self._model.season == season)
def get_by_id(self, player_id: int): def get_by_id(self, player_id: int):
"""Get player by ID.""" """Get player by ID."""
return self._model.get_or_none(self._model.id == player_id) return self._model.get_or_none(self._model.id == player_id)
def get_or_none(self, **conditions): def get_or_none(self, **conditions):
"""Get player matching conditions.""" """Get player matching conditions."""
try: try:
return self._model.get_or_none(**conditions) return self._model.get_or_none(**conditions)
except Exception: except Exception:
return None return None
def update(self, data: Dict, player_id: int) -> int: def update(self, data: Dict, player_id: int) -> int:
"""Update player.""" """Update player."""
return Player.update(**data).where(Player.id == player_id).execute() return Player.update(**data).where(Player.id == player_id).execute()

139
app/services/team_service.py
View File

@ -10,6 +10,7 @@ import copy
import logging import logging
from typing import List, Optional, Dict, Any, Literal, TYPE_CHECKING from typing import List, Optional, Dict, Any, Literal, TYPE_CHECKING
from playhouse.shortcuts import model_to_dict
from .base import BaseService from .base import BaseService
from .interfaces import AbstractTeamRepository from .interfaces import AbstractTeamRepository
@ -28,17 +29,14 @@ except ImportError:
self.detail = detail self.detail = detail
super().__init__(detail) super().__init__(detail)
logger = logging.getLogger('discord_app')
logger = logging.getLogger("discord_app")
class TeamService(BaseService): class TeamService(BaseService):
"""Service for team-related operations.""" """Service for team-related operations."""
cache_patterns = [ cache_patterns = ["teams*", "team*", "team-roster*"]
"teams*",
"team*",
"team-roster*"
]
# Class-level repository for dependency injection # Class-level repository for dependency injection
_injected_repo: Optional[AbstractTeamRepository] = None _injected_repo: Optional[AbstractTeamRepository] = None
@ -46,8 +44,8 @@ class TeamService(BaseService):
def __init__( def __init__(
self, self,
team_repo: Optional[AbstractTeamRepository] = None, team_repo: Optional[AbstractTeamRepository] = None,
config: Optional['ServiceConfig'] = None, config: Optional["ServiceConfig"] = None,
**kwargs **kwargs,
): ):
""" """
Initialize TeamService with optional repository. Initialize TeamService with optional repository.
@ -75,11 +73,12 @@ class TeamService(BaseService):
return cls._get_real_repo() return cls._get_real_repo()
@classmethod @classmethod
def _get_real_repo(cls) -> 'RealTeamRepository': def _get_real_repo(cls) -> "RealTeamRepository":
"""Get a real DB repository for production use.""" """Get a real DB repository for production use."""
from ..db_engine import Team # Lazy import to avoid loading DB in tests from ..db_engine import Team # Lazy import to avoid loading DB in tests
return RealTeamRepository(Team) return RealTeamRepository(Team)
@classmethod @classmethod
def get_teams( def get_teams(
cls, cls,
@ -89,7 +88,7 @@ class TeamService(BaseService):
team_abbrev: Optional[List[str]] = None, team_abbrev: Optional[List[str]] = None,
active_only: bool = False, active_only: bool = False,
short_output: bool = False, short_output: bool = False,
as_csv: bool = False as_csv: bool = False,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
""" """
Get teams with filtering. Get teams with filtering.
@ -118,21 +117,27 @@ class TeamService(BaseService):
# Apply filters # Apply filters
if manager_id: if manager_id:
teams_list = [t for t in teams_list teams_list = [
if cls._team_has_manager(t, manager_id)] t for t in teams_list if cls._team_has_manager(t, manager_id)
]
if owner_id: if owner_id:
teams_list = [t for t in teams_list teams_list = [t for t in teams_list if cls._team_has_owner(t, owner_id)]
if cls._team_has_owner(t, owner_id)]
if team_abbrev: if team_abbrev:
abbrev_list = [x.lower() for x in team_abbrev] abbrev_list = [x.lower() for x in team_abbrev]
teams_list = [t for t in teams_list teams_list = [
if cls._get_team_field(t, 'abbrev', '').lower() in abbrev_list] t
for t in teams_list
if cls._get_team_field(t, "abbrev", "").lower() in abbrev_list
]
if active_only: if active_only:
teams_list = [t for t in teams_list teams_list = [
if not cls._get_team_field(t, 'abbrev', '').endswith(('IL', 'MiL'))] t
for t in teams_list
if not cls._get_team_field(t, "abbrev", "").endswith(("IL", "MiL"))
]
# Convert to dicts # Convert to dicts
teams_data = [cls._team_to_dict(t, short_output) for t in teams_list] teams_data = [cls._team_to_dict(t, short_output) for t in teams_list]
@ -140,18 +145,15 @@ class TeamService(BaseService):
if as_csv: if as_csv:
return cls._format_team_csv(teams_data) return cls._format_team_csv(teams_data)
return { return {"count": len(teams_data), "teams": teams_data}
"count": len(teams_data),
"teams": teams_data
}
except Exception as e: except Exception as e:
temp_service = cls() temp_service = cls()
temp_service.handle_error(f"Error fetching teams", e) temp_service.handle_error("Error fetching teams", e)
finally: finally:
temp_service = cls() temp_service = cls()
temp_service.close_db() temp_service.close_db()
@classmethod @classmethod
def get_team(cls, team_id: int) -> Optional[Dict[str, Any]]: def get_team(cls, team_id: int) -> Optional[Dict[str, Any]]:
"""Get a single team by ID.""" """Get a single team by ID."""
@ -167,13 +169,10 @@ class TeamService(BaseService):
finally: finally:
temp_service = cls() temp_service = cls()
temp_service.close_db() temp_service.close_db()
@classmethod @classmethod
def get_team_roster( def get_team_roster(
cls, cls, team_id: int, which: Literal["current", "next"], sort: Optional[str] = None
team_id: int,
which: Literal['current', 'next'],
sort: Optional[str] = None
) -> Dict[str, Any]: ) -> Dict[str, Any]:
""" """
Get team roster with IL lists. Get team roster with IL lists.
@ -192,26 +191,28 @@ class TeamService(BaseService):
team = Team.get_by_id(team_id) team = Team.get_by_id(team_id)
if which == 'current': if which == "current":
full_roster = team.get_this_week() full_roster = team.get_this_week()
else: else:
full_roster = team.get_next_week() full_roster = team.get_next_week()
# Deep copy and convert to dicts # Deep copy and convert to dicts
result = { result = {
'active': {'players': []}, "active": {"players": []},
'shortil': {'players': []}, "shortil": {"players": []},
'longil': {'players': []} "longil": {"players": []},
} }
for section in ['active', 'shortil', 'longil']: for section in ["active", "shortil", "longil"]:
players = copy.deepcopy(full_roster[section]['players']) players = copy.deepcopy(full_roster[section]["players"])
result[section]['players'] = [model_to_dict(p) for p in players] result[section]["players"] = [model_to_dict(p) for p in players]
# Apply sorting # Apply sorting
if sort == 'wara-desc': if sort == "wara-desc":
for section in ['active', 'shortil', 'longil']: for section in ["active", "shortil", "longil"]:
result[section]['players'].sort(key=lambda p: p.get("wara", 0), reverse=True) result[section]["players"].sort(
key=lambda p: p.get("wara", 0), reverse=True
)
return result return result
@ -221,9 +222,11 @@ class TeamService(BaseService):
finally: finally:
temp_service = cls() temp_service = cls()
temp_service.close_db() temp_service.close_db()
@classmethod @classmethod
def update_team(cls, team_id: int, data: Dict[str, Any], token: str) -> Dict[str, Any]: def update_team(
cls, team_id: int, data: Dict[str, Any], token: str
) -> Dict[str, Any]:
"""Update a team (partial update).""" """Update a team (partial update)."""
temp_service = cls() temp_service = cls()
temp_service.require_auth(token) temp_service.require_auth(token)
@ -232,7 +235,9 @@ class TeamService(BaseService):
repo = cls._get_team_repo() repo = cls._get_team_repo()
team = repo.get_by_id(team_id) team = repo.get_by_id(team_id)
if not team: if not team:
raise HTTPException(status_code=404, detail=f"Team ID {team_id} not found") raise HTTPException(
status_code=404, detail=f"Team ID {team_id} not found"
)
# Apply updates using repo # Apply updates using repo
repo.update(data, team_id=team_id) repo.update(data, team_id=team_id)
@ -244,9 +249,11 @@ class TeamService(BaseService):
finally: finally:
temp_service.invalidate_related_cache(cls.cache_patterns) temp_service.invalidate_related_cache(cls.cache_patterns)
temp_service.close_db() temp_service.close_db()
@classmethod @classmethod
def create_teams(cls, teams_data: List[Dict[str, Any]], token: str) -> Dict[str, str]: def create_teams(
cls, teams_data: List[Dict[str, Any]], token: str
) -> Dict[str, str]:
"""Create multiple teams.""" """Create multiple teams."""
temp_service = cls() temp_service = cls()
temp_service.require_auth(token) temp_service.require_auth(token)
@ -256,13 +263,12 @@ class TeamService(BaseService):
repo = cls._get_team_repo() repo = cls._get_team_repo()
for team in teams_data: for team in teams_data:
dupe = repo.get_or_none( dupe = repo.get_or_none(
season=team.get("season"), season=team.get("season"), abbrev=team.get("abbrev")
abbrev=team.get("abbrev")
) )
if dupe: if dupe:
raise HTTPException( raise HTTPException(
status_code=500, status_code=500,
detail=f"Team {team.get('abbrev')} already exists in Season {team.get('season')}" detail=f"Team {team.get('abbrev')} already exists in Season {team.get('season')}",
) )
# Insert teams # Insert teams
@ -271,11 +277,11 @@ class TeamService(BaseService):
return {"message": f"Inserted {len(teams_data)} teams"} return {"message": f"Inserted {len(teams_data)} teams"}
except Exception as e: except Exception as e:
temp_service.handle_error(f"Error creating teams", e) temp_service.handle_error("Error creating teams", e)
finally: finally:
temp_service.invalidate_related_cache(cls.cache_patterns) temp_service.invalidate_related_cache(cls.cache_patterns)
temp_service.close_db() temp_service.close_db()
@classmethod @classmethod
def delete_team(cls, team_id: int, token: str) -> Dict[str, str]: def delete_team(cls, team_id: int, token: str) -> Dict[str, str]:
"""Delete a team.""" """Delete a team."""
@ -285,7 +291,9 @@ class TeamService(BaseService):
try: try:
repo = cls._get_team_repo() repo = cls._get_team_repo()
if not repo.get_by_id(team_id): if not repo.get_by_id(team_id):
raise HTTPException(status_code=404, detail=f"Team ID {team_id} not found") raise HTTPException(
status_code=404, detail=f"Team ID {team_id} not found"
)
repo.delete_by_id(team_id) repo.delete_by_id(team_id)
@ -301,17 +309,25 @@ class TeamService(BaseService):
@classmethod @classmethod
def _team_has_manager(cls, team, manager_ids: List[int]) -> bool: def _team_has_manager(cls, team, manager_ids: List[int]) -> bool:
"""Check if team has any of the specified managers.""" """Check if team has any of the specified managers."""
team_dict = team if isinstance(team, dict) else cls._team_to_dict(team, short_output=True) team_dict = (
manager1 = team_dict.get('manager1_id') team
manager2 = team_dict.get('manager2_id') if isinstance(team, dict)
else cls._team_to_dict(team, short_output=True)
)
manager1 = team_dict.get("manager1_id")
manager2 = team_dict.get("manager2_id")
return manager1 in manager_ids or manager2 in manager_ids return manager1 in manager_ids or manager2 in manager_ids
@classmethod @classmethod
def _team_has_owner(cls, team, owner_ids: List[int]) -> bool: def _team_has_owner(cls, team, owner_ids: List[int]) -> bool:
"""Check if team has any of the specified owners.""" """Check if team has any of the specified owners."""
team_dict = team if isinstance(team, dict) else cls._team_to_dict(team, short_output=True) team_dict = (
gmid = team_dict.get('gmid') team
gmid2 = team_dict.get('gmid2') if isinstance(team, dict)
else cls._team_to_dict(team, short_output=True)
)
gmid = team_dict.get("gmid")
gmid2 = team_dict.get("gmid2")
return gmid in owner_ids or gmid2 in owner_ids return gmid in owner_ids or gmid2 in owner_ids
@classmethod @classmethod
@ -341,14 +357,16 @@ class TeamService(BaseService):
from ..db_engine import query_to_csv, Team # Lazy import - CSV needs DB from ..db_engine import query_to_csv, Team # Lazy import - CSV needs DB
# Get team IDs from the list # Get team IDs from the list
team_ids = [t.get('id') for t in teams if t.get('id')] team_ids = [t.get("id") for t in teams if t.get("id")]
if not team_ids: if not team_ids:
return "" return ""
# Query for CSV formatting # Query for CSV formatting
query = Team.select().where(Team.id << team_ids) query = Team.select().where(Team.id << team_ids)
return query_to_csv(query, exclude=[Team.division_legacy, Team.mascot, Team.gsheet]) return query_to_csv(
query, exclude=[Team.division_legacy, Team.mascot, Team.gsheet]
)
class RealTeamRepository: class RealTeamRepository:
@ -377,11 +395,13 @@ class RealTeamRepository:
def update(self, data: Dict, team_id: int) -> int: def update(self, data: Dict, team_id: int) -> int:
"""Update team.""" """Update team."""
from ..db_engine import Team # Lazy import - only used in production from ..db_engine import Team # Lazy import - only used in production
return Team.update(**data).where(Team.id == team_id).execute() return Team.update(**data).where(Team.id == team_id).execute()
def insert_many(self, data: List[Dict]) -> int: def insert_many(self, data: List[Dict]) -> int:
"""Insert multiple teams.""" """Insert multiple teams."""
from ..db_engine import Team, db # Lazy import - only used in production from ..db_engine import Team, db # Lazy import - only used in production
with db.atomic(): with db.atomic():
Team.insert_many(data).on_conflict_ignore().execute() Team.insert_many(data).on_conflict_ignore().execute()
return len(data) return len(data)
@ -389,4 +409,5 @@ class RealTeamRepository:
def delete_by_id(self, team_id: int) -> int: def delete_by_id(self, team_id: int) -> int:
"""Delete team by ID.""" """Delete team by ID."""
from ..db_engine import Team # Lazy import - only used in production from ..db_engine import Team # Lazy import - only used in production
return Team.delete().where(Team.id == team_id).execute() return Team.delete().where(Team.id == team_id).execute()