nhl_scrabble
NHL Roster Scrabble Score Analyzer.
A tool for fetching NHL roster data and calculating Scrabble scores for player names.
1"""NHL Roster Scrabble Score Analyzer. 2 3A tool for fetching NHL roster data and calculating Scrabble scores for player names. 4""" 5 6__version__ = "2.0.0" 7__author__ = "Brandon Perkins" 8 9from nhl_scrabble.api.nhl_client import NHLApiClient 10from nhl_scrabble.scoring.scrabble import ScrabbleScorer 11from nhl_scrabble.validators import ValidationError 12 13__all__ = [ 14 "NHLApiClient", 15 "ScrabbleScorer", 16 "ValidationError", 17 "__version__", 18]
47class NHLApiClient: 48 """Client for interacting with the NHL API. 49 50 This client provides methods to fetch team standings and roster data 51 from the official NHL API with built-in retry logic, rate limiting, 52 SSRF protection, DoS prevention, and enforced SSL/TLS certificate verification. 53 54 SSL/TLS Security: 55 - Certificate verification is always enabled and cannot be disabled 56 - Uses certifi CA bundle for up-to-date certificate authorities 57 - SSL errors are caught and logged for security monitoring 58 59 DoS Prevention: 60 - Circuit breaker pattern to prevent cascading failures 61 - Connection pool limits to prevent resource exhaustion 62 - Configurable failure thresholds and timeouts 63 64 Attributes: 65 base_url: Base URL for the NHL API (SSRF-validated) 66 timeout: Request timeout in seconds 67 retries: Number of retry attempts for failed requests 68 rate_limiter: Token bucket rate limiter for API requests 69 circuit_breaker: Circuit breaker for DoS prevention 70 ca_bundle: Path to CA bundle for SSL verification (uses certifi) 71 """ 72 73 BASE_URL = "https://api-web.nhle.com/v1" # Default base URL 74 _instances: ClassVar[set[weakref.ref[Any]]] = set() # Track all instances for cleanup 75 76 @classmethod 77 def _cleanup_callback(cls, ref: weakref.ref[Any]) -> None: 78 """Remove dead instance from tracking set. 79 80 Args: 81 ref: Weak reference to the instance being garbage collected. 82 """ 83 cls._instances.discard(ref) 84 85 @classmethod 86 def _cleanup_all(cls) -> None: 87 """Close all remaining open sessions at program exit (safety net).""" 88 alive_instances = [ref() for ref in cls._instances if ref() is not None] 89 if alive_instances: 90 logger.warning( 91 f"Cleaning up {len(alive_instances)} unclosed NHLApiClient session(s) at exit" 92 ) 93 for instance in alive_instances: 94 if instance and not instance._closed: # noqa: SLF001 95 instance.close() 96 97 def __init__( # noqa: PLR0913 98 self, 99 base_url: str | None = None, 100 timeout: int = 10, 101 retries: int = 3, 102 rate_limit_max_requests: int = 30, 103 rate_limit_window: float = 60.0, 104 backoff_factor: float = 2.0, 105 max_backoff: float = 30.0, 106 cache_enabled: bool = True, 107 cache_expiry: int = 3600, 108 verify_ssl: bool = True, 109 dos_max_connections: int = 10, 110 dos_max_per_host: int = 5, 111 dos_circuit_breaker_threshold: int = 5, 112 dos_circuit_breaker_timeout: float = 60.0, 113 ) -> None: 114 """Initialize the NHL API client. 115 116 Args: 117 base_url: Base URL for NHL API (default: https://api-web.nhle.com/v1). 118 Will be validated for SSRF protection on first request. 119 timeout: Request timeout in seconds (default: 10) 120 retries: Number of retry attempts for failed requests (default: 3) 121 rate_limit_max_requests: Maximum requests per time window (default: 30) 122 rate_limit_window: Time window for rate limiting in seconds (default: 60.0) 123 backoff_factor: Exponential backoff multiplier (default: 2.0) 124 max_backoff: Maximum backoff delay in seconds (default: 30.0) 125 cache_enabled: Enable HTTP caching (default: True) 126 cache_expiry: Cache expiration in seconds (default: 3600 = 1 hour) 127 verify_ssl: SSL verification (must be True, cannot be disabled for security) 128 dos_max_connections: Maximum connection pool connections (default: 10) 129 dos_max_per_host: Maximum connections per host (default: 5) 130 dos_circuit_breaker_threshold: Circuit breaker failure threshold (default: 5) 131 dos_circuit_breaker_timeout: Circuit breaker timeout in seconds (default: 60.0) 132 133 Raises: 134 NHLApiError: If base_url fails SSRF protection validation 135 ValueError: If verify_ssl is False (SSL verification cannot be disabled) 136 """ 137 # Initialize state tracking FIRST (before any potential exceptions) 138 # This prevents AttributeError in __del__ if __init__ fails 139 self._closed = False 140 141 # Enforce SSL verification - cannot be disabled 142 if not verify_ssl: 143 error_msg = "SSL verification cannot be disabled for security reasons" 144 logger.error(error_msg) 145 raise ValueError(error_msg) 146 147 # Use provided base_url or fall back to class default 148 self.base_url = base_url or self.BASE_URL 149 150 self.timeout = timeout 151 self.retries = retries 152 self.backoff_factor = backoff_factor 153 self.max_backoff = max_backoff 154 self.cache_enabled = cache_enabled 155 self.cache_expiry = cache_expiry 156 157 # Initialize rate limiter 158 self.rate_limiter = RateLimiter( 159 max_requests=rate_limit_max_requests, time_window=rate_limit_window 160 ) 161 logger.info( 162 f"Rate limiter initialized: {rate_limit_max_requests} requests per {rate_limit_window}s" 163 ) 164 165 # Initialize circuit breaker for DoS prevention 166 self.circuit_breaker = CircuitBreaker( 167 failure_threshold=dos_circuit_breaker_threshold, 168 timeout=dos_circuit_breaker_timeout, 169 expected_exception=( 170 requests.exceptions.RequestException, 171 NHLApiError, 172 ), 173 ) 174 logger.info( 175 f"Circuit breaker initialized: threshold={dos_circuit_breaker_threshold}, " 176 f"timeout={dos_circuit_breaker_timeout}s" 177 ) 178 179 # Use certifi CA bundle for SSL verification 180 self.ca_bundle = certifi.where() 181 logger.debug(f"Using CA bundle for SSL verification: {self.ca_bundle}") 182 183 # Session can be either CachedSession or regular Session 184 self.session: requests_cache.CachedSession | requests.Session 185 if cache_enabled: 186 # Create cached session 187 self.session = requests_cache.CachedSession( 188 cache_name=".nhl_cache", 189 backend="sqlite", 190 expire_after=timedelta(seconds=cache_expiry), 191 allowable_codes=[200], # Only cache successful responses 192 allowable_methods=["GET"], 193 cache_control=True, # Respect Cache-Control headers 194 ) 195 if logger.isEnabledFor(logging.DEBUG): 196 logger.debug(f"HTTP caching enabled (expiry: {cache_expiry}s)") 197 else: 198 self.session = requests.Session() 199 logger.debug("HTTP caching disabled") 200 201 # Configure connection pool limits for DoS protection 202 adapter = HTTPAdapter( 203 pool_connections=dos_max_connections, 204 pool_maxsize=dos_max_per_host, 205 max_retries=0, # We handle retries ourselves via @retry decorator 206 ) 207 self.session.mount("https://", adapter) 208 self.session.mount("http://", adapter) 209 logger.info( 210 f"Connection pool configured: max_connections={dos_max_connections}, " 211 f"max_per_host={dos_max_per_host}" 212 ) 213 214 self.session.headers.update({"User-Agent": "NHL-Scrabble/2.0"}) 215 216 # Register instance for cleanup at exit (safety net) 217 self._instances.add(weakref.ref(self, self._cleanup_callback)) 218 atexit.register(self._cleanup_all) 219 220 def __del__(self) -> None: 221 """Destructor - close session if not already closed (safety net).""" 222 if not self._closed: 223 logger.warning( 224 "NHLApiClient session was not explicitly closed - cleaning up in destructor" 225 ) 226 self.close() 227 228 def _validate_request_url(self, url: str) -> None: 229 """Validate URL with SSRF protection before making request. 230 231 Args: 232 url: Full URL to validate 233 234 Raises: 235 NHLApiError: If URL fails SSRF protection validation 236 """ 237 try: 238 validate_url_for_ssrf(url, allow_private=False) 239 except SSRFProtectionError as e: 240 logger.error(f"SSRF protection blocked request to {url}: {e}") 241 raise NHLApiError(f"Request blocked by security protection: {e}") from e 242 243 def _get_retry_after(self, response: requests.Response) -> float: 244 """Extract Retry-After header value from 429 response. 245 246 Args: 247 response: HTTP response with 429 status 248 249 Returns: 250 Seconds to wait before retry 251 252 Examples: 253 >>> client = NHLApiClient() 254 >>> from unittest.mock import Mock 255 >>> response = Mock() 256 >>> response.headers = {"Retry-After": "60"} 257 >>> client._get_retry_after(response) 258 60.0 259 """ 260 retry_after = response.headers.get("Retry-After") 261 262 if retry_after: 263 try: 264 # Try as integer (seconds) 265 return float(retry_after) 266 except ValueError: 267 # Could be HTTP date format, but uncommon for 429 268 # Default to exponential backoff 269 pass 270 271 # No Retry-After header, use exponential backoff 272 # Start with 1 second 273 return 1.0 274 275 def _calculate_backoff_delay(self, attempt: int, retry_after: int | None = None) -> float: 276 """Calculate backoff delay with exponential backoff and jitter. 277 278 Args: 279 attempt: Current attempt number (0-indexed) 280 retry_after: Optional Retry-After header value from 429 response 281 282 Returns: 283 Delay in seconds with jitter applied 284 285 Examples: 286 >>> client = NHLApiClient() 287 >>> client._calculate_backoff_delay(0) # First retry 288 0.75 # ~1.0 * (2.0 ** 0) with ±25% jitter 289 >>> client._calculate_backoff_delay(3) # Fourth retry 290 6.5 # ~8.0 * (2.0 ** 3) with ±25% jitter, capped at max_backoff 291 """ 292 if retry_after is not None: 293 # Respect Retry-After header from API (429 responses) 294 return min(float(retry_after), self.max_backoff) 295 296 # Exponential backoff: base_delay * (backoff_factor ** attempt) 297 base_delay = 1.0 298 delay = min(base_delay * (self.backoff_factor**attempt), self.max_backoff) 299 300 # Add jitter: randomize ±25% to prevent thundering herd 301 # Safe: Using random for jitter, not cryptography 302 jitter = delay * 0.25 303 delay = delay + random.uniform(-jitter, jitter) # noqa: S311 304 305 return max(0, delay) 306 307 def _is_url_cached(self, url: str) -> bool: 308 """Check if a URL response is cached and not expired. 309 310 Args: 311 url: The URL to check 312 313 Returns: 314 True if the URL response is cached and valid, False otherwise 315 316 Examples: 317 >>> client = NHLApiClient(cache_enabled=True) 318 >>> client._is_url_cached("https://api-web.nhle.com/v1/roster/TOR/current") 319 False # Not cached initially 320 """ 321 if not self.cache_enabled: 322 return False 323 324 if not hasattr(self.session, "cache"): 325 return False 326 327 try: 328 # Check if URL is in cache using has_url() method (requests-cache 1.0+) 329 if hasattr(self.session.cache, "has_url"): 330 return self.session.cache.has_url(url) # type: ignore[no-any-return] 331 332 # Fallback: check using contains() method 333 if hasattr(self.session.cache, "contains"): 334 return self.session.cache.contains(url=url) # type: ignore[no-any-return] 335 336 # If no cache checking method available, assume not cached 337 return False 338 except Exception: # noqa: BLE001 339 # If anything goes wrong checking cache, assume not cached 340 # This ensures we always apply rate limiting if uncertain 341 return False 342 343 def get_teams(self, season: str | None = None) -> dict[str, dict[str, str]]: 344 """Fetch all NHL teams with division and conference information. 345 346 This method uses the retry decorator to automatically retry on network errors. 347 The URL is validated with SSRF protection before making the request. 348 349 Args: 350 season: Optional season in format 'YYYYYYYY' (e.g., '20222023' for 2022-23). 351 If None, fetches current season data. 352 353 Returns: 354 Dictionary mapping team abbreviations to their metadata: 355 { 356 'TOR': {'division': 'Atlantic', 'conference': 'Eastern'}, 357 'MTL': {'division': 'Atlantic', 'conference': 'Eastern'}, 358 ... 359 } 360 361 Raises: 362 NHLApiConnectionError: If unable to connect to the API 363 NHLApiError: For other API errors, including SSRF protection blocks 364 365 Examples: 366 >>> client = NHLApiClient() 367 >>> teams = client.get_teams() 368 >>> "TOR" in teams 369 True 370 >>> teams_2022 = client.get_teams(season="20222023") 371 >>> "TOR" in teams_2022 372 True 373 """ 374 # Use season-specific endpoint or current season endpoint 375 endpoint = f"standings/{season}" if season else "standings/now" 376 url = f"{self.base_url}/{endpoint}" 377 378 season_desc = f"season {season}" if season else "current season" 379 logger.info(f"Fetching NHL teams from standings endpoint for {season_desc}") 380 381 # Validate URL with SSRF protection 382 self._validate_request_url(url) 383 384 @retry( 385 max_attempts=self.retries, 386 backoff_factor=self.backoff_factor, 387 max_backoff=self.max_backoff, 388 exceptions=( 389 requests.exceptions.Timeout, 390 requests.exceptions.ConnectionError, 391 ), 392 ) 393 def _fetch_teams() -> dict[str, dict[str, str]]: 394 """Fetch teams with retry logic.""" 395 # Check if URL is cached 396 is_cached = self._is_url_cached(url) 397 398 # Only rate limit for actual API calls (not cached responses) 399 if not is_cached: 400 if logger.isEnabledFor(logging.DEBUG): 401 logger.debug("Rate limiting: acquiring token for teams request") 402 self.rate_limiter.acquire() 403 404 try: 405 response = self.session.get( 406 url, 407 timeout=self.timeout, 408 verify=self.ca_bundle, # Explicit SSL verification with certifi CA bundle 409 ) 410 411 # Handle rate limiting (429) 412 if response.status_code == 429: 413 retry_after = self._get_retry_after(response) 414 logger.warning(f"Rate limited (429). Waiting {retry_after}s before retry.") 415 time.sleep(retry_after) 416 # Raise to trigger retry 417 response.raise_for_status() 418 419 response.raise_for_status() 420 data = response.json() 421 422 teams_info: dict[str, dict[str, str]] = {} 423 for team in data["standings"]: 424 team_abbrev = team["teamAbbrev"]["default"] 425 teams_info[team_abbrev] = { 426 "division": team.get("divisionName", "Unknown"), 427 "conference": team.get("conferenceName", "Unknown"), 428 } 429 430 logger.info(f"Successfully fetched {len(teams_info)} teams") 431 432 # Log cache status 433 from_cache = ( 434 hasattr(response, "from_cache") 435 and isinstance(response.from_cache, bool) 436 and response.from_cache 437 ) 438 if from_cache: 439 logger.debug("Cache hit - skipped rate limiting") 440 else: 441 logger.debug("Real API request - rate limited") 442 443 return teams_info 444 445 except requests.exceptions.SSLError as e: 446 logger.error(f"SSL certificate verification failed for {url}: {e}") 447 raise NHLApiSSLError(f"SSL certificate verification failed for {url}: {e}") from e 448 except requests.exceptions.HTTPError as e: 449 logger.error(f"HTTP error while fetching teams: {e}") 450 raise NHLApiError(f"HTTP error: {e}") from e 451 except (KeyError, ValueError) as e: 452 logger.error(f"Error parsing teams response: {e}") 453 raise NHLApiError(f"Invalid API response format: {e}") from e 454 455 try: 456 # Wrap with circuit breaker for DoS prevention 457 return self.circuit_breaker.call(_fetch_teams) 458 except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: 459 # Convert to NHLApiConnectionError after retries exhausted 460 logger.error(f"Connection error after retries: {e}") 461 raise NHLApiConnectionError("Unable to connect to NHL API after retries") from e 462 463 def _sanitize_roster_player_names(self, roster_data: dict[str, Any]) -> None: 464 """Sanitize player names in roster data to prevent injection attacks. 465 466 Validates and sanitizes all player names (firstName and lastName) in the 467 roster data for all positions (forwards, defensemen, goalies). 468 469 Args: 470 roster_data: Roster data dictionary with forwards, defensemen, goalies 471 472 Raises: 473 NHLApiError: If player names contain invalid characters (potential attack) 474 475 Note: 476 Modifies roster_data in-place for efficiency 477 """ 478 for position in ["forwards", "defensemen", "goalies"]: 479 if position not in roster_data: 480 continue 481 482 for player in roster_data[position]: 483 # Validate and sanitize first name 484 if ( 485 "firstName" in player 486 and isinstance(player["firstName"], dict) 487 and "default" in player["firstName"] 488 ): 489 try: 490 player["firstName"]["default"] = validate_player_name( 491 player["firstName"]["default"] 492 ) 493 except ValidationError as e: 494 logger.warning(f"Invalid player first name in API response: {e}") 495 # Use sanitized version or skip 496 player["firstName"]["default"] = "Unknown" 497 498 # Validate and sanitize last name 499 if ( 500 "lastName" in player 501 and isinstance(player["lastName"], dict) 502 and "default" in player["lastName"] 503 ): 504 try: 505 player["lastName"]["default"] = validate_player_name( 506 player["lastName"]["default"] 507 ) 508 except ValidationError as e: 509 logger.warning(f"Invalid player last name in API response: {e}") 510 # Use sanitized version or skip 511 player["lastName"]["default"] = "Unknown" 512 513 def get_team_roster( # noqa: PLR0915 514 self, team_abbrev: str, season: str | None = None 515 ) -> dict[str, Any]: 516 """Fetch the roster for a specific team with input and response validation. 517 518 Validates team abbreviation before making API call and validates response 519 structure to prevent errors from malformed data. 520 521 The URL is validated with SSRF protection before making the request. 522 523 Args: 524 team_abbrev: Team abbreviation (e.g., 'TOR', 'MTL') 525 season: Optional season in format 'YYYYYYYY' (e.g., '20222023' for 2022-23). 526 If None, fetches current season roster. 527 528 Returns: 529 Dictionary containing roster data with 'forwards', 'defensemen', and 'goalies' keys 530 531 Raises: 532 ValidationError: If team abbreviation is invalid 533 NHLApiNotFoundError: If the roster is not found (404 response) 534 NHLApiConnectionError: If unable to connect to the API after all retries 535 NHLApiError: For other API errors, including SSRF protection blocks and malformed responses 536 537 Security: 538 - Validates team abbreviation to prevent injection attacks 539 - Validates response structure to prevent KeyError exceptions 540 - Sanitizes player names from API responses 541 - SSRF protection on all API requests 542 543 Examples: 544 >>> client = NHLApiClient() 545 >>> roster = client.get_team_roster("TOR") 546 >>> "forwards" in roster 547 True 548 >>> roster_2022 = client.get_team_roster("TOR", season="20222023") 549 >>> "forwards" in roster_2022 550 True 551 >>> client.get_team_roster("INVALID") 552 Traceback (most recent call last): 553 ValidationError: Team abbreviation must be 2-3 characters... 554 """ 555 # Validate team abbreviation BEFORE making API call 556 try: 557 validated_abbrev = validate_team_abbreviation(team_abbrev) 558 except ValidationError: 559 # Re-raise validation errors for consistency with other API errors 560 logger.error(f"Invalid team abbreviation: {team_abbrev}") 561 raise 562 563 # Use season-specific endpoint or current season endpoint 564 endpoint = ( 565 f"roster/{validated_abbrev}/{season}" 566 if season 567 else f"roster/{validated_abbrev}/current" 568 ) 569 url = f"{self.base_url}/{endpoint}" 570 571 season_desc = f"season {season}" if season else "current season" 572 if logger.isEnabledFor(logging.DEBUG): 573 logger.debug(f"Fetching roster for {validated_abbrev} ({season_desc})") 574 575 # Validate URL with SSRF protection 576 self._validate_request_url(url) 577 578 def _fetch_roster() -> dict[str, Any]: # noqa: PLR0915 579 """Fetch roster with retry logic.""" 580 for attempt in range(self.retries): 581 try: 582 # Check if URL is cached 583 is_cached = self._is_url_cached(url) 584 585 # Only rate limit for actual API calls (not cached responses) 586 if not is_cached: 587 if logger.isEnabledFor(logging.DEBUG): 588 logger.debug(f"Rate limiting: acquiring token for {team_abbrev} roster") 589 self.rate_limiter.acquire() 590 591 response = self.session.get( 592 url, 593 timeout=self.timeout, 594 verify=self.ca_bundle, # Explicit SSL verification with certifi CA bundle 595 ) 596 597 if response.status_code == 404: 598 logger.warning(f"No roster data available for {team_abbrev}") 599 raise NHLApiNotFoundError(f"Roster not found for team: {team_abbrev}") 600 601 # Handle 429 rate limiting with exponential backoff 602 if response.status_code == 429: 603 if attempt < self.retries - 1: 604 retry_after = self._get_retry_after(response) 605 logger.warning( 606 f"Rate limited (429) for {team_abbrev} " 607 f"(attempt {attempt + 1}/{self.retries}), " 608 f"retrying in {retry_after:.2f}s..." 609 ) 610 time.sleep(retry_after) 611 continue 612 logger.error( 613 f"Rate limited (429) for {team_abbrev} after {self.retries} attempts" 614 ) 615 raise NHLApiConnectionError( 616 f"Rate limited after {self.retries} attempts" 617 ) from None 618 619 response.raise_for_status() 620 data = response.json() 621 622 # Validate response structure 623 try: 624 validate_api_response_structure( 625 data, 626 required_keys=["forwards", "defensemen", "goalies"], 627 context=f"Team roster response for {validated_abbrev}", 628 ) 629 except ValidationError as e: 630 logger.error( 631 f"Invalid roster response structure for {validated_abbrev}: {e}" 632 ) 633 raise NHLApiError(f"Invalid API response: {e}") from e 634 635 # Sanitize player names in response 636 self._sanitize_roster_player_names(data) 637 638 if logger.isEnabledFor(logging.DEBUG): 639 logger.debug( 640 f"Successfully fetched and validated roster for {validated_abbrev}" 641 ) 642 643 # Log cache status 644 from_cache = ( 645 hasattr(response, "from_cache") 646 and isinstance(response.from_cache, bool) 647 and response.from_cache 648 ) 649 if from_cache: 650 logger.debug("Cache hit - skipped rate limiting") 651 else: 652 logger.debug("Real API request - rate limited") 653 654 return data # type: ignore[no-any-return] 655 656 except requests.exceptions.Timeout: 657 if attempt < self.retries - 1: 658 backoff_delay = self._calculate_backoff_delay(attempt) 659 logger.warning( 660 f"Timeout fetching {team_abbrev} " 661 f"(attempt {attempt + 1}/{self.retries}), " 662 f"retrying in {backoff_delay:.2f}s..." 663 ) 664 time.sleep(backoff_delay) 665 else: 666 logger.error(f"Failed to fetch {team_abbrev} after {self.retries} attempts") 667 raise NHLApiConnectionError( 668 f"Request timed out after {self.retries} attempts" 669 ) from None 670 671 except requests.exceptions.SSLError as e: 672 # SSL errors should not be retried - certificate validation failure is permanent 673 logger.error(f"SSL certificate verification failed for {team_abbrev}: {e}") 674 raise NHLApiSSLError( 675 f"SSL certificate verification failed for {url}: {e}" 676 ) from e 677 678 except requests.exceptions.ConnectionError: 679 if attempt < self.retries - 1: 680 backoff_delay = self._calculate_backoff_delay(attempt) 681 logger.warning( 682 f"Connection error for {team_abbrev} " 683 f"(attempt {attempt + 1}/{self.retries}), " 684 f"retrying in {backoff_delay:.2f}s..." 685 ) 686 time.sleep(backoff_delay) 687 else: 688 logger.error(f"Failed to fetch {team_abbrev} after {self.retries} attempts") 689 raise NHLApiConnectionError( 690 f"Connection failed after {self.retries} attempts" 691 ) from None 692 693 except requests.exceptions.HTTPError as e: 694 logger.error(f"HTTP error fetching {team_abbrev}: {e}") 695 raise NHLApiError(f"HTTP error: {e}") from e 696 697 # This should never be reached as all paths above either return or raise 698 raise NHLApiError("Unexpected error: retry loop completed without returning data") 699 700 # Wrap with circuit breaker for DoS prevention 701 return self.circuit_breaker.call(_fetch_roster) 702 703 def get_rate_limit_stats(self) -> dict[str, Any]: 704 """Get rate limiter statistics. 705 706 Returns: 707 Dictionary with rate limiter statistics including: 708 - total_requests: Total requests made 709 - total_waits: Total times waited for tokens 710 - total_wait_time: Total time spent waiting 711 - average_wait: Average wait time per wait 712 - current_tokens: Current token count 713 - max_tokens: Maximum token capacity 714 715 Examples: 716 >>> client = NHLApiClient() 717 >>> stats = client.get_rate_limit_stats() 718 >>> "total_requests" in stats 719 True 720 """ 721 return self.rate_limiter.get_stats() 722 723 def clear_cache(self) -> None: 724 """Clear the HTTP cache.""" 725 if self.cache_enabled and hasattr(self.session, "cache"): 726 self.session.cache.clear() 727 logger.info("API cache cleared") 728 else: 729 logger.debug("Cache not available or caching disabled") 730 731 def close(self) -> None: 732 """Close the session and release resources.""" 733 if not self._closed and hasattr(self, "session"): 734 self.session.close() 735 self._closed = True 736 logger.debug("NHL API client session closed") 737 738 def __enter__(self) -> "NHLApiClient": 739 """Support context manager protocol.""" 740 return self 741 742 def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: 743 """Close session when exiting context manager.""" 744 self.close()
Client for interacting with the NHL API.
This client provides methods to fetch team standings and roster data from the official NHL API with built-in retry logic, rate limiting, SSRF protection, DoS prevention, and enforced SSL/TLS certificate verification.
SSL/TLS Security: - Certificate verification is always enabled and cannot be disabled - Uses certifi CA bundle for up-to-date certificate authorities - SSL errors are caught and logged for security monitoring
DoS Prevention: - Circuit breaker pattern to prevent cascading failures - Connection pool limits to prevent resource exhaustion - Configurable failure thresholds and timeouts
Attributes: base_url: Base URL for the NHL API (SSRF-validated) timeout: Request timeout in seconds retries: Number of retry attempts for failed requests rate_limiter: Token bucket rate limiter for API requests circuit_breaker: Circuit breaker for DoS prevention ca_bundle: Path to CA bundle for SSL verification (uses certifi)
97 def __init__( # noqa: PLR0913 98 self, 99 base_url: str | None = None, 100 timeout: int = 10, 101 retries: int = 3, 102 rate_limit_max_requests: int = 30, 103 rate_limit_window: float = 60.0, 104 backoff_factor: float = 2.0, 105 max_backoff: float = 30.0, 106 cache_enabled: bool = True, 107 cache_expiry: int = 3600, 108 verify_ssl: bool = True, 109 dos_max_connections: int = 10, 110 dos_max_per_host: int = 5, 111 dos_circuit_breaker_threshold: int = 5, 112 dos_circuit_breaker_timeout: float = 60.0, 113 ) -> None: 114 """Initialize the NHL API client. 115 116 Args: 117 base_url: Base URL for NHL API (default: https://api-web.nhle.com/v1). 118 Will be validated for SSRF protection on first request. 119 timeout: Request timeout in seconds (default: 10) 120 retries: Number of retry attempts for failed requests (default: 3) 121 rate_limit_max_requests: Maximum requests per time window (default: 30) 122 rate_limit_window: Time window for rate limiting in seconds (default: 60.0) 123 backoff_factor: Exponential backoff multiplier (default: 2.0) 124 max_backoff: Maximum backoff delay in seconds (default: 30.0) 125 cache_enabled: Enable HTTP caching (default: True) 126 cache_expiry: Cache expiration in seconds (default: 3600 = 1 hour) 127 verify_ssl: SSL verification (must be True, cannot be disabled for security) 128 dos_max_connections: Maximum connection pool connections (default: 10) 129 dos_max_per_host: Maximum connections per host (default: 5) 130 dos_circuit_breaker_threshold: Circuit breaker failure threshold (default: 5) 131 dos_circuit_breaker_timeout: Circuit breaker timeout in seconds (default: 60.0) 132 133 Raises: 134 NHLApiError: If base_url fails SSRF protection validation 135 ValueError: If verify_ssl is False (SSL verification cannot be disabled) 136 """ 137 # Initialize state tracking FIRST (before any potential exceptions) 138 # This prevents AttributeError in __del__ if __init__ fails 139 self._closed = False 140 141 # Enforce SSL verification - cannot be disabled 142 if not verify_ssl: 143 error_msg = "SSL verification cannot be disabled for security reasons" 144 logger.error(error_msg) 145 raise ValueError(error_msg) 146 147 # Use provided base_url or fall back to class default 148 self.base_url = base_url or self.BASE_URL 149 150 self.timeout = timeout 151 self.retries = retries 152 self.backoff_factor = backoff_factor 153 self.max_backoff = max_backoff 154 self.cache_enabled = cache_enabled 155 self.cache_expiry = cache_expiry 156 157 # Initialize rate limiter 158 self.rate_limiter = RateLimiter( 159 max_requests=rate_limit_max_requests, time_window=rate_limit_window 160 ) 161 logger.info( 162 f"Rate limiter initialized: {rate_limit_max_requests} requests per {rate_limit_window}s" 163 ) 164 165 # Initialize circuit breaker for DoS prevention 166 self.circuit_breaker = CircuitBreaker( 167 failure_threshold=dos_circuit_breaker_threshold, 168 timeout=dos_circuit_breaker_timeout, 169 expected_exception=( 170 requests.exceptions.RequestException, 171 NHLApiError, 172 ), 173 ) 174 logger.info( 175 f"Circuit breaker initialized: threshold={dos_circuit_breaker_threshold}, " 176 f"timeout={dos_circuit_breaker_timeout}s" 177 ) 178 179 # Use certifi CA bundle for SSL verification 180 self.ca_bundle = certifi.where() 181 logger.debug(f"Using CA bundle for SSL verification: {self.ca_bundle}") 182 183 # Session can be either CachedSession or regular Session 184 self.session: requests_cache.CachedSession | requests.Session 185 if cache_enabled: 186 # Create cached session 187 self.session = requests_cache.CachedSession( 188 cache_name=".nhl_cache", 189 backend="sqlite", 190 expire_after=timedelta(seconds=cache_expiry), 191 allowable_codes=[200], # Only cache successful responses 192 allowable_methods=["GET"], 193 cache_control=True, # Respect Cache-Control headers 194 ) 195 if logger.isEnabledFor(logging.DEBUG): 196 logger.debug(f"HTTP caching enabled (expiry: {cache_expiry}s)") 197 else: 198 self.session = requests.Session() 199 logger.debug("HTTP caching disabled") 200 201 # Configure connection pool limits for DoS protection 202 adapter = HTTPAdapter( 203 pool_connections=dos_max_connections, 204 pool_maxsize=dos_max_per_host, 205 max_retries=0, # We handle retries ourselves via @retry decorator 206 ) 207 self.session.mount("https://", adapter) 208 self.session.mount("http://", adapter) 209 logger.info( 210 f"Connection pool configured: max_connections={dos_max_connections}, " 211 f"max_per_host={dos_max_per_host}" 212 ) 213 214 self.session.headers.update({"User-Agent": "NHL-Scrabble/2.0"}) 215 216 # Register instance for cleanup at exit (safety net) 217 self._instances.add(weakref.ref(self, self._cleanup_callback)) 218 atexit.register(self._cleanup_all)
Initialize the NHL API client.
Args: base_url: Base URL for NHL API (default: https://api-web.nhle.com/v1). Will be validated for SSRF protection on first request. timeout: Request timeout in seconds (default: 10) retries: Number of retry attempts for failed requests (default: 3) rate_limit_max_requests: Maximum requests per time window (default: 30) rate_limit_window: Time window for rate limiting in seconds (default: 60.0) backoff_factor: Exponential backoff multiplier (default: 2.0) max_backoff: Maximum backoff delay in seconds (default: 30.0) cache_enabled: Enable HTTP caching (default: True) cache_expiry: Cache expiration in seconds (default: 3600 = 1 hour) verify_ssl: SSL verification (must be True, cannot be disabled for security) dos_max_connections: Maximum connection pool connections (default: 10) dos_max_per_host: Maximum connections per host (default: 5) dos_circuit_breaker_threshold: Circuit breaker failure threshold (default: 5) dos_circuit_breaker_timeout: Circuit breaker timeout in seconds (default: 60.0)
Raises: NHLApiError: If base_url fails SSRF protection validation ValueError: If verify_ssl is False (SSL verification cannot be disabled)
343 def get_teams(self, season: str | None = None) -> dict[str, dict[str, str]]: 344 """Fetch all NHL teams with division and conference information. 345 346 This method uses the retry decorator to automatically retry on network errors. 347 The URL is validated with SSRF protection before making the request. 348 349 Args: 350 season: Optional season in format 'YYYYYYYY' (e.g., '20222023' for 2022-23). 351 If None, fetches current season data. 352 353 Returns: 354 Dictionary mapping team abbreviations to their metadata: 355 { 356 'TOR': {'division': 'Atlantic', 'conference': 'Eastern'}, 357 'MTL': {'division': 'Atlantic', 'conference': 'Eastern'}, 358 ... 359 } 360 361 Raises: 362 NHLApiConnectionError: If unable to connect to the API 363 NHLApiError: For other API errors, including SSRF protection blocks 364 365 Examples: 366 >>> client = NHLApiClient() 367 >>> teams = client.get_teams() 368 >>> "TOR" in teams 369 True 370 >>> teams_2022 = client.get_teams(season="20222023") 371 >>> "TOR" in teams_2022 372 True 373 """ 374 # Use season-specific endpoint or current season endpoint 375 endpoint = f"standings/{season}" if season else "standings/now" 376 url = f"{self.base_url}/{endpoint}" 377 378 season_desc = f"season {season}" if season else "current season" 379 logger.info(f"Fetching NHL teams from standings endpoint for {season_desc}") 380 381 # Validate URL with SSRF protection 382 self._validate_request_url(url) 383 384 @retry( 385 max_attempts=self.retries, 386 backoff_factor=self.backoff_factor, 387 max_backoff=self.max_backoff, 388 exceptions=( 389 requests.exceptions.Timeout, 390 requests.exceptions.ConnectionError, 391 ), 392 ) 393 def _fetch_teams() -> dict[str, dict[str, str]]: 394 """Fetch teams with retry logic.""" 395 # Check if URL is cached 396 is_cached = self._is_url_cached(url) 397 398 # Only rate limit for actual API calls (not cached responses) 399 if not is_cached: 400 if logger.isEnabledFor(logging.DEBUG): 401 logger.debug("Rate limiting: acquiring token for teams request") 402 self.rate_limiter.acquire() 403 404 try: 405 response = self.session.get( 406 url, 407 timeout=self.timeout, 408 verify=self.ca_bundle, # Explicit SSL verification with certifi CA bundle 409 ) 410 411 # Handle rate limiting (429) 412 if response.status_code == 429: 413 retry_after = self._get_retry_after(response) 414 logger.warning(f"Rate limited (429). Waiting {retry_after}s before retry.") 415 time.sleep(retry_after) 416 # Raise to trigger retry 417 response.raise_for_status() 418 419 response.raise_for_status() 420 data = response.json() 421 422 teams_info: dict[str, dict[str, str]] = {} 423 for team in data["standings"]: 424 team_abbrev = team["teamAbbrev"]["default"] 425 teams_info[team_abbrev] = { 426 "division": team.get("divisionName", "Unknown"), 427 "conference": team.get("conferenceName", "Unknown"), 428 } 429 430 logger.info(f"Successfully fetched {len(teams_info)} teams") 431 432 # Log cache status 433 from_cache = ( 434 hasattr(response, "from_cache") 435 and isinstance(response.from_cache, bool) 436 and response.from_cache 437 ) 438 if from_cache: 439 logger.debug("Cache hit - skipped rate limiting") 440 else: 441 logger.debug("Real API request - rate limited") 442 443 return teams_info 444 445 except requests.exceptions.SSLError as e: 446 logger.error(f"SSL certificate verification failed for {url}: {e}") 447 raise NHLApiSSLError(f"SSL certificate verification failed for {url}: {e}") from e 448 except requests.exceptions.HTTPError as e: 449 logger.error(f"HTTP error while fetching teams: {e}") 450 raise NHLApiError(f"HTTP error: {e}") from e 451 except (KeyError, ValueError) as e: 452 logger.error(f"Error parsing teams response: {e}") 453 raise NHLApiError(f"Invalid API response format: {e}") from e 454 455 try: 456 # Wrap with circuit breaker for DoS prevention 457 return self.circuit_breaker.call(_fetch_teams) 458 except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: 459 # Convert to NHLApiConnectionError after retries exhausted 460 logger.error(f"Connection error after retries: {e}") 461 raise NHLApiConnectionError("Unable to connect to NHL API after retries") from e
Fetch all NHL teams with division and conference information.
This method uses the retry decorator to automatically retry on network errors. The URL is validated with SSRF protection before making the request.
Args: season: Optional season in format 'YYYYYYYY' (e.g., '20222023' for 2022-23). If None, fetches current season data.
Returns: Dictionary mapping team abbreviations to their metadata: { 'TOR': {'division': 'Atlantic', 'conference': 'Eastern'}, 'MTL': {'division': 'Atlantic', 'conference': 'Eastern'}, ... }
Raises: NHLApiConnectionError: If unable to connect to the API NHLApiError: For other API errors, including SSRF protection blocks
Examples:
client = NHLApiClient() teams = client.get_teams() "TOR" in teams True teams_2022 = client.get_teams(season="20222023") "TOR" in teams_2022 True
513 def get_team_roster( # noqa: PLR0915 514 self, team_abbrev: str, season: str | None = None 515 ) -> dict[str, Any]: 516 """Fetch the roster for a specific team with input and response validation. 517 518 Validates team abbreviation before making API call and validates response 519 structure to prevent errors from malformed data. 520 521 The URL is validated with SSRF protection before making the request. 522 523 Args: 524 team_abbrev: Team abbreviation (e.g., 'TOR', 'MTL') 525 season: Optional season in format 'YYYYYYYY' (e.g., '20222023' for 2022-23). 526 If None, fetches current season roster. 527 528 Returns: 529 Dictionary containing roster data with 'forwards', 'defensemen', and 'goalies' keys 530 531 Raises: 532 ValidationError: If team abbreviation is invalid 533 NHLApiNotFoundError: If the roster is not found (404 response) 534 NHLApiConnectionError: If unable to connect to the API after all retries 535 NHLApiError: For other API errors, including SSRF protection blocks and malformed responses 536 537 Security: 538 - Validates team abbreviation to prevent injection attacks 539 - Validates response structure to prevent KeyError exceptions 540 - Sanitizes player names from API responses 541 - SSRF protection on all API requests 542 543 Examples: 544 >>> client = NHLApiClient() 545 >>> roster = client.get_team_roster("TOR") 546 >>> "forwards" in roster 547 True 548 >>> roster_2022 = client.get_team_roster("TOR", season="20222023") 549 >>> "forwards" in roster_2022 550 True 551 >>> client.get_team_roster("INVALID") 552 Traceback (most recent call last): 553 ValidationError: Team abbreviation must be 2-3 characters... 554 """ 555 # Validate team abbreviation BEFORE making API call 556 try: 557 validated_abbrev = validate_team_abbreviation(team_abbrev) 558 except ValidationError: 559 # Re-raise validation errors for consistency with other API errors 560 logger.error(f"Invalid team abbreviation: {team_abbrev}") 561 raise 562 563 # Use season-specific endpoint or current season endpoint 564 endpoint = ( 565 f"roster/{validated_abbrev}/{season}" 566 if season 567 else f"roster/{validated_abbrev}/current" 568 ) 569 url = f"{self.base_url}/{endpoint}" 570 571 season_desc = f"season {season}" if season else "current season" 572 if logger.isEnabledFor(logging.DEBUG): 573 logger.debug(f"Fetching roster for {validated_abbrev} ({season_desc})") 574 575 # Validate URL with SSRF protection 576 self._validate_request_url(url) 577 578 def _fetch_roster() -> dict[str, Any]: # noqa: PLR0915 579 """Fetch roster with retry logic.""" 580 for attempt in range(self.retries): 581 try: 582 # Check if URL is cached 583 is_cached = self._is_url_cached(url) 584 585 # Only rate limit for actual API calls (not cached responses) 586 if not is_cached: 587 if logger.isEnabledFor(logging.DEBUG): 588 logger.debug(f"Rate limiting: acquiring token for {team_abbrev} roster") 589 self.rate_limiter.acquire() 590 591 response = self.session.get( 592 url, 593 timeout=self.timeout, 594 verify=self.ca_bundle, # Explicit SSL verification with certifi CA bundle 595 ) 596 597 if response.status_code == 404: 598 logger.warning(f"No roster data available for {team_abbrev}") 599 raise NHLApiNotFoundError(f"Roster not found for team: {team_abbrev}") 600 601 # Handle 429 rate limiting with exponential backoff 602 if response.status_code == 429: 603 if attempt < self.retries - 1: 604 retry_after = self._get_retry_after(response) 605 logger.warning( 606 f"Rate limited (429) for {team_abbrev} " 607 f"(attempt {attempt + 1}/{self.retries}), " 608 f"retrying in {retry_after:.2f}s..." 609 ) 610 time.sleep(retry_after) 611 continue 612 logger.error( 613 f"Rate limited (429) for {team_abbrev} after {self.retries} attempts" 614 ) 615 raise NHLApiConnectionError( 616 f"Rate limited after {self.retries} attempts" 617 ) from None 618 619 response.raise_for_status() 620 data = response.json() 621 622 # Validate response structure 623 try: 624 validate_api_response_structure( 625 data, 626 required_keys=["forwards", "defensemen", "goalies"], 627 context=f"Team roster response for {validated_abbrev}", 628 ) 629 except ValidationError as e: 630 logger.error( 631 f"Invalid roster response structure for {validated_abbrev}: {e}" 632 ) 633 raise NHLApiError(f"Invalid API response: {e}") from e 634 635 # Sanitize player names in response 636 self._sanitize_roster_player_names(data) 637 638 if logger.isEnabledFor(logging.DEBUG): 639 logger.debug( 640 f"Successfully fetched and validated roster for {validated_abbrev}" 641 ) 642 643 # Log cache status 644 from_cache = ( 645 hasattr(response, "from_cache") 646 and isinstance(response.from_cache, bool) 647 and response.from_cache 648 ) 649 if from_cache: 650 logger.debug("Cache hit - skipped rate limiting") 651 else: 652 logger.debug("Real API request - rate limited") 653 654 return data # type: ignore[no-any-return] 655 656 except requests.exceptions.Timeout: 657 if attempt < self.retries - 1: 658 backoff_delay = self._calculate_backoff_delay(attempt) 659 logger.warning( 660 f"Timeout fetching {team_abbrev} " 661 f"(attempt {attempt + 1}/{self.retries}), " 662 f"retrying in {backoff_delay:.2f}s..." 663 ) 664 time.sleep(backoff_delay) 665 else: 666 logger.error(f"Failed to fetch {team_abbrev} after {self.retries} attempts") 667 raise NHLApiConnectionError( 668 f"Request timed out after {self.retries} attempts" 669 ) from None 670 671 except requests.exceptions.SSLError as e: 672 # SSL errors should not be retried - certificate validation failure is permanent 673 logger.error(f"SSL certificate verification failed for {team_abbrev}: {e}") 674 raise NHLApiSSLError( 675 f"SSL certificate verification failed for {url}: {e}" 676 ) from e 677 678 except requests.exceptions.ConnectionError: 679 if attempt < self.retries - 1: 680 backoff_delay = self._calculate_backoff_delay(attempt) 681 logger.warning( 682 f"Connection error for {team_abbrev} " 683 f"(attempt {attempt + 1}/{self.retries}), " 684 f"retrying in {backoff_delay:.2f}s..." 685 ) 686 time.sleep(backoff_delay) 687 else: 688 logger.error(f"Failed to fetch {team_abbrev} after {self.retries} attempts") 689 raise NHLApiConnectionError( 690 f"Connection failed after {self.retries} attempts" 691 ) from None 692 693 except requests.exceptions.HTTPError as e: 694 logger.error(f"HTTP error fetching {team_abbrev}: {e}") 695 raise NHLApiError(f"HTTP error: {e}") from e 696 697 # This should never be reached as all paths above either return or raise 698 raise NHLApiError("Unexpected error: retry loop completed without returning data") 699 700 # Wrap with circuit breaker for DoS prevention 701 return self.circuit_breaker.call(_fetch_roster)
Fetch the roster for a specific team with input and response validation.
Validates team abbreviation before making API call and validates response structure to prevent errors from malformed data.
The URL is validated with SSRF protection before making the request.
Args: team_abbrev: Team abbreviation (e.g., 'TOR', 'MTL') season: Optional season in format 'YYYYYYYY' (e.g., '20222023' for 2022-23). If None, fetches current season roster.
Returns: Dictionary containing roster data with 'forwards', 'defensemen', and 'goalies' keys
Raises: ValidationError: If team abbreviation is invalid NHLApiNotFoundError: If the roster is not found (404 response) NHLApiConnectionError: If unable to connect to the API after all retries NHLApiError: For other API errors, including SSRF protection blocks and malformed responses
Security: - Validates team abbreviation to prevent injection attacks - Validates response structure to prevent KeyError exceptions - Sanitizes player names from API responses - SSRF protection on all API requests
Examples:
client = NHLApiClient() roster = client.get_team_roster("TOR") "forwards" in roster True roster_2022 = client.get_team_roster("TOR", season="20222023") "forwards" in roster_2022 True client.get_team_roster("INVALID") Traceback (most recent call last): ValidationError: Team abbreviation must be 2-3 characters...
703 def get_rate_limit_stats(self) -> dict[str, Any]: 704 """Get rate limiter statistics. 705 706 Returns: 707 Dictionary with rate limiter statistics including: 708 - total_requests: Total requests made 709 - total_waits: Total times waited for tokens 710 - total_wait_time: Total time spent waiting 711 - average_wait: Average wait time per wait 712 - current_tokens: Current token count 713 - max_tokens: Maximum token capacity 714 715 Examples: 716 >>> client = NHLApiClient() 717 >>> stats = client.get_rate_limit_stats() 718 >>> "total_requests" in stats 719 True 720 """ 721 return self.rate_limiter.get_stats()
Get rate limiter statistics.
Returns: Dictionary with rate limiter statistics including: - total_requests: Total requests made - total_waits: Total times waited for tokens - total_wait_time: Total time spent waiting - average_wait: Average wait time per wait - current_tokens: Current token count - max_tokens: Maximum token capacity
Examples:
client = NHLApiClient() stats = client.get_rate_limit_stats() "total_requests" in stats True
723 def clear_cache(self) -> None: 724 """Clear the HTTP cache.""" 725 if self.cache_enabled and hasattr(self.session, "cache"): 726 self.session.cache.clear() 727 logger.info("API cache cleared") 728 else: 729 logger.debug("Cache not available or caching disabled")
Clear the HTTP cache.
15class ScrabbleScorer: 16 """Calculate Scrabble scores for player names using configurable letter values. 17 18 This class provides methods to calculate scores based on letter point values. 19 By default, uses standard English Scrabble values, but supports custom 20 scoring systems via the letter_values parameter. 21 22 Default letter values (standard Scrabble): 23 - 1 point: A, E, I, O, U, L, N, S, T, R 24 - 2 points: D, G 25 - 3 points: B, C, M, P 26 - 4 points: F, H, V, W, Y 27 - 5 points: K 28 - 8 points: J, X 29 - 10 points: Q, Z 30 31 Custom scoring systems can be provided via the letter_values parameter, 32 enabling alternative scoring methods (e.g., Wordle scoring, uniform values). 33 """ 34 35 LETTER_VALUES: ClassVar[dict[str, int]] = { 36 "A": 1, 37 "E": 1, 38 "I": 1, 39 "O": 1, 40 "U": 1, 41 "L": 1, 42 "N": 1, 43 "S": 1, 44 "T": 1, 45 "R": 1, 46 "D": 2, 47 "G": 2, 48 "B": 3, 49 "C": 3, 50 "M": 3, 51 "P": 3, 52 "F": 4, 53 "H": 4, 54 "V": 4, 55 "W": 4, 56 "Y": 4, 57 "K": 5, 58 "J": 8, 59 "X": 8, 60 "Q": 10, 61 "Z": 10, 62 } 63 64 def __init__(self, letter_values: dict[str, int] | None = None) -> None: 65 """Initialize the scorer with custom or default letter values. 66 67 Args: 68 letter_values: Optional custom letter-to-points mapping. 69 If None, uses standard Scrabble values. 70 71 Examples: 72 >>> # Standard Scrabble scoring 73 >>> scorer = ScrabbleScorer() 74 >>> scorer.calculate_score("ALEX") 75 11 76 77 >>> # Custom scoring (all letters worth 1 point) 78 >>> uniform_values = {chr(i): 1 for i in range(65, 91)} 79 >>> scorer = ScrabbleScorer(letter_values=uniform_values) 80 >>> scorer.calculate_score("ALEX") 81 4 82 """ 83 self._letter_values = letter_values if letter_values is not None else self.LETTER_VALUES 84 logger.debug(f"ScrabbleScorer initialized with {len(self._letter_values)} letter values") 85 86 @staticmethod 87 @lru_cache(maxsize=2048) 88 def _calculate_with_values(name: str, values_tuple: tuple[tuple[str, int], ...]) -> int: 89 """Calculate score with provided letter values (cached). 90 91 This static method enables LRU caching while supporting custom letter values. 92 The letter values are passed as a hashable tuple for cache key uniqueness. 93 94 Args: 95 name: Name to score 96 values_tuple: Letter values as tuple of (letter, value) pairs 97 98 Returns: 99 Total score for the name 100 """ 101 values_dict = dict(values_tuple) 102 return sum(values_dict.get(char.upper(), 0) for char in name) 103 104 @staticmethod 105 def calculate_score(name: str) -> int: 106 """Calculate the Scrabble score for a given name using standard values. 107 108 This static method provides convenient scoring with default Scrabble letter values. 109 For custom scoring values, create a ScrabbleScorer instance and use 110 the calculate_score_custom() method. 111 112 This method uses LRU caching to avoid recomputing scores for duplicate 113 names, which significantly improves performance when processing ~700 NHL 114 players with many duplicate first/last names. 115 116 Cache size: 2048 entries (sufficient for all unique name components) 117 118 Args: 119 name: The name to score (can include spaces and special characters) 120 121 Returns: 122 The total Scrabble score (non-letter characters are worth 0 points) 123 124 Examples: 125 >>> ScrabbleScorer.calculate_score("ALEX") 126 11 127 >>> ScrabbleScorer.calculate_score("Ovechkin") 128 20 129 """ 130 # Use default Scrabble values 131 values_tuple = tuple(sorted(ScrabbleScorer.LETTER_VALUES.items())) 132 return ScrabbleScorer._calculate_with_values(name, values_tuple) 133 134 def calculate_score_custom(self, name: str) -> int: 135 """Calculate score using custom letter values configured in this instance. 136 137 Use this method when you've created a ScrabbleScorer with custom letter 138 values. For default Scrabble scoring, use the static calculate_score() method. 139 140 Args: 141 name: The name to score (can include spaces and special characters) 142 143 Returns: 144 The total score using custom letter values 145 146 Examples: 147 >>> uniform_values = {chr(i): 1 for i in range(65, 91)} 148 >>> scorer = ScrabbleScorer(letter_values=uniform_values) 149 >>> scorer.calculate_score_custom("ALEX") 150 4 151 """ 152 # Convert dict to hashable tuple for caching 153 values_tuple = tuple(sorted(self._letter_values.items())) 154 return self._calculate_with_values(name, values_tuple) 155 156 def score_player( 157 self, player_data: dict[str, Any], team: str, division: str, conference: str 158 ) -> PlayerScore: 159 """Score a player and return a PlayerScore object. 160 161 Uses custom letter values if configured, otherwise uses default Scrabble values. 162 163 Args: 164 player_data: Dictionary with 'firstName' and 'lastName' keys containing 'default' values 165 team: Team abbreviation 166 division: Division name 167 conference: Conference name 168 169 Returns: 170 PlayerScore object with all scoring information 171 172 Examples: 173 >>> scorer = ScrabbleScorer() 174 >>> player = {"firstName": {"default": "Connor"}, "lastName": {"default": "McDavid"}} 175 >>> result = scorer.score_player(player, "EDM", "Pacific", "Western") 176 >>> result.full_score 177 24 178 """ 179 first_name = player_data["firstName"]["default"] 180 last_name = player_data["lastName"]["default"] 181 full_name = f"{first_name} {last_name}" 182 183 # Use custom scoring if custom values are set 184 if self._letter_values is not self.LETTER_VALUES: 185 first_score = self.calculate_score_custom(first_name) 186 last_score = self.calculate_score_custom(last_name) 187 else: 188 first_score = self.calculate_score(first_name) 189 last_score = self.calculate_score(last_name) 190 191 full_score = first_score + last_score 192 193 return PlayerScore( 194 first_name=first_name, 195 last_name=last_name, 196 full_name=full_name, 197 first_score=first_score, 198 last_score=last_score, 199 full_score=full_score, 200 team=team, 201 division=division, 202 conference=conference, 203 ) 204 205 @staticmethod 206 def get_cache_info() -> dict[str, int]: 207 """Get cache statistics for the score calculation cache. 208 209 Returns: 210 Dictionary with cache statistics: 211 - hits: Number of cache hits 212 - misses: Number of cache misses 213 - maxsize: Maximum cache size 214 - currsize: Current cache size 215 216 Examples: 217 >>> info = ScrabbleScorer.get_cache_info() 218 >>> info['maxsize'] 219 2048 220 """ 221 cache_info = ScrabbleScorer._calculate_with_values.cache_info() 222 return { 223 "hits": cache_info.hits, 224 "misses": cache_info.misses, 225 "maxsize": cache_info.maxsize or 0, 226 "currsize": cache_info.currsize, 227 } 228 229 @staticmethod 230 def log_cache_stats() -> None: 231 """Log cache statistics for monitoring and performance analysis. 232 233 Logs hit rate, total calls, and cache utilization at INFO level. 234 """ 235 stats = ScrabbleScorer.get_cache_info() 236 total_calls = stats["hits"] + stats["misses"] 237 238 if total_calls > 0: 239 hit_rate = (stats["hits"] / total_calls) * 100 240 utilization = ( 241 (stats["currsize"] / stats["maxsize"]) * 100 if stats["maxsize"] > 0 else 0 242 ) 243 244 logger.info( 245 "Scrabble scoring cache stats: " 246 f"hits={stats['hits']}, " 247 f"misses={stats['misses']}, " 248 f"hit_rate={hit_rate:.1f}%, " 249 f"size={stats['currsize']}/{stats['maxsize']} " 250 f"({utilization:.1f}% full)" 251 ) 252 else: 253 logger.info("Scrabble scoring cache: No calls yet") 254 255 @staticmethod 256 def clear_cache() -> None: 257 """Clear the score calculation cache. 258 259 Useful for testing or when memory needs to be freed. 260 """ 261 ScrabbleScorer._calculate_with_values.cache_clear() 262 logger.debug("Scrabble scoring cache cleared")
Calculate Scrabble scores for player names using configurable letter values.
This class provides methods to calculate scores based on letter point values. By default, uses standard English Scrabble values, but supports custom scoring systems via the letter_values parameter.
Default letter values (standard Scrabble): - 1 point: A, E, I, O, U, L, N, S, T, R - 2 points: D, G - 3 points: B, C, M, P - 4 points: F, H, V, W, Y - 5 points: K - 8 points: J, X - 10 points: Q, Z
Custom scoring systems can be provided via the letter_values parameter, enabling alternative scoring methods (e.g., Wordle scoring, uniform values).
64 def __init__(self, letter_values: dict[str, int] | None = None) -> None: 65 """Initialize the scorer with custom or default letter values. 66 67 Args: 68 letter_values: Optional custom letter-to-points mapping. 69 If None, uses standard Scrabble values. 70 71 Examples: 72 >>> # Standard Scrabble scoring 73 >>> scorer = ScrabbleScorer() 74 >>> scorer.calculate_score("ALEX") 75 11 76 77 >>> # Custom scoring (all letters worth 1 point) 78 >>> uniform_values = {chr(i): 1 for i in range(65, 91)} 79 >>> scorer = ScrabbleScorer(letter_values=uniform_values) 80 >>> scorer.calculate_score("ALEX") 81 4 82 """ 83 self._letter_values = letter_values if letter_values is not None else self.LETTER_VALUES 84 logger.debug(f"ScrabbleScorer initialized with {len(self._letter_values)} letter values")
Initialize the scorer with custom or default letter values.
Args: letter_values: Optional custom letter-to-points mapping. If None, uses standard Scrabble values.
Examples:
Standard Scrabble scoring
scorer = ScrabbleScorer() scorer.calculate_score("ALEX") 11
>>> # Custom scoring (all letters worth 1 point) >>> uniform_values = {chr(i): 1 for i in range(65, 91)} >>> scorer = ScrabbleScorer(letter_values=uniform_values) >>> scorer.calculate_score("ALEX") 4
104 @staticmethod 105 def calculate_score(name: str) -> int: 106 """Calculate the Scrabble score for a given name using standard values. 107 108 This static method provides convenient scoring with default Scrabble letter values. 109 For custom scoring values, create a ScrabbleScorer instance and use 110 the calculate_score_custom() method. 111 112 This method uses LRU caching to avoid recomputing scores for duplicate 113 names, which significantly improves performance when processing ~700 NHL 114 players with many duplicate first/last names. 115 116 Cache size: 2048 entries (sufficient for all unique name components) 117 118 Args: 119 name: The name to score (can include spaces and special characters) 120 121 Returns: 122 The total Scrabble score (non-letter characters are worth 0 points) 123 124 Examples: 125 >>> ScrabbleScorer.calculate_score("ALEX") 126 11 127 >>> ScrabbleScorer.calculate_score("Ovechkin") 128 20 129 """ 130 # Use default Scrabble values 131 values_tuple = tuple(sorted(ScrabbleScorer.LETTER_VALUES.items())) 132 return ScrabbleScorer._calculate_with_values(name, values_tuple)
Calculate the Scrabble score for a given name using standard values.
This static method provides convenient scoring with default Scrabble letter values. For custom scoring values, create a ScrabbleScorer instance and use the calculate_score_custom() method.
This method uses LRU caching to avoid recomputing scores for duplicate names, which significantly improves performance when processing ~700 NHL players with many duplicate first/last names.
Cache size: 2048 entries (sufficient for all unique name components)
Args: name: The name to score (can include spaces and special characters)
Returns: The total Scrabble score (non-letter characters are worth 0 points)
Examples:
ScrabbleScorer.calculate_score("ALEX") 11 ScrabbleScorer.calculate_score("Ovechkin") 20
134 def calculate_score_custom(self, name: str) -> int: 135 """Calculate score using custom letter values configured in this instance. 136 137 Use this method when you've created a ScrabbleScorer with custom letter 138 values. For default Scrabble scoring, use the static calculate_score() method. 139 140 Args: 141 name: The name to score (can include spaces and special characters) 142 143 Returns: 144 The total score using custom letter values 145 146 Examples: 147 >>> uniform_values = {chr(i): 1 for i in range(65, 91)} 148 >>> scorer = ScrabbleScorer(letter_values=uniform_values) 149 >>> scorer.calculate_score_custom("ALEX") 150 4 151 """ 152 # Convert dict to hashable tuple for caching 153 values_tuple = tuple(sorted(self._letter_values.items())) 154 return self._calculate_with_values(name, values_tuple)
Calculate score using custom letter values configured in this instance.
Use this method when you've created a ScrabbleScorer with custom letter values. For default Scrabble scoring, use the static calculate_score() method.
Args: name: The name to score (can include spaces and special characters)
Returns: The total score using custom letter values
Examples:
uniform_values = {chr(i): 1 for i in range(65, 91)} scorer = ScrabbleScorer(letter_values=uniform_values) scorer.calculate_score_custom("ALEX") 4
156 def score_player( 157 self, player_data: dict[str, Any], team: str, division: str, conference: str 158 ) -> PlayerScore: 159 """Score a player and return a PlayerScore object. 160 161 Uses custom letter values if configured, otherwise uses default Scrabble values. 162 163 Args: 164 player_data: Dictionary with 'firstName' and 'lastName' keys containing 'default' values 165 team: Team abbreviation 166 division: Division name 167 conference: Conference name 168 169 Returns: 170 PlayerScore object with all scoring information 171 172 Examples: 173 >>> scorer = ScrabbleScorer() 174 >>> player = {"firstName": {"default": "Connor"}, "lastName": {"default": "McDavid"}} 175 >>> result = scorer.score_player(player, "EDM", "Pacific", "Western") 176 >>> result.full_score 177 24 178 """ 179 first_name = player_data["firstName"]["default"] 180 last_name = player_data["lastName"]["default"] 181 full_name = f"{first_name} {last_name}" 182 183 # Use custom scoring if custom values are set 184 if self._letter_values is not self.LETTER_VALUES: 185 first_score = self.calculate_score_custom(first_name) 186 last_score = self.calculate_score_custom(last_name) 187 else: 188 first_score = self.calculate_score(first_name) 189 last_score = self.calculate_score(last_name) 190 191 full_score = first_score + last_score 192 193 return PlayerScore( 194 first_name=first_name, 195 last_name=last_name, 196 full_name=full_name, 197 first_score=first_score, 198 last_score=last_score, 199 full_score=full_score, 200 team=team, 201 division=division, 202 conference=conference, 203 )
Score a player and return a PlayerScore object.
Uses custom letter values if configured, otherwise uses default Scrabble values.
Args: player_data: Dictionary with 'firstName' and 'lastName' keys containing 'default' values team: Team abbreviation division: Division name conference: Conference name
Returns: PlayerScore object with all scoring information
Examples:
scorer = ScrabbleScorer() player = {"firstName": {"default": "Connor"}, "lastName": {"default": "McDavid"}} result = scorer.score_player(player, "EDM", "Pacific", "Western") result.full_score 24
205 @staticmethod 206 def get_cache_info() -> dict[str, int]: 207 """Get cache statistics for the score calculation cache. 208 209 Returns: 210 Dictionary with cache statistics: 211 - hits: Number of cache hits 212 - misses: Number of cache misses 213 - maxsize: Maximum cache size 214 - currsize: Current cache size 215 216 Examples: 217 >>> info = ScrabbleScorer.get_cache_info() 218 >>> info['maxsize'] 219 2048 220 """ 221 cache_info = ScrabbleScorer._calculate_with_values.cache_info() 222 return { 223 "hits": cache_info.hits, 224 "misses": cache_info.misses, 225 "maxsize": cache_info.maxsize or 0, 226 "currsize": cache_info.currsize, 227 }
Get cache statistics for the score calculation cache.
Returns: Dictionary with cache statistics: - hits: Number of cache hits - misses: Number of cache misses - maxsize: Maximum cache size - currsize: Current cache size
Examples:
info = ScrabbleScorer.get_cache_info() info['maxsize'] 2048
229 @staticmethod 230 def log_cache_stats() -> None: 231 """Log cache statistics for monitoring and performance analysis. 232 233 Logs hit rate, total calls, and cache utilization at INFO level. 234 """ 235 stats = ScrabbleScorer.get_cache_info() 236 total_calls = stats["hits"] + stats["misses"] 237 238 if total_calls > 0: 239 hit_rate = (stats["hits"] / total_calls) * 100 240 utilization = ( 241 (stats["currsize"] / stats["maxsize"]) * 100 if stats["maxsize"] > 0 else 0 242 ) 243 244 logger.info( 245 "Scrabble scoring cache stats: " 246 f"hits={stats['hits']}, " 247 f"misses={stats['misses']}, " 248 f"hit_rate={hit_rate:.1f}%, " 249 f"size={stats['currsize']}/{stats['maxsize']} " 250 f"({utilization:.1f}% full)" 251 ) 252 else: 253 logger.info("Scrabble scoring cache: No calls yet")
Log cache statistics for monitoring and performance analysis.
Logs hit rate, total calls, and cache utilization at INFO level.
255 @staticmethod 256 def clear_cache() -> None: 257 """Clear the score calculation cache. 258 259 Useful for testing or when memory needs to be freed. 260 """ 261 ScrabbleScorer._calculate_with_values.cache_clear() 262 logger.debug("Scrabble scoring cache cleared")
Clear the score calculation cache.
Useful for testing or when memory needs to be freed.
22class ValidationError(ValueError): 23 """Raised when input validation fails. 24 25 This exception provides clear, actionable error messages that 26 indicate what was invalid and what the expected format is. 27 28 Examples: 29 >>> raise ValidationError("top_players must be between 1 and 100, got 999") 30 """
Raised when input validation fails.
This exception provides clear, actionable error messages that indicate what was invalid and what the expected format is.
Examples:
raise ValidationError("top_players must be between 1 and 100, got 999")