Auf Daten zugreifen
In diesem Tutorial holen Sie ein Access-Token, senden Ihre erste GraphQL-Abfrage und lesen dann Runtime-Entities, folgen Assoziationen und fragen Zeitreihen ab: Rohwerte, Summen je Tag und Monat sowie vorab verdichtete Rollups. Alle Beispiele lesen nur.
Die Beispiele verwenden das Energiegemeinschafts-Modell aus Tutorial 3, weil es Stammdaten und Zeitreihen hat. Die Abfragemuster sind für jedes Construction Kit dieselben.
Dauer: etwa 60 Minuten.
Der Endpunkt
Jeder Tenant hat einen GraphQL-Endpunkt am Asset Repository:
https://assets.<your-domain>/tenants/<tenant>/GraphQL
Senden Sie Abfragen als POST mit einem JSON-Body {"query": "...", "variables": {...}}
und dem Header Authorization: Bearer <access-token>. Verwenden Sie den Pfad
genau wie gezeigt, mit /tenants/ vor der Tenant-Id; die Schreibweise GraphQL
ist die in dieser Dokumentation durchgängig verwendete.
Zum interaktiven Erkunden stellt derselbe Dienst einen Playground mit
Schema-Browser und Autovervollständigung unter
https://assets.<your-domain>/tenants/<tenant>/graphql/playground bereit.
Schritt 1: Ein Access-Token holen
Schnellstart: die octo-cli-Anmeldung wiederverwenden
Wenn Sie Tutorial 1 gefolgt sind, hält octo-cli bereits
ein Token für Ihren Kontext. Für Experimente können Sie es aus der Kontextdatei lesen:
octo-cli --context <context-name> -c AuthStatus > /dev/null # refreshes the token if needed
export OCTO_TOKEN=$(jq -r '.Contexts["<context-name>"].Authentication.AccessToken' ~/.octo-cli/contexts.json)
export OCTO_GRAPHQL=https://assets.<your-domain>/tenants/<tenant>/GraphQL
Das Token ist etwa eine Stunde gültig. Es trägt die Rollen Ihres Benutzers; ein Skript, das es verwendet, kann also genau das, was Sie im Studio können.
Eigene Anwendung: Authorization Code mit PKCE
Eine Web- oder Desktop-Anwendung, die im Namen eines Benutzers handelt,
verwendet den OpenID-Connect-Flow Authorization Code mit PKCE. Ihre
Administration registriert dafür einen OAuth-Client im Tenant (siehe
Eine neue Webanwendung registrieren)
und gewährt ihm den Scope octo_api (oder octo_api.read_only für eine
Anwendung, die nur liest).
Der Ablauf:
- Ihre App erzeugt einen zufälligen
code_verifierund dessen SHA-256-Hash, diecode_challenge. - Sie leitet den Browser zum Authorize-Endpunkt des Identity Service, mit
acr_values=tenant:<tenant>, damit sich die Anmeldeseite des richtigen Tenants öffnet. - Der Benutzer meldet sich an, der Browser kehrt mit einem
codezu Ihrerredirect_urizurück. - Ihre App tauscht
codepluscode_verifieram Token-Endpunkt gegen Tokens.
Der Identity Service veröffentlicht alle Endpunkte in seinem Discovery-Dokument
unter https://connect.<your-domain>/.well-known/openid-configuration. Eine
minimale Python-Implementierung für ein lokales Werkzeug, nur mit der
Standardbibliothek:
import base64, hashlib, http.server, json, os, secrets, urllib.parse, urllib.request, webbrowser
ISSUER = "https://connect.<your-domain>"
CLIENT_ID = "<your-client-id>" # registered with redirect URI http://localhost:8765/callback
TENANT = "<tenant>"
REDIRECT_URI = "http://localhost:8765/callback"
verifier = secrets.token_urlsafe(64)
challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
state = secrets.token_urlsafe(16)
params = {
"response_type": "code",
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"scope": "openid profile email role octo_api offline_access",
"acr_values": f"tenant:{TENANT}",
"code_challenge": challenge,
"code_challenge_method": "S256",
"state": state,
}
webbrowser.open(f"{ISSUER}/connect/authorize?{urllib.parse.urlencode(params)}")
class Callback(http.server.BaseHTTPRequestHandler):
def do_GET(self):
query = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
assert query["state"][0] == state, "state mismatch"
self.server.code = query["code"][0]
self.send_response(200); self.end_headers()
self.wfile.write(b"Login complete - you can close this window.")
server = http.server.HTTPServer(("localhost", 8765), Callback)
server.handle_request()
token_request = urllib.parse.urlencode({
"grant_type": "authorization_code",
"client_id": CLIENT_ID,
"code": server.code,
"redirect_uri": REDIRECT_URI,
"code_verifier": verifier,
}).encode()
tokens = json.load(urllib.request.urlopen(f"{ISSUER}/connect/token", data=token_request))
access_token = tokens["access_token"] # keep it in memory, never log it
Für Browser-Anwendungen verwenden Sie statt handgeschriebenem Code eine bewährte
OIDC-Bibliothek (zum Beispiel angular-auth-oidc-client oder oidc-client-ts).
Die Hintergründe beschreibt
Authentifizierung.
Dienste ohne Benutzer: Client Credentials
Ein Hintergrunddienst oder ein Gerät verwendet den Flow Client Credentials mit eigener Client-Id und eigenem Secret. Das behandelt Tutorial 5.
Schritt 2: Ihre erste Abfrage
curl -s "$OCTO_GRAPHQL" \
-H "Authorization: Bearer $OCTO_TOKEN" \
-H "Content-Type: application/json" \
-d '{"query":"{ runtime { energyCommunityCustomer(first: 2) { totalCount items { rtId rtWellKnownName } } } }"}'
{
"data": {
"runtime": {
"energyCommunityCustomer": {
"totalCount": 45,
"items": [
{ "rtId": "6ac75edd1f682dc27a1b1d9f", "rtWellKnownName": "customer-018" },
{ "rtId": "6ac75ede1f682dc27a1b1e12", "rtWellKnownName": "customer-041" }
]
}
}
}
}
Das Wurzelfeld runtime enthält ein typisiertes Feld pro CK-Typ des Tenants.
Sein Name ist die CK-Typ-Id in camelCase ohne Trennzeichen:
aus EnergyCommunity/Customer wird energyCommunityCustomer,
aus Basic.Energy/EnergyMeasurement wird basicEnergyEnergyMeasurement.
Schritt 3: Runtime-Entities, Filter und Paging
Attribute auswählen
Typisierte Felder bieten die Attribute des Typs als GraphQL-Felder an, Records als verschachtelte Objekte:
query Customers {
runtime {
energyCommunityCustomer(first: 10) {
totalCount
items {
rtId
customerNumber
contact { firstName lastName companyName }
}
}
}
}
Filtern
fieldFilter filtert nach Attributwerten. Mehrere Filter werden mit UND verknüpft:
query ConsumptionAnchors {
runtime {
basicEnergyEnergyMeasurement(
first: 50
fieldFilter: [{ attributePath: "obisCode", operator: EQUALS, comparisonValue: "1-1:1.9.0 G.01" }]
) {
totalCount
items { rtId rtWellKnownName obisCode }
}
}
}
Alle Operatoren, die Textsuche und die Sortierung beschreibt Suche und Filter.
Große Ergebnisse seitenweise lesen
Connections werden mit first (Seitengröße) und after (Cursor) geblättert.
Lesen Sie pageInfo.endCursor und übergeben Sie ihn als after, bis
hasNextPage den Wert false hat:
query Page($after: String) {
runtime {
basicEnergyEnergyMeasurement(first: 500, after: $after) {
pageInfo { hasNextPage endCursor }
items { rtId obisCode }
}
}
}
Setzen Sie immer first. Ohne Seitengröße liefert eine Abfrage alles, was sie
findet, und das ist bei großen Tenants langsam.
Schritt 4: Assoziationen folgen
Assoziationen erscheinen als Navigationsfelder, benannt nach der
Assoziationsrolle, zum Beispiel facilities an einem Kunden oder children an
einer Anlage. Da das Ziel einer Rolle mehrere Typen haben kann, verlangt ein
Navigationsfeld die Liste der Zieltypen (ckTypeIds) und liefert eine Union.
Wählen Sie die Felder je Typ mit einem Inline-Fragment (... on <GraphQLTypeName>):
query CustomerTree {
runtime {
energyCommunityCustomer(first: 2) {
items {
customerNumber
facilities(ckTypeIds: ["Basic.Energy/OperatingFacility"]) {
items {
... on BasicEnergyOperatingFacility {
name
children(ckTypeIds: ["EnergyCommunity/Consumer", "EnergyCommunity/Producer"]) {
items {
... on EnergyCommunityConsumer { ckTypeId meteringPointNumber partitionFactor }
... on EnergyCommunityProducer { ckTypeId meteringPointNumber productionType }
}
}
}
}
}
}
}
}
}
{
"customerNumber": "20041",
"facilities": {
"items": [
{
"name": "PV plant 041",
"children": {
"items": [
{
"ckTypeId": "EnergyCommunity/Producer",
"meteringPointNumber": "AT0099990000090012000000000000009",
"productionType": "SOLAR"
}
]
}
}
]
}
}
Die Gegenrichtung funktioniert genauso: Ein Energiemessungs-Anker erreicht seinen
Zählpunkt über parent(ckTypeIds: [...]). Abfragen über beliebige
Assoziationsrollen und transitive Beziehungen beschreibt
Assoziationen.
Schritt 5: Stream Data abfragen
Zeitreihen werden aus einem Archiv gelesen, adressiert über dessen rtId.
Die Archive Ihres Tenants finden Sie im Studio unter Archive, oder Sie fragen
GraphQL, welche Archivfamilie (das Basisarchiv und alle seine Rollups) welchen
Zeitraum abdeckt:
query Coverage {
streamData {
coverageFor(rtId: "<archive-rtId>") {
archiveRtId
rtWellKnownName
status
isBase
bucketSizeMs
bucketAlignment
storedFunctions
availableFrom
availableTo
}
}
}
Ad-hoc-Abfragen laufen über streamData.transientStreamDataQuery, das vier
Abfragearten bietet: simple (Zeilen), aggregation, groupingAggregation und
downsampling. Das Ergebnis enthält die Spaltenliste in columns und die Daten
in rows, einer geblätterten Connection von Zeilen mit cells.
Rohwerte (15-Minuten-Fenster)
Das Rohdatenarchiv der Energiegemeinschaft ist ein Time-Range-Archiv: Jede
Zeile deckt ein Fenster [window_start, window_end) ab, hier 15 Minuten. rtIds
beschränkt die Abfrage auf die Anker, die Sie interessieren, arg.from/arg.to
auf einen Zeitraum (UTC):
query RawValues($archive: OctoObjectId!, $anchor: OctoObjectId!, $from: DateTime!, $to: DateTime!) {
streamData {
transientStreamDataQuery {
simple(
archiveRtId: $archive
rtIds: [$anchor]
columnPaths: ["window_start", "window_end", "amount.value", "amount.unit", "obisCode", "dataQuality"]
arg: { from: $from, to: $to, queryMode: DEFAULT }
sortOrder: [{ attributePath: "window_start", sortOrder: ASCENDING }]
) {
items {
columns { attributePath }
rows(first: 96) {
totalCount
pageInfo { hasNextPage endCursor }
items { rtId timestamp cells { items { attributePath value } } }
}
}
}
}
}
}
Mit from = 2026-10-05T22:00:00Z und to = 2026-10-06T22:00:00Z (ein
Kalendertag in Wien, Sommerzeit) hat das Ergebnis totalCount: 96, eine Zeile
pro Viertelstunde:
{
"rtId": "6ac75edd1f682dc27a1b1dba",
"timestamp": "2026-10-05T22:15:00Z",
"cells": {
"items": [
{ "attributePath": "window_start", "value": "2026-10-05T22:00:00Z" },
{ "attributePath": "window_end", "value": "2026-10-05T22:15:00Z" },
{ "attributePath": "amount.value", "value": 0.065513 },
{ "attributePath": "amount.unit", "value": "KWh" },
{ "attributePath": "obisCode", "value": "1-1:1.9.0 G.01" },
{ "attributePath": "dataQuality", "value": "L1" }
]
}
}
Spaltenpfade sind die CK-Attributpfade in camelCase (amount.value,
obisCode); Time-Range-Archive ergänzen die Fensterspalten window_start und
window_end. Zeilen werden mit rows(first:, after:) geblättert, genau wie
Runtime-Connections.
Summen: SUM je Register für einen Tag
groupingAggregation aggregiert auf dem Server. Diese Abfrage summiert alle
15-Minuten-Werte eines Tages je OBIS-Code, über alle Anker des Tenants:
query DailyTotalsByObis($archive: OctoObjectId!, $from: DateTime!, $to: DateTime!) {
streamData {
transientStreamDataQuery {
groupingAggregation(
archiveRtId: $archive
groupByColumnPaths: ["obisCode"]
columnPaths: [{ attributePath: "amount.value", aggregationType: SUM }]
arg: { from: $from, to: $to, queryMode: DEFAULT }
) {
items {
columns { attributePath }
rows { items { cells { items { attributePath value } } } }
}
}
}
}
}
Die Ergebnisspalten sind obisCode und amountvalue_sum, zum Beispiel:
| obisCode | amountvalue_sum (kWh) |
|---|---|
1-1:1.9.0 G.01 | 431,36 |
1-1:2.9.0 G.01 | 474,27 |
1-1:2.9.0 G.01T | 474,01 |
1-1:2.9.0 G.03 | 207,63 |
1-1:2.9.0 P.01T | 266,38 |
Tutorial 3 erklärt, was diese Register bedeuten.
Rollups: SUM je Tag oder Monat, ohne Rohdaten anzufassen
Ein Jahr 15-Minuten-Werte zu summieren heißt, etwa 35.000 Zeilen pro Anker zu
lesen. Rollup-Archive enthalten die Summen bereits. In einem Rollup heißen die
aggregierten Spalten <column>_<function>, zum Beispiel amountvalue_sum und
dataquality_max. Welches logische Attribut ein Rollup verdichtet, liefert
streamData.rollupQueryMetadata(rtId:).
Tageswerte eines Ankers aus dem Tages-Rollup:
query DailyRollup($archive: OctoObjectId!, $anchor: OctoObjectId!, $from: DateTime!, $to: DateTime!) {
streamData {
transientStreamDataQuery {
simple(
archiveRtId: $archive
rtIds: [$anchor]
columnPaths: ["window_start", "window_end", "amountvalue_sum", "dataquality_max"]
arg: { from: $from, to: $to, queryMode: DEFAULT }
sortOrder: [{ attributePath: "window_start", sortOrder: ASCENDING }]
) {
items { rows(first: 31) { totalCount items { cells { items { attributePath value } } } } }
}
}
}
}
Kalender-Rollups sind an einer Referenzzeitzone ausgerichtet (zum Beispiel
Europe/Vienna), ein „Tag“ beginnt also um lokale Mitternacht: 22:00Z im
Sommer, 23:00Z im Winter. Übergeben Sie from und to auf diesen Grenzen.
Für Monatssummen einer Gruppe von Ankern fragen Sie das Monats-Rollup mit der Liste der Anker-Ids ab und gruppieren nach dem Fensterbeginn, wie es das Python-Beispiel unten tut.
Schritt 6: Ein vollständiges Python-Beispiel
Das Skript berechnet je Monat zwei Kennzahlen einer Energiegemeinschaft, die Tutorial 3 erklärt: den Anteil des Verbrauchs, der aus der Gemeinschaft gedeckt wurde, und den Anteil der angebotenen Erzeugung, der nicht verteilt wurde. Es kombiniert alles aus diesem Tutorial: eine gefilterte Runtime-Abfrage mit Paging, um die Anker jedes Registers zu finden, und eine gruppierte Aggregation über das Monats-Rollup.
"""Monthly coverage ratio and surplus ratio of an energy community.
export OCTO_GRAPHQL=https://assets.<your-domain>/tenants/<tenant>/GraphQL
export OCTO_TOKEN=... # see step 1
python monthly_kpis.py 2026-06 2026-09
"""
import os
import sys
from datetime import datetime
from zoneinfo import ZoneInfo
import requests
ENDPOINT = os.environ["OCTO_GRAPHQL"]
TOKEN = os.environ["OCTO_TOKEN"]
VIENNA = ZoneInfo("Europe/Vienna")
MONTHLY_ROLLUP = "<monthly-rollup-rtId>" # e.g. the archive "energy-measurements-monthly"
REGISTERS = {
"consumption": "1-1:1.9.0 G.01",
"self_coverage": "1-1:2.9.0 G.03",
"offered": "1-1:2.9.0 G.01T",
"surplus": "1-1:2.9.0 P.01T",
}
ANCHORS_QUERY = """
query Anchors($obis: SimpleScalar!, $after: String) {
runtime {
basicEnergyEnergyMeasurement(
first: 500, after: $after,
fieldFilter: [{ attributePath: "obisCode", operator: EQUALS, comparisonValue: $obis }]
) {
pageInfo { hasNextPage endCursor }
items { rtId }
}
}
}
"""
MONTHLY_QUERY = """
query Monthly($archive: OctoObjectId!, $rtIds: [OctoObjectId], $from: DateTime!, $to: DateTime!) {
streamData {
transientStreamDataQuery {
groupingAggregation(
archiveRtId: $archive
rtIds: $rtIds
groupByColumnPaths: ["window_start"]
columnPaths: [{ attributePath: "amountvalue_sum", aggregationType: SUM }]
arg: { from: $from, to: $to, queryMode: DEFAULT }
) {
items { rows { items { cells { items { attributePath value } } } } }
}
}
}
}
"""
def graphql(query: str, variables: dict) -> dict:
response = requests.post(
ENDPOINT,
json={"query": query, "variables": variables},
headers={"Authorization": f"Bearer {TOKEN}"},
timeout=60,
)
response.raise_for_status() # 401, 400 (invalid query), 5xx
body = response.json()
if body.get("errors"): # execution errors arrive with HTTP 200!
raise RuntimeError(body["errors"][0]["message"])
return body["data"]
def anchors_for(obis: str) -> list[str]:
ids, after = [], None
while True:
page = graphql(ANCHORS_QUERY, {"obis": obis, "after": after})
conn = page["runtime"]["basicEnergyEnergyMeasurement"]
ids += [item["rtId"] for item in conn["items"]]
if not conn["pageInfo"]["hasNextPage"]:
return ids
after = conn["pageInfo"]["endCursor"]
def month_start(year: int, month: int) -> str:
"""Local midnight in Europe/Vienna, expressed in UTC."""
return datetime(year, month, 1, tzinfo=VIENNA).astimezone(ZoneInfo("UTC")).isoformat()
def monthly_sums(rt_ids: list[str], first: str, last: str) -> dict[str, float]:
y1, m1 = map(int, first.split("-"))
y2, m2 = map(int, last.split("-"))
y2, m2 = (y2 + 1, 1) if m2 == 12 else (y2, m2 + 1) # exclusive upper bound
data = graphql(MONTHLY_QUERY, {
"archive": MONTHLY_ROLLUP,
"rtIds": rt_ids,
"from": month_start(y1, m1),
"to": month_start(y2, m2),
})
result = {}
rows = data["streamData"]["transientStreamDataQuery"]["groupingAggregation"]["items"][0]["rows"]["items"]
for row in rows:
cells = {c["attributePath"]: c["value"] for c in row["cells"]["items"]}
local = datetime.fromisoformat(cells["window_start"].replace("Z", "+00:00")).astimezone(VIENNA)
result[local.strftime("%Y-%m")] = cells["amountvalue_sum_sum"]
return result
def main() -> None:
first, last = sys.argv[1], sys.argv[2]
sums = {name: monthly_sums(anchors_for(obis), first, last) for name, obis in REGISTERS.items()}
print(f"{'month':8} {'consumption':>12} {'self-cov.':>10} {'coverage':>9} {'surplus %':>9}")
for month in sorted(sums["consumption"]):
consumption = sums["consumption"].get(month, 0.0)
covered = sums["self_coverage"].get(month, 0.0)
offered = sums["offered"].get(month, 0.0)
surplus = sums["surplus"].get(month, 0.0)
coverage = covered / consumption if consumption else 0.0
surplus_ratio = surplus / offered if offered else 0.0
print(f"{month:8} {consumption:12.1f} {covered:10.1f} {coverage:9.1%} {surplus_ratio:9.1%}")
if __name__ == "__main__":
main()
Beispielausgabe:
month consumption self-cov. coverage surplus %
2026-06 12940.8 7950.0 61.4% 61.7%
2026-07 13372.1 8093.9 60.5% 61.4%
2026-08 13372.1 7623.0 57.0% 60.2%
2026-09 12737.3 6720.7 52.8% 58.1%
Typische Fallen
GraphQL-Fehler kommen mit HTTP 200. Nur Transportprobleme erzeugen einen
HTTP-Fehlerstatus: 401 bei fehlendem oder abgelaufenem Token, 400 bei einer
Abfrage, die nicht zum Schema passt. Eine gültige Abfrage, die bei der Ausführung
scheitert (ein unbekannter Spaltenpfad, eine fehlende Berechtigung auf einem Feld,
eine serverseitige Ausnahme), liefert HTTP 200 mit einem errors-Array und
null-Daten für den gescheiterten Teil:
{
"errors": [{ "message": "Invalid column paths: timeRange.from", "extensions": { "code": "ASSET1004" } }],
"data": { "streamData": { "transientStreamDataQuery": { "simple": null } } }
}
Prüfen Sie immer errors im Antwort-Body, wie es graphql() im Beispiel tut.
Ohne attributeNames kommen alle Attribute. Die generische Abfrage
runtimeEntities liefert Attribute als Name/Wert-Paare über
attributes(attributeNames: [...]). Lassen Sie das Argument weg, erhalten Sie
alle Attribute der Entity, einschließlich Zugangsdaten, die an
Konfigurations-Entities gespeichert sind. Übergeben Sie immer eine explizite,
nicht leere Liste. Die Namen werden nur in camelCase erkannt (["host"]
funktioniert, ["Host"] liefert nichts), und wenn Sie ein Record-Attribut lesen,
geben Sie auch die Unterattribute des Records an. Typisierte Abfragen wie
energyCommunityCustomer { contact { ... } } haben dieses Problem nicht, weil
Sie jedes Feld explizit auswählen. Siehe
Secret-Attribute.
Den exakten Endpunktpfad verwenden. Er lautet /tenants/<tenant>/GraphQL auf
dem Host des Asset Repository, nicht auf dem Studio-Host und nicht ohne /tenants/.
Zeitzonen. Archive speichern UTC. Kalender-Rollups werden um lokale
Mitternacht ihrer Referenzzeitzone geschnitten. Rechnen Sie lokale Tages- oder
Monatsgrenzen vor der Abfrage in UTC um, wie es month_start() tut.
Angeschnittene Fenster. Eine Time-Range-Abfrage liefert jedes Fenster, das
[from, to) überlappt. Legen Sie from und to auf das Fensterraster (volle
Viertelstunden beim Rohdatenarchiv, lokale Mitternacht bei Tages-Rollups), sonst
zählt ein nur teilweise enthaltenes Fenster mit seinem vollen Wert.
Typnamen. Das typisierte GraphQL-Feld ist camelCase
(basicEnergyOperatingFacility), der in Fragmenten verwendete GraphQL-Typ ist
PascalCase (BasicEnergyOperatingFacility), und Filter sowie ckTypeIds
verwenden die CK-Typ-Id (Basic.Energy/OperatingFacility).
Zusammenfassung
- Ein Token holen Sie für Experimente aus
octo-cli, für eine eigene benutzerbezogene App über Authorization Code + PKCE oder für einen Dienst über Client Credentials. runtimehat ein typisiertes Feld pro CK-Typ, mitfieldFilter, Paging und Navigationsfeldern für Assoziationen.streamData.transientStreamDataQueryliest Rohzeilen, aggregiert und gruppiert auf dem Server; Rollup-Archive liefern Tages- oder Monatssummen direkt.- Prüfen Sie das
errors-Array auch dann, wenn der HTTP-Status 200 ist.
Weiter: Energiedaten verstehen erklärt das Modell und die Register, die Sie gerade abgefragt haben.