Installation

pip install requests

Minimal client

import requests
from typing import Optional, Dict

class NatomradeClient:
    BASE = "https://natomrade.airpipe.ai"

    def __init__(self, api_key: Optional[str] = None):
        self.session = requests.Session()
        if api_key:
            self.session.headers["X-API-Key"] = api_key

    def lookup_postal(self, postnr: str) -> Dict:
        prefix = "/api/v1" if "X-API-Key" in self.session.headers else "/web"
        r = self.session.get(f"{self.BASE}{prefix}/lookup/postal",
                              params={"code": postnr}, timeout=10)
        r.raise_for_status()
        return r.json()

    def lookup_address(self, adress: str, ort: str = "", postnr: str = "") -> Dict:
        prefix = "/api/v1" if "X-API-Key" in self.session.headers else "/web"
        r = self.session.get(f"{self.BASE}{prefix}/lookup/address",
                              params={"adress": adress, "ort": ort, "postnr": postnr},
                              timeout=10)
        r.raise_for_status()
        return r.json()

    def spot_prices_now(self) -> Dict:
        r = self.session.get(f"{self.BASE}/api/v1/prices/current", timeout=10)
        r.raise_for_status()
        return r.json()

    def production_now(self) -> Dict:
        r = self.session.get(f"{self.BASE}/api/v1/production/current", timeout=10)
        r.raise_for_status()
        return r.json()

Användning

client = NatomradeClient(api_key="zk_din_nyckel")

r = client.lookup_postal("11151")
print(r["grid_owner"]["name"])     # Ellevio AB
print(r["elomrade"]["code"])       # SE3

prices = client.spot_prices_now()
for p in prices["now"]:
    print(f"{p['area_code']}: {p['sek_per_kwh']*100:.1f} öre/kWh")

Med automatisk retry

import time

class NatomradeClient:
    def _request(self, method, path, **kw):
        for attempt in range(5):
            r = self.session.request(method, f"{self.BASE}{path}", timeout=10, **kw)
            if r.status_code == 200:
                return r.json()
            if r.status_code == 429:
                wait = r.json().get("throttle_minutes", 0) * 60 or 2**attempt
                time.sleep(wait); continue
            if r.status_code >= 500:
                time.sleep(2**attempt); continue
            r.raise_for_status()
        raise Exception("Max retries")

Async-version (aiohttp)

import aiohttp

class AsyncNatomradeClient:
    BASE = "https://natomrade.airpipe.ai"

    def __init__(self, api_key=None):
        self.headers = {"X-API-Key": api_key} if api_key else {}

    async def lookup_postal(self, postnr):
        async with aiohttp.ClientSession() as s:
            async with s.get(f"{self.BASE}/api/v1/lookup/postal",
                              params={"code": postnr},
                              headers=self.headers) as r:
                return await r.json()