octopus-kraken 0.1.0

A tiny, tweepy-style Python wrapper for the Octopus Energy API (read-only)

🐙 octopy

A tiny, read-only Python wrapper for the Octopus Energy API, modelled on tweepy: construct a client once, then call get_* methods that return typed model objects.

Targets Octopus Energy Spain (api.oees-kraken.energy, Kraken-token auth) and Octopus UK (api.octopus.energy, HTTP Basic auth), whose public endpoints are handy for credential-free exploration and tests.

REST or GraphQL? The UK has public REST consumption endpoints, so the Client is a GET-only REST reader. Octopus Spain exposes its account and consumption data through the Kraken GraphQL API instead — that path lives in octopy.graphql.KrakenGraphQLClient. Auth for Spain is a Kraken token minted via a GraphQL mutation, and there is no public products endpoint.

Looking for the web UI? It now lives in its own repo: octopus-energy-web.

What you get

• octopy.Client — GET methods for products, tariff unit rates & standing charges, electricity/gas consumption, account details and postcode→region (GSP) lookups; typed models; a tweepy-style Paginator; a status-code exception hierarchy; and a gas m³→kWh helper. Factories: Client.spain(token=…) and Client.uk(api_key=…).
• octopy.graphql.KrakenGraphQLClient — the Spain path (GraphQL, the one that actually carries ES data): login (obtainKrakenToken), account_numbers, hourly_consumption, account_summary.
• octopy.aio.AsyncKrakenGraphQLClient — an awaitable version of the above for asyncio / Home Assistant (optional [async] extra).
• scripts/obtain_token.py — mints a Kraken token for Spain (the GET-only client can't authenticate itself there).
• scripts/explore_spain.py — lists accounts + recent hourly consumption in one command (the quickest end-to-end check).

Layout

octopy/                # the package (pip-installable, depends only on requests)
├─ client.py           #   Client + factories + all GET methods
├─ models.py           #   Product, Consumption, Rate, Account, …
├─ countries.py        #   per-country config registry (base URLs, auth)
├─ pagination.py       #   Page + Paginator (tweepy-style)
├─ exceptions.py       #   HTTPException, Unauthorized, NotFound, …
├─ units.py            #   m3_to_kwh / kwh_to_m3
├─ graphql.py          #   KrakenGraphQLClient — the Spain (GraphQL) path
└─ aio.py              #   AsyncKrakenGraphQLClient (optional [async] extra)
scripts/               # obtain_token.py (mint token) · explore_spain.py (demo)
tests/                 # test_client.py · test_graphql.py · smoke_live.py
pyproject.toml · requirements.txt · .env.example

Install

System Python may be old; use a modern one in a venv.

python3.13 -m venv .venv
.venv/bin/pip install -e .          # editable install of the package
.venv/bin/pip install -e ".[async]" # …plus the async client (aiohttp)

.venv/bin/python -m unittest discover -s tests   # offline tests
.venv/bin/python tests/smoke_live.py             # live UK-public smoke test

Once published to PyPI: pip install octopus-kraken (imports as octopy), or pip install "octopus-kraken[async]" for the async client.

Using the wrapper

import octopy
from datetime import datetime, timedelta, timezone

# --- Octopus Energy Spain: data is via GraphQL (Kraken) ---
from octopy.graphql import KrakenGraphQLClient
es = KrakenGraphQLClient.login(api_key="<kraken-api-key>")   # or email=/password=
end = datetime.now(timezone.utc); start = end - timedelta(days=2)
for acct in es.account_numbers():
    rows = es.hourly_consumption(acct, start, end)           # electricity, hourly
    print(acct, len(rows), "measurements")
# one-shot from the shell:  python scripts/explore_spain.py --api-key <KEY>

# --- Octopus UK: public data needs no key (great for exploring/testing) ---
uk = octopy.Client.uk()
for p in uk.get_products(is_green=True):
    print(p.code, p.display_name)
region = uk.get_grid_supply_points("BS1")[0].group_id          # "_B"
tariff = uk.get_product("AGILE-24-10-01").electricity_tariff_code(region)
rates  = uk.get_electricity_standard_unit_rates("AGILE-24-10-01", tariff)

# Walk every page, tweepy-style; convert SMETS2 gas m³ → kWh
from octopy import Paginator
all_rates = list(Paginator(uk.get_electricity_standard_unit_rates,
                           "AGILE-24-10-01", tariff).flatten())
print(octopy.m3_to_kwh(120.0), "kWh")

Getting a Spain (Kraken) token

python scripts/obtain_token.py --api-key <KRAKEN_API_KEY>     # or --email/--password

Pass the printed token to Client.spain(token=…) / KrakenGraphQLClient(token=…).

Choosing a country

Octopus runs a separate API deployment per country. Select one with country= (a code/alias, case-insensitive) on either client — it sets the base URL and GraphQL endpoint for you. The Client.uk() / Client.spain() factories are shorthands for country="GB" / country="ES".

octopy.Client(country="GB", api_key="sk_live_…")   # == Client.uk(...)
octopy.Client(country="ES", token="…")             # == Client.spain(...)
octopy.KrakenGraphQLClient(country="GB")           # UK Kraken GraphQL endpoint
octopy.KrakenGraphQLClient.login(country="ES", api_key="…")

Defaults reflect each transport's reality: the REST Client defaults to the UK (public, key-free data) while the GraphQL client defaults to Spain (the deployment whose account/consumption data is GraphQL-only).

Built-in countries are GB (aliases: UK) and ES. Add another Kraken deployment at runtime:

octopy.register_country(octopy.Country(
    code="DE", name="Germany",
    base_url="https://api.oeg-kraken.energy/v1",   # example — verify the host
))
octopy.Client(country="DE", token="…")

Async (asyncio / Home Assistant)

For asyncio apps, octopy.aio.AsyncKrakenGraphQLClient is an awaitable counterpart to KrakenGraphQLClient with the same methods. Install the extra (pip install "octopus-kraken[async]"); the base package stays aiohttp-free.

from octopy.aio import AsyncKrakenGraphQLClient

async with await AsyncKrakenGraphQLClient.login(country="ES", api_key="…") as es:
    for number in await es.account_numbers():
        rows = await es.hourly_consumption(number, start, end)

Pass an existing aiohttp.ClientSession to reuse a shared connection pool — in Home Assistant, async_get_clientsession(hass). The auth token is sent per request, so sharing a session never leaks your credential onto it, and octopy only closes sessions it created itself.

Notes & safety

• Read-only by design. No data is ever modified. (Token minting is the one isolated POST, in scripts/obtain_token.py.)
• Gas units: SMETS1 reports kWh, SMETS2 reports m³ — use octopy.m3_to_kwh / kwh_to_m3 to convert.
• Errors map to typed exceptions (Unauthorized, NotFound, TooManyRequests, …), all subclasses of octopy.OctopyException.