M6-T05: add Tibber GraphQL client + price refresh service + schedule
- app/integrations/tibber/client.py: fetch_price_range/fetch_current_price (priceInfoRange QUARTER_HOURLY, startsAt->UTC, no fixed node count assumption), TibberError/TibberAuthError; token never logged or put in exception text. - app/services/tibber_prices.py: refresh_prices upserts by starts_at (idempotent), no-op unless an active kind=tibber contract + token exist. - main.py: hourly tibber-refresh job (existing jobs/MQTT untouched). Tests added.
This commit is contained in:
@@ -0,0 +1,10 @@
|
||||
"""Tibber dynamic electricity pricing integration.
|
||||
|
||||
This package provides:
|
||||
- ``client``: HTTP client for the Tibber GraphQL API (price fetching).
|
||||
- Custom exceptions: ``TibberError`` (base) and ``TibberAuthError`` (auth failure).
|
||||
|
||||
Usage::
|
||||
|
||||
from app.integrations.tibber.client import fetch_price_range, TibberAuthError
|
||||
"""
|
||||
@@ -0,0 +1,329 @@
|
||||
"""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(resolution: QUARTER_HOURLY, first: 96)`` fetches up to
|
||||
# 96 quarter-hourly slots which covers today + tomorrow (2 × 24 × 4 = 192 max,
|
||||
# but the Tibber API typically starts from the current slot and returns at
|
||||
# most the remaining hours of today plus tomorrow, so 96 is a good cap for
|
||||
# "today + tomorrow").
|
||||
_PRICE_RANGE_QUERY = """
|
||||
{
|
||||
viewer {
|
||||
homes {
|
||||
id
|
||||
currentSubscription {
|
||||
priceInfoRange(resolution: QUARTER_HOURLY, first: 96) {
|
||||
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, first: 96)`` query
|
||||
and parses every returned node into a ``PricePoint``. 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")
|
||||
+39
@@ -31,6 +31,7 @@ from app.services.config_page import build_runtime_settings, seed_missing_config
|
||||
from app.services.public_ip import check_public_ipv4_and_notify
|
||||
from app.services.modbus_poll import poll_all_enabled_devices, BASE_POLL_TICK_SECONDS
|
||||
from app.services.ha_discovery import publish_discovery, publish_states
|
||||
from app.services.tibber_prices import refresh_prices
|
||||
from scripts.app_db_adopt import AppDatabaseAdoptionError, validate_app_runtime_db
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -63,6 +64,33 @@ def _run_scheduled_modbus_poll() -> None:
|
||||
session.close()
|
||||
|
||||
|
||||
def _run_scheduled_tibber_refresh() -> None:
|
||||
"""Scheduled job: fetch Tibber 15-minute prices and upsert into tibber_price.
|
||||
|
||||
Runs every hour so that:
|
||||
- Today's prices are available from startup.
|
||||
- Tomorrow's prices (published by Tibber around 13:00 CET / 11:00 UTC) are
|
||||
picked up within an hour of publication without requiring a server restart.
|
||||
|
||||
The job is a no-op when:
|
||||
- No active energy contract with kind="tibber" exists.
|
||||
- The Tibber API token is empty in the runtime settings.
|
||||
|
||||
Any client exceptions (auth failures, network errors) are caught and logged
|
||||
so that a single failed fetch does not crash the scheduler or affect the
|
||||
other background jobs.
|
||||
"""
|
||||
session_local = get_session_local()
|
||||
session: Session = session_local()
|
||||
try:
|
||||
runtime_settings = build_runtime_settings(session, get_settings())
|
||||
refresh_prices(session, runtime_settings)
|
||||
except Exception:
|
||||
logger.exception("_run_scheduled_tibber_refresh: unexpected error")
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
def _run_scheduled_ha_state_publish() -> None:
|
||||
"""Periodic job: publish discovery configs + state + availability for all enabled exposed entities.
|
||||
|
||||
@@ -148,6 +176,17 @@ async def lifespan(_: FastAPI):
|
||||
max_instances=1,
|
||||
coalesce=True,
|
||||
)
|
||||
# Tibber price refresh: fetch today + tomorrow every hour.
|
||||
# The job is a no-op when no active tibber contract or token is configured,
|
||||
# so it is safe to register unconditionally.
|
||||
scheduler.add_job(
|
||||
_run_scheduled_tibber_refresh,
|
||||
trigger=IntervalTrigger(hours=1),
|
||||
id="tibber-refresh",
|
||||
replace_existing=True,
|
||||
max_instances=1,
|
||||
coalesce=True,
|
||||
)
|
||||
scheduler.start()
|
||||
|
||||
# MQTT: connect using DB-merged runtime settings so broker configured via UI
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
"""Service layer for fetching and persisting Tibber 15-minute electricity prices.
|
||||
|
||||
``refresh_prices`` is the main entry point. It is designed to be called from a
|
||||
scheduled background job (see ``app/main.py``) and from tests.
|
||||
|
||||
Design decisions
|
||||
----------------
|
||||
- **Guard clause**: if no active contract with ``kind="tibber"`` exists, or if
|
||||
the Tibber API token is empty in the runtime settings, the function is a
|
||||
complete no-op (returns 0) and does not raise. This means the scheduler can
|
||||
call ``refresh_prices`` unconditionally; the service itself decides whether to
|
||||
do anything based on current configuration.
|
||||
- **Upsert idempotency**: rows are matched by ``starts_at`` (the unique
|
||||
constraint on ``tibber_price``). If a row already exists, its price fields
|
||||
and ``fetched_at`` are updated in-place; if it does not exist, a new row is
|
||||
inserted. Running ``refresh_prices`` twice in a row must not double-insert.
|
||||
- **No destructive operations**: this service never deletes ``tibber_price``
|
||||
rows. Only INSERT or UPDATE.
|
||||
- **Exception propagation**: exceptions from the Tibber client are *not*
|
||||
swallowed here. The scheduled job wrapper in ``main.py`` is responsible for
|
||||
catching and logging errors so that a single fetch failure does not crash the
|
||||
scheduler.
|
||||
- **SQLite timezone note**: ``DateTime(timezone=True)`` columns come back as
|
||||
timezone-naive UTC datetimes on read. We store UTC-aware datetimes on write
|
||||
(``starts_at`` from ``PricePoint`` is already UTC-aware; ``fetched_at`` is
|
||||
``datetime.now(UTC)``). Reads elsewhere that compare against these values
|
||||
must normalise accordingly.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.integrations.tibber.client import fetch_price_range
|
||||
from app.models.energy import EnergyContract, TibberPrice
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _active_tibber_contract_exists(session: Session) -> bool:
|
||||
"""Return True if there is an active contract with kind='tibber'."""
|
||||
row = session.execute(
|
||||
select(EnergyContract).where(
|
||||
EnergyContract.active.is_(True),
|
||||
EnergyContract.kind == "tibber",
|
||||
).limit(1)
|
||||
).scalar_one_or_none()
|
||||
return row is not None
|
||||
|
||||
|
||||
def refresh_prices(session: Session, settings: object) -> int:
|
||||
"""Fetch today-and-tomorrow Tibber prices and upsert them into ``tibber_price``.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
session:
|
||||
An active SQLAlchemy session. The caller is responsible for closing it;
|
||||
this function commits each upserted row individually to keep transactions
|
||||
short.
|
||||
settings:
|
||||
A runtime settings object that exposes ``tibber_api_token`` and
|
||||
``tibber_home_id`` attributes (typically a ``Settings`` or merged
|
||||
runtime-settings instance from ``build_runtime_settings``).
|
||||
|
||||
Returns
|
||||
-------
|
||||
int
|
||||
Number of rows upserted (inserted or updated). Returns 0 for a no-op.
|
||||
|
||||
Notes
|
||||
-----
|
||||
- No-op (returns 0) when:
|
||||
* No active contract exists with ``kind="tibber"``, OR
|
||||
* ``settings.tibber_api_token`` is empty or whitespace.
|
||||
- Exceptions from the Tibber client are *not* caught here; they propagate to
|
||||
the caller (the scheduled job wrapper swallows them and logs).
|
||||
"""
|
||||
token: str = getattr(settings, "tibber_api_token", "") or ""
|
||||
if not token.strip():
|
||||
logger.debug("refresh_prices: tibber_api_token is empty — no-op")
|
||||
return 0
|
||||
|
||||
if not _active_tibber_contract_exists(session):
|
||||
logger.debug("refresh_prices: no active tibber contract — no-op")
|
||||
return 0
|
||||
|
||||
home_id: str = getattr(settings, "tibber_home_id", "") or ""
|
||||
home_id_or_none: str | None = home_id.strip() or None
|
||||
|
||||
logger.info("refresh_prices: fetching Tibber price range (home_id=%r)", home_id_or_none)
|
||||
|
||||
# May raise TibberError or TibberAuthError — let them propagate.
|
||||
price_points = fetch_price_range(token, home_id_or_none)
|
||||
|
||||
if not price_points:
|
||||
logger.info("refresh_prices: API returned zero price points — nothing to upsert")
|
||||
return 0
|
||||
|
||||
fetched_at = datetime.now(UTC)
|
||||
upserted = 0
|
||||
|
||||
for point in price_points:
|
||||
existing = session.execute(
|
||||
select(TibberPrice).where(TibberPrice.starts_at == point.starts_at)
|
||||
).scalar_one_or_none()
|
||||
|
||||
if existing is not None:
|
||||
# Update in-place — price fields may change (e.g. Tibber corrects a
|
||||
# forecast), but starts_at and resolution stay the same.
|
||||
existing.total = point.total
|
||||
existing.energy = point.energy
|
||||
existing.tax = point.tax
|
||||
existing.level = point.level
|
||||
existing.currency = point.currency
|
||||
existing.fetched_at = fetched_at
|
||||
else:
|
||||
session.add(
|
||||
TibberPrice(
|
||||
starts_at=point.starts_at,
|
||||
resolution=point.resolution,
|
||||
total=point.total,
|
||||
energy=point.energy,
|
||||
tax=point.tax,
|
||||
level=point.level,
|
||||
currency=point.currency,
|
||||
fetched_at=fetched_at,
|
||||
)
|
||||
)
|
||||
|
||||
upserted += 1
|
||||
|
||||
session.commit()
|
||||
|
||||
logger.info("refresh_prices: upserted %d price points", upserted)
|
||||
return upserted
|
||||
Reference in New Issue
Block a user