"""Tibber GraphQL API client for fetching electricity price data. Design decisions ---------------- - Uses ``httpx`` (already a project dependency) for all HTTP requests. - Token is **never** written to log messages or exception strings to prevent credential leakage into log aggregators. - ``fetch_price_range`` returns a list of ``PricePoint`` objects with UTC-aware ``starts_at`` datetimes; the number of nodes is not assumed — all returned nodes are parsed regardless of count. - ``fetch_current_price`` uses a separate, simpler query that asks for the *current* price point only; used by the connection-test endpoint (T09) to produce a fast three-state result (success / auth-error / network-error). - Authentication failures (HTTP 401/403) are raised as ``TibberAuthError`` so that callers (the test endpoint) can distinguish them from generic network or API errors. """ from __future__ import annotations import logging from dataclasses import dataclass from datetime import UTC, datetime from typing import Any import httpx logger = logging.getLogger(__name__) _TIBBER_API_URL = "https://api.tibber.com/v1-beta/gql" _DEFAULT_TIMEOUT = 15.0 # GraphQL query to fetch a range of 15-minute price nodes. # ``priceInfoRange`` is a Relay-style cursor connection over the subscription's # entire available price history. With no before/after cursor: # * ``first: N`` returns the OLDEST N nodes (anchored at the subscription # start date) — a fixed window that never advances, so it is WRONG for # "current + upcoming" prices. # * ``last: N`` returns the NEWEST N nodes (ending at the latest published # slot: tomorrow 23:45 once day-ahead prices are out, else today 23:45). # We want the newest slots, so we use ``last``. 192 = 2 days × 24 h × 4 slots, # which fully covers the "today + tomorrow" window that GET /api/energy/prices # queries (the earliest of the 192 newest slots reaches back past today 00:00 # UTC in either publish state). _PRICE_RANGE_QUERY = """ { viewer { homes { id currentSubscription { priceInfoRange(resolution: QUARTER_HOURLY, last: 192) { nodes { startsAt total energy tax currency level } } } } } } """ # GraphQL query to fetch the single *current* price point. _CURRENT_PRICE_QUERY = """ { viewer { homes { id currentSubscription { priceInfo { current { startsAt total energy tax currency level } } } } } } """ # --------------------------------------------------------------------------- # Custom exceptions # --------------------------------------------------------------------------- class TibberError(Exception): """Base class for all Tibber client errors. Raised for network failures, timeouts, unexpected HTTP status codes, and malformed API responses. The exception message will **never** contain the API token. """ class TibberAuthError(TibberError): """Raised when the Tibber API rejects the provided token (HTTP 401/403). Callers that implement a three-state connection test should catch this exception separately from ``TibberError`` to distinguish auth problems from network / API problems. """ # --------------------------------------------------------------------------- # Data model # --------------------------------------------------------------------------- @dataclass(slots=True) class PricePoint: """One 15-minute price slot returned by the Tibber API. All monetary values are in ``currency`` and include VAT (user-facing). ``starts_at`` is always a timezone-aware UTC datetime regardless of the timezone offset that Tibber returns in ``startsAt``. """ starts_at: datetime # UTC-aware total: float # full all-in price (energy + tax) energy: float # spot energy component tax: float # tax component currency: str # ISO 4217 (typically "EUR") level: str | None # e.g. "CHEAP", "NORMAL", "EXPENSIVE"; None if absent resolution: str # label for the slot resolution (e.g. "QUARTER_HOURLY") # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _parse_starts_at(raw: str) -> datetime: """Parse an ISO 8601 timestamp with timezone offset and convert to UTC. Tibber returns timestamps like ``2026-06-23T00:00:00.000+02:00``. ``datetime.fromisoformat`` handles this format in Python 3.11+; the result is then converted to UTC via ``.astimezone(UTC)``. """ return datetime.fromisoformat(raw).astimezone(UTC) def _parse_node(node: dict[str, Any], resolution: str) -> PricePoint: """Parse a single price node dict into a ``PricePoint``.""" return PricePoint( starts_at=_parse_starts_at(node["startsAt"]), total=float(node["total"]), energy=float(node["energy"]), tax=float(node["tax"]), currency=str(node["currency"]), level=node.get("level") or None, resolution=resolution, ) def _pick_home(homes: list[dict[str, Any]], home_id: str | None) -> dict[str, Any]: """Return the home dict matching *home_id*, or the first home if None.""" if not homes: raise TibberError("Tibber API returned no homes") if home_id is not None: for h in homes: if h.get("id") == home_id: return h raise TibberError("Tibber home id not found in API response") return homes[0] def _post_graphql(token: str, query: str, timeout: float) -> dict[str, Any]: """POST the GraphQL *query* and return the parsed ``data`` dict. Raises ------ TibberAuthError On HTTP 401 or 403. TibberError On timeouts, network errors, non-2xx responses, or unexpected body shape. """ headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } try: response = httpx.post( _TIBBER_API_URL, json={"query": query}, headers=headers, timeout=timeout, ) except httpx.TimeoutException as exc: raise TibberError("Tibber API request timed out") from exc except httpx.HTTPError as exc: raise TibberError("Tibber API request failed") from exc if response.status_code in (401, 403): raise TibberAuthError("Tibber API authentication failed") try: response.raise_for_status() except httpx.HTTPStatusError as exc: raise TibberError(f"Tibber API returned unexpected status {response.status_code}") from exc try: body = response.json() except Exception as exc: raise TibberError("Tibber API returned non-JSON response") from exc if "errors" in body: # GraphQL errors are not HTTP errors; surface them as TibberError. # Do not include token in the message. raise TibberError("Tibber GraphQL returned errors") data = body.get("data") if data is None: raise TibberError("Tibber API response missing 'data' field") return data # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def fetch_price_range( token: str, home_id: str | None = None, *, timeout: float = _DEFAULT_TIMEOUT, ) -> list[PricePoint]: """Fetch a range of 15-minute price nodes from the Tibber API. Sends the ``priceInfoRange(resolution: QUARTER_HOURLY, last: 192)`` query and parses every returned node into a ``PricePoint``. ``last`` (not ``first``) is used so the newest slots are returned; ``first`` would anchor at the subscription start date and never advance. The number of nodes is not assumed — all returned nodes are parsed regardless of count. Parameters ---------- token: Tibber API token. **Never** logged or included in exception messages. home_id: If given, the home with this Tibber home ID is selected. If ``None``, the first home in the account is used. timeout: HTTP request timeout in seconds (default 15 s). Returns ------- list[PricePoint] List of price points with UTC-aware ``starts_at`` values. Raises ------ TibberAuthError If the token is rejected (HTTP 401/403). TibberError For network failures, timeouts, or unexpected API responses. """ data = _post_graphql(token, _PRICE_RANGE_QUERY, timeout) try: homes = data["viewer"]["homes"] except (KeyError, TypeError) as exc: raise TibberError("Tibber API response has unexpected shape") from exc home = _pick_home(homes, home_id) try: nodes = home["currentSubscription"]["priceInfoRange"]["nodes"] except (KeyError, TypeError) as exc: raise TibberError("Tibber API response missing priceInfoRange nodes") from exc return [_parse_node(node, "QUARTER_HOURLY") for node in nodes] def fetch_current_price( token: str, home_id: str | None = None, *, timeout: float = _DEFAULT_TIMEOUT, ) -> PricePoint: """Fetch the *current* price point from the Tibber API. Uses the lighter ``priceInfo { current { ... } }`` query rather than the full range query, making it suitable for a fast connection test. Parameters ---------- token: Tibber API token. **Never** logged or included in exception messages. home_id: If given, the home with this Tibber home ID is selected. If ``None``, the first home in the account is used. timeout: HTTP request timeout in seconds (default 15 s). Returns ------- PricePoint The current price point with a UTC-aware ``starts_at``. Raises ------ TibberAuthError If the token is rejected (HTTP 401/403). TibberError For network failures, timeouts, missing current price, or unexpected API responses. """ data = _post_graphql(token, _CURRENT_PRICE_QUERY, timeout) try: homes = data["viewer"]["homes"] except (KeyError, TypeError) as exc: raise TibberError("Tibber API response has unexpected shape") from exc home = _pick_home(homes, home_id) try: current = home["currentSubscription"]["priceInfo"]["current"] except (KeyError, TypeError) as exc: raise TibberError("Tibber API response missing current price") from exc if current is None: raise TibberError("Tibber API returned null for current price") return _parse_node(current, "QUARTER_HOURLY")