refactor: fix code review issues across routes and services
- 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
This commit is contained in:
51
obb_utils.py
Normal file
51
obb_utils.py
Normal file
@@ -0,0 +1,51 @@
|
||||
"""Shared OpenBB result conversion utilities."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
|
||||
def to_list(result: Any) -> list[dict[str, Any]]:
|
||||
"""Convert OBBject result to list of dicts with serialized dates."""
|
||||
if result is None or result.results is None:
|
||||
return []
|
||||
items = result.results
|
||||
if not isinstance(items, list):
|
||||
items = [items]
|
||||
out = []
|
||||
for item in items:
|
||||
if hasattr(item, "model_dump"):
|
||||
d = item.model_dump()
|
||||
else:
|
||||
raw = vars(item)
|
||||
d = dict(raw) if raw else {}
|
||||
d = {
|
||||
k: v.isoformat() if hasattr(v, "isoformat") else v
|
||||
for k, v in d.items()
|
||||
}
|
||||
out.append(d)
|
||||
return out
|
||||
|
||||
|
||||
def extract_single(result: Any) -> dict[str, Any]:
|
||||
"""Extract data from an OBBject result (single model or list)."""
|
||||
if result is None:
|
||||
return {}
|
||||
items = getattr(result, "results", None)
|
||||
if items is None:
|
||||
return {}
|
||||
if hasattr(items, "model_dump"):
|
||||
return items.model_dump()
|
||||
if isinstance(items, list) and items:
|
||||
last = items[-1]
|
||||
return last.model_dump() if hasattr(last, "model_dump") else {}
|
||||
return {}
|
||||
|
||||
|
||||
def safe_last(result: Any) -> dict[str, Any] | None:
|
||||
"""Get the last item from a list result, or None."""
|
||||
if result is None:
|
||||
return None
|
||||
items = getattr(result, "results", None)
|
||||
if items is None or not isinstance(items, list) or not items:
|
||||
return None
|
||||
last = items[-1]
|
||||
return last.model_dump() if hasattr(last, "model_dump") else None
|
||||
Reference in New Issue
Block a user