Python

The GraphQL endpoint is called via POST requests with a JSON body. No authentication is required.

Single query

import requests

ENDPOINT = "https://data.bafu.admin.ch/api"

query = """
{
  water {
    observations {
      stations(where: { status: { _eq: "Aufgebaut" } }, limit: 5) {
        no
        name
        riverName
        latitude
        longitude
      }
    }
  }
}
"""

response = requests.post(ENDPOINT, json={"query": query}, timeout=30)
response.raise_for_status()
payload = response.json()

stations = payload["data"]["water"]["observations"]["stations"]
for station in stations:
    print(station["no"], station["name"])

Query with variables

Variables separate values from the query text and avoid quoting issues.

import requests

ENDPOINT = "https://data.bafu.admin.ch/api"

query = """
query DailyMean($from: AWSDateTime!, $to: AWSDateTime!, $station: String!) {
  water {
    observations {
      data_1day_mean(
        where: {
          station: { no: { _eq: $station } }
          timestamp: { _gte: $from, _lt: $to }
        }
      ) {
        timestamp
        parameterName
        value
        unitSymbol
      }
    }
  }
}
"""

variables = {
    "from": "2026-01-01T00:00:00Z",
    "to": "2026-02-01T00:00:00Z",
    "station": "2009",
}

response = requests.post(
    ENDPOINT,
    json={"query": query, "variables": variables},
    timeout=30,
)
response.raise_for_status()
rows = response.json()["data"]["water"]["observations"]["data_1day_mean"]

Error handling

Errors are returned in the standard GraphQL errors field. A 200 status does not necessarily indicate a successful query.

payload = response.json()
if "errors" in payload:
    for err in payload["errors"]:
        print(err.get("message"))
    raise RuntimeError("GraphQL query failed")

Iterating over a larger window

A single query returns at most 10 000 rows. Longer time series are split into windows; see Pagination.

from datetime import datetime, timedelta, timezone

start = datetime(2024, 1, 1, tzinfo=timezone.utc)
end = datetime(2026, 1, 1, tzinfo=timezone.utc)
window = timedelta(days=30)

cursor = start
all_rows = []
while cursor < end:
    next_cursor = min(cursor + window, end)
    variables = {
        "from": cursor.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "to": next_cursor.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "station": "2009",
    }
    response = requests.post(
        ENDPOINT,
        json={"query": query, "variables": variables},
        timeout=60,
    )
    response.raise_for_status()
    payload = response.json()
    if "errors" in payload:
        raise RuntimeError(payload["errors"])
    all_rows.extend(payload["data"]["water"]["observations"]["data_1day_mean"])
    cursor = next_cursor