Beispiele¶
Lauffähige Python-Beispiele, die ZEDAL nur über die REST-API ansprechen: HTTPS und JSON, ohne SDK und ohne gRPC. Der Code auf dieser Seite ist der Entwicklungsstand.
Herunterladen¶
| Stand | Download |
|---|---|
Entwicklungsstand (main, 0b48d4dd) |
zedal-api-samples-latest.zip |
Voraussetzungen¶
- Python 3.9 oder neuer und die Abhängigkeiten aus
requirements.txt(nurrequests). - Zugang zur Test-Umgebung: Provider-Id, Subscriber, Account und Passwort. Du bekommst sie vom ZEDAL-Support; was in die Anfrage gehört, steht unter Erste Schritte.
- Ein App-Token für deine Anwendung, ebenfalls vom ZEDAL-Support.
Ausführen¶
Im Ordner python/ des entpackten Pakets:
Exit-Codes: 0 Erfolg, 1 ein API-Aufruf ist fehlgeschlagen, 2 Aufruf- oder Konfigurationsfehler.
| Variable | Bedeutung |
|---|---|
ZEDAL_TARGET |
Test, Preview oder Production; Standard Test |
ZEDAL_PROVIDER_ID |
Numerische Provider-Id der Zielumgebung |
ZEDAL_APP_TOKEN |
App-Token deiner Anwendung |
ZEDAL_SUBSCRIBER |
Subscriber, der eigentliche Login |
ZEDAL_ACCOUNT |
Account unter dem Subscriber; leer lassen, wenn dein Login keinen eigenen Account hat |
ZEDAL_PASSWORD |
Passwort des Accounts bzw. des Subscribers |
ZEDAL_REFRESH_TOKEN_FILE |
Datei für das Refresh-Token; Standard zedal-refresh-token.txt im Arbeitsverzeichnis |
Subscriber und Passwort braucht das Beispiel nur, solange kein gültiges Refresh-Token gespeichert ist.
Die Beispiele¶
Ungelesene Eingänge auflisten: inbox.py¶
python inbox.py # all unread documents in the inbox
python inbox.py 7 # unread documents of the last 7 days
Meldet sich mit App-Token und Zugangsdaten an (AuthApi/IssueRefreshToken), tauscht das Refresh-Token gegen ein
Access-Token (AuthApi/RequestAccess) und listet die Dokument-Ids der ungelesenen Eingänge
(DocumentApi/ListInbox mit unread.value=true, optional from und to). RequestAccess gibt jedes Mal ein neues
Refresh-Token aus und macht das vorgelegte ungültig; das Beispiel speichert das neue deshalb sofort. Ab dem zweiten
Lauf reicht das gespeicherte Refresh-Token, ist es nicht mehr gültig, meldet sich das Beispiel neu an.
Aufrufform, Header und Anmeldung im Einzelnen stehen unter REST / JSON.
inbox.py
#!/usr/bin/env python3
"""List the unread documents in your ZEDAL inbox, using the REST API only.
No SDK and no gRPC: every call is a plain HTTPS request with JSON against the OpenAPI
facade of the platform (https://openapi.<environment>.zedal.de, specification at
/openapi.yaml). The flow:
1. Log in. AuthApi/IssueRefreshToken exchanges your credentials and the app token for a
long-lived refresh token. Later runs use the stored refresh token instead, so the
password is only needed for the first run.
2. AuthApi/RequestAccess exchanges the refresh token for a short-lived access token and a
NEW refresh token. The old one is invalid from that moment on, so the new one is stored
right away.
3. DocumentApi/ListInbox with the access token.
Usage:
python inbox.py all unread documents in the inbox
python inbox.py 7 unread documents of the last 7 days
Exit codes: 0 success, 1 the API call failed, 2 usage or configuration error.
"""
import datetime
import os
import sys
from pathlib import Path
import requests
# ZEDAL_TARGET uses the names of the .NET samples; the REST host uses the short form.
HOSTS = {"test": "test", "preview": "preview", "production": "live"}
class ConfigError(Exception):
pass
class ApiError(Exception):
def __init__(self, operation, status, message):
super().__init__(f"{operation} failed: HTTP {status}, {message}")
self.status = status
def setting(name, default=None):
value = os.environ.get(name, default)
if value is None:
raise ConfigError(f"environment variable {name} is not set")
return value
def target():
name = setting("ZEDAL_TARGET", "Test").lower()
if name not in HOSTS:
raise ConfigError(f"ZEDAL_TARGET must be one of {', '.join(HOSTS)}, not '{name}'")
return name
TOKEN_FILE = Path(os.environ.get("ZEDAL_REFRESH_TOKEN_FILE", "zedal-refresh-token.txt"))
def call(http_method, operation, token=None, body=None, params=None):
"""One REST call. The path of every operation is its gRPC method path,
/<package>.<Service>/<Method>; the specification lists them all."""
headers = {
# Selects the provider's backend. The header name has an underscore; without it,
# or spelled provider-id, the call fails with HTTP 501. Only the discovery
# (DiscoveryApi/ListProvider) works without it.
"provider_id": setting("ZEDAL_PROVIDER_ID"),
}
if token:
headers["Authorization"] = f"Bearer {token}"
base_url = f"https://openapi.{HOSTS[target()]}.zedal.de"
response = requests.request(http_method, f"{base_url}/infotech.zedal.{operation}",
headers=headers, json=body, params=params, timeout=30)
if response.status_code != 200:
# Errors carry the gRPC status code and a message: {"code": 16, "message": "..."}.
# Unknown fields in a request body are rejected with HTTP 400 instead of ignored.
try:
message = response.json().get("message")
except ValueError:
message = response.text
if response.status_code == 501:
message += " (is ZEDAL_PROVIDER_ID a provider of this environment?)"
raise ApiError(operation, response.status_code, message)
# Field names are lowerCamelCase, 64-bit numbers arrive as strings, and fields with a
# default value (0, "", false, empty list) are left out of the response.
return response.json()
def log_in():
"""First login with credentials. Returns a refresh token."""
response = call("POST", "core.v1.api.AuthApi/IssueRefreshToken", body={
"appToken": setting("ZEDAL_APP_TOKEN"),
"usernameCredentials": {
"subscriberId": setting("ZEDAL_SUBSCRIBER"),
# Leave ZEDAL_ACCOUNT empty if your login has no separate account.
"accountId": setting("ZEDAL_ACCOUNT", ""),
"password": setting("ZEDAL_PASSWORD"),
},
})
return response["refreshToken"]
def save_refresh_token(token):
# A plain file keeps the sample simple. In production store the refresh token like a
# password: OS key store, secrets vault or an encrypted database column.
fd = os.open(TOKEN_FILE, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "w") as file:
file.write(token)
def request_access(refresh_token):
"""Exchanges the refresh token for an access token. Rotates the refresh token."""
# The refresh token goes into the Authorization header AND the body.
response = call("POST", "core.v1.api.AuthApi/RequestAccess", token=refresh_token,
body={"refreshToken": refresh_token})
# Store the new refresh token before doing anything else: the presented one is used
# up, and if the new one is lost, the next run has to log in with the password again.
save_refresh_token(response["newRefreshToken"])
return response["accessToken"]
def access_token():
if TOKEN_FILE.exists():
try:
return request_access(TOKEN_FILE.read_text().strip())
except ApiError as error:
if error.status != 401:
raise
# Expired, revoked or already rotated by another process: log in again.
print(f"Stored refresh token in {TOKEN_FILE} is no longer valid, logging in again.")
return request_access(log_in())
def list_inbox(token, days=None):
# GET operations take their parameters from the query string. Nested fields use dot
# notation (unread.value), timestamps are RFC 3339 in UTC.
params = {"unread.value": "true"}
if days is not None:
now = datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0)
params["from"] = (now - datetime.timedelta(days=days)).isoformat().replace("+00:00", "Z")
params["to"] = now.isoformat().replace("+00:00", "Z")
response = call("GET", "core.v1.api.DocumentApi/ListInbox", token=token, params=params)
return [doc_id["id"] for doc_id in response.get("docIds", [])]
def main(args):
if len(args) > 1 or (args and not args[0].isdigit()):
print(__doc__)
return 2
days = int(args[0]) if args else None
try:
environment = target()
token = access_token()
documents = list_inbox(token, days)
except ConfigError as error:
print(f"Configuration error: {error}", file=sys.stderr)
return 2
except (ApiError, requests.RequestException) as error:
print(error, file=sys.stderr)
return 1
period = f" of the last {days} days" if days is not None else ""
print(f"{len(documents)} unread document(s) in the inbox{period} ({environment}):")
for document_id in documents:
print(f" {document_id}")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))