home-automation/app/integrations/tibber/client.py

"""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 the forward-looking today + tomorrow price curve at
# 15-minute resolution.
#
# ``priceInfo(resolution: QUARTER_HOURLY)`` returns two node lists:
#   * ``today``    — always the full current local day (96 quarter-hourly slots,
#                    00:00 → 23:45 local), regardless of the current time.
#   * ``tomorrow`` — the full next local day (96 slots) once Tibber publishes the
#                    day-ahead prices (around 13:00–15:00 local); empty before that.
#
# This is deliberately NOT ``priceInfoRange``: that field is a historical cursor
# connection whose range ends at "now" (it never returns future slots), so it
# cannot supply upcoming prices.  ``priceInfo`` is forward-looking, so every
# 15-minute slot's price is present in the DB *before* the slot closes — which is
# what makes per-slot billing accurate (each period finds its own exact slot
# instead of falling back to a stale earlier price) and keeps the live current-
# price entity fresh.  The hourly refresh job re-runs this query, so tomorrow's
# prices are picked up within an hour of publication without a restart.
_PRICE_RANGE_QUERY = """
{
  viewer {
    homes {
      id
      currentSubscription {
        priceInfo(resolution: QUARTER_HOURLY) {
          today {
            startsAt
            total
            energy
            tax
            currency
            level
          }
          tomorrow {
            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 the forward-looking today + tomorrow 15-minute price curve from Tibber.

    Sends the ``priceInfo(resolution: QUARTER_HOURLY) { today tomorrow }`` query
    and parses every node from both lists (today first, then tomorrow) into a
    ``PricePoint``.  ``priceInfo`` is forward-looking — ``today`` is always the
    full current local day and ``tomorrow`` is populated once Tibber publishes the
    day-ahead prices — so upcoming slots are returned, unlike ``priceInfoRange``
    which only reaches "now".  ``tomorrow`` may be empty (before publication); the
    number of nodes is not assumed and all returned nodes are parsed.

    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:
        price_info = home["currentSubscription"]["priceInfo"]
        today = price_info["today"]
        tomorrow = price_info["tomorrow"]
    except (KeyError, TypeError) as exc:
        raise TibberError("Tibber API response missing priceInfo today/tomorrow") from exc

    # tomorrow is null/empty until Tibber publishes the day-ahead prices; treat
    # a missing list as empty so we still return today's slots.
    nodes = list(today or []) + list(tomorrow or [])
    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")