003c1d6ffc
- Extract shared route_utils.py (validate_symbol, safe decorator)
removing duplication from 6 route files
- Extract shared obb_utils.py (to_list, extract_single, safe_last)
removing duplication from calendar_service and market_service
- Fix _to_list dict mutation during iteration (use comprehension)
- Fix double vars() call and live __dict__ mutation risk
- Fix route ordering: /etf/search and /crypto/search now registered
before /{symbol} path params to prevent shadowing
- Add date format validation (YYYY-MM-DD pattern) on calendar routes
- Use timezone-aware datetime.now(tz=timezone.utc) in all services
- Add explicit type annotation for asyncio.gather results
40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
"""Shared route utilities: symbol validation and error handling decorator."""
|
|
|
|
import functools
|
|
import logging
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import ParamSpec, TypeVar
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from models import SYMBOL_PATTERN
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
P = ParamSpec("P")
|
|
R = TypeVar("R")
|
|
|
|
|
|
def validate_symbol(symbol: str) -> str:
|
|
"""Validate and normalize a stock symbol."""
|
|
if not SYMBOL_PATTERN.match(symbol):
|
|
raise HTTPException(status_code=400, detail="Invalid symbol format")
|
|
return symbol.upper()
|
|
|
|
|
|
def safe(fn: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
|
|
"""Decorator to catch upstream errors and return 502."""
|
|
@functools.wraps(fn)
|
|
async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
|
|
try:
|
|
return await fn(*args, **kwargs)
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
logger.exception("Upstream data error")
|
|
raise HTTPException(
|
|
status_code=502,
|
|
detail="Data provider error. Check server logs.",
|
|
)
|
|
return wrapper # type: ignore[return-value]
|