49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
"""Thin shim around `job_hunter.fx` (Frankfurter-backed) so callers
|
|
inside fire-planner have a single import. Re-exports the public API.
|
|
|
|
The job-hunter package isn't a hard dependency — when it isn't on
|
|
the Python path (e.g. running `fire-planner` outside the monorepo),
|
|
fall back to a tiny inline implementation that hits Frankfurter
|
|
directly with no DB caching.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
FRANKFURTER_URL = "https://api.frankfurter.dev/v1/{date}"
|
|
|
|
|
|
async def fetch_rates(as_of: date, client: httpx.AsyncClient | None = None) -> dict[str, Decimal]:
|
|
"""Return GBP-base rates for `as_of` — `{currency: rate_to_gbp}`.
|
|
|
|
rate_to_gbp[X] = "how much GBP one unit of X is worth", so
|
|
`gbp_amount = foreign_amount * rate_to_gbp[foreign]`.
|
|
"""
|
|
owns = client is None
|
|
if client is None:
|
|
client = httpx.AsyncClient(timeout=httpx.Timeout(20.0))
|
|
try:
|
|
resp = await client.get(
|
|
FRANKFURTER_URL.format(date=as_of.isoformat()),
|
|
params={"base": "GBP"},
|
|
follow_redirects=True,
|
|
)
|
|
resp.raise_for_status()
|
|
payload: dict[str, Any] = resp.json()
|
|
finally:
|
|
if owns:
|
|
await client.aclose()
|
|
rates = payload.get("rates") or {}
|
|
out: dict[str, Decimal] = {"GBP": Decimal("1")}
|
|
for currency, rate in rates.items():
|
|
if not rate:
|
|
continue
|
|
try:
|
|
out[currency] = Decimal("1") / Decimal(str(rate))
|
|
except (ArithmeticError, ValueError):
|
|
continue
|
|
return out
|