AI translation

This page was translated from German with AI. If in doubt, the German original applies.

Samples

Runnable Python samples that talk to ZEDAL over the REST API only: HTTPS and JSON, no SDK and no gRPC. The code on this page is the development state.

Download

State	Download
Development state (main, 80bb967e)	zedal-api-samples-latest.zip

Prerequisites

• Python 3.9 or later and the dependencies from requirements.txt (only requests).
• Access to the Test environment: provider id, subscriber, account and password. You get them from ZEDAL support; what belongs in the request is described under Getting started.
• An app token for your application, also from ZEDAL support.

Running

In the python/ folder of the extracted package:

pip install -r requirements.txt
python inbox.py

Exit codes: 0 success, 1 an API call failed, 2 usage or configuration error.

Variable	Meaning
ZEDAL_TARGET	Test, Preview, Production or Development; default Test
ZEDAL_PROVIDER_ID	Numeric provider id of the target environment
ZEDAL_APP_TOKEN	App token of your application
ZEDAL_SUBSCRIBER	Subscriber, the actual login
ZEDAL_ACCOUNT	Account below the subscriber; leave it empty if your login has no separate account
ZEDAL_PASSWORD	Password of the account or of the subscriber
ZEDAL_REFRESH_TOKEN_FILE	File for the refresh token; default zedal-refresh-token.txt in the working directory

Subscriber and password are only needed while no valid refresh token is stored.

BashPowerShell

export ZEDAL_TARGET=Test
export ZEDAL_PROVIDER_ID=<provider id>
export ZEDAL_APP_TOKEN=<app token>
export ZEDAL_SUBSCRIBER=<subscriber>
export ZEDAL_PASSWORD=<password>

$env:ZEDAL_TARGET = "Test"
$env:ZEDAL_PROVIDER_ID = "<provider id>"
$env:ZEDAL_APP_TOKEN = "<app token>"
$env:ZEDAL_SUBSCRIBER = "<subscriber>"
$env:ZEDAL_PASSWORD = "<password>"

The samples

Listing unread inbox documents: inbox.py

python inbox.py        # all unread documents in the inbox
python inbox.py 7      # unread documents of the last 7 days

Logs in with the app token and credentials (AuthApi/IssueRefreshToken), exchanges the refresh token for an access token (AuthApi/RequestAccess) and lists the document ids of the unread inbox documents (DocumentApi/ListInbox with unread.value=true, optionally from and to). RequestAccess issues a new refresh token every time and invalidates the one presented, so the sample stores the new one right away. From the second run on, the stored refresh token is enough; if it is no longer valid, the sample logs in again.

Call format, headers and login in detail are described under 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", "development": "dev"}


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