Python

L'interface GraphQL est appelée via des requêtes POST avec un corps JSON. Aucune authentification n'est requise.

Requête unique

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"])

Requête avec variables

Les variables séparent les valeurs du texte de la requête et évitent les problèmes d'échappement.

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"]

Gestion des erreurs

Les erreurs sont renvoyées dans le champ GraphQL standard errors. Un statut 200 n'indique pas nécessairement une requête réussie.

payload = response.json()
if "errors" in payload:
    for err in payload["errors"]:
        print(err.get("message"))
    raise RuntimeError("La requête GraphQL a échoué")

Itération sur des fenêtres plus larges

Une requête renvoie au maximum 10 000 enregistrements. Les séries temporelles plus longues sont découpées en fenêtres ; voir 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