User API example

This example explains how user-api.py authenticates and runs a basic workflow. It is a starting point for automating dashboard workflows from Python.

Sample files

These files live next to this document:

• Script: docs/5-examples/1-user-api/user-api.py
• Operation YAML: docs/5-examples/1-user-api/operation-example.yaml
• File to uplink: docs/5-examples/1-user-api/file-to-uplink.txt

Prerequisites

• A valid username/password pair for your user
• The backend reachable via BASE_URL
• Python 3.10+ with requests, pyyaml, and python-dotenv
• A file to be uplinked called file-to-uplink.txt
• An operation file operation-example.yaml

python3 -m venv venv
source venv/bin/activate
pip3 install requests pyyaml dotenv

Authentication

The script authenticates with username/password and exchanges them for a short-lived access token. It then calls /api/v1/users/me/tenant to resolve the user's default tenant and uses that value for subsequent tenant-scoped requests. If the token expires, it re-authenticates and retries the request.

A tenant is the DPhi account boundary used to scope resources such as namespaces, operations, executions, and files. Full multi-tenant support in the public examples will come later, so for now you can mostly ignore it and rely on the script resolving your default tenant automatically.

Environment variables

The script reads the following environment variables from a local .env file or your shell:

• BASE_URL (required)
• USERNAME (required)
• PASSWORD (required)
• OPERATION_FILE (required)
• SCHEDULE_DATE (optional)

Usage

export BASE_URL="https://cg2.dphispace.com"
export USER="<your username>"
export PASSWORD="<your password>"
export OPERATION_FILE="/path/to/operation-example.yaml"

python3 user-api.py

Optional schedule date (query param). If omitted, the script uses the current UTC time:

export SCHEDULE_DATE="2026-03-20T10:00:00Z"
python user-api.py

The resolved tenant is cached in memory for the duration of the script run.

Operation template

The script loads the YAML file below and reads id, namespace, and system from it:

id: example-op
namespace: test
system: cg2.em
description: Example operation
labels:
  name: Business Automation
tasks:
  - id: uplink_example
    type: dphi.space.cg2.uplink
    description: Upload the local input file to the payload volume
    source:
      - file-to-uplink.txt
    destination: /example
    volume: payload
  - id: pod_run_job
    pod_name: test
    type: dphi.space.cg2.pod.run
    description: Read the uplinked file and generate a processed output file
    image:
      name: alpine
      tag: latest
    command: ["/bin/sh"]
    args:
      [
        "-c",
        "printf 'Input file contents:\\n' > /data/example/processed-from-uplink.txt && cat /data/example/file-to-uplink.txt >> /data/example/processed-from-uplink.txt && printf '\\n\\nProcessed by pod.\\n' >> /data/example/processed-from-uplink.txt",
      ]
    env:
      - name: LOG_LEVEL
        value: info
    ports:
      - container: 8080
        host: 8080
        protocol: tcp
    node: MPU
    volume: payload
    max_duration: 1
  - id: downlink_results
    type: dphi.space.cg2.downlink
    description: Download the processed file generated by the pod
    source:
      - /example/processed-from-uplink.txt
    destination: /
    volume: payload

Create a file with the content above, named operation-example.yaml.

What it does

At a high level, the script:

• Authenticates with the DPhi API.
• Resolves the authenticated user's default tenant.
• Ensures the namespace exists.
• Submits the operation YAML.
• Triggers an execution and waits for outputs.
• Downloads the resulting file.

Optional helpers

The script also includes helper functions for namespace file uploads and downloads:

• upload_namespace_file(namespace, file_path, dest_path=None)
• download_namespace_file(namespace, remote_path, output_path, revision=None)

You can call these helpers before creating an execution (to ensure input files exist) or after an execution completes (to retrieve results).

API reference

The full API definition can be downloaded at BASE_URL/api/specs. You can load that YAML into Swagger or any OpenAPI tool to explore the endpoints.

Full script

#!/usr/bin/env python3

import json
import time
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

import requests
import yaml
from dotenv import load_dotenv

load_dotenv()

BASE_URL = os.environ.get("BASE_URL").rstrip("/")
SCHEDULE_DATE = os.environ.get("SCHEDULE_DATE")
OPERATION_FILE = os.environ.get("OPERATION_FILE")

USERNAME = os.environ.get("USERNAME")
PASSWORD = os.environ.get("PASSWORD")
TOKEN = None
TENANT = None
FILE_TO_UPLINK = "file-to-uplink.txt"
# comes from the yaml operation file
FILE_TO_DOWNLINK = "processed-from-uplink.txt"


def pretty_print_json(payload):
    print(json.dumps(payload, indent=2, sort_keys=True))


def safe_json(response):
    content_type = response.headers.get("Content-Type", "")
    if "json" not in content_type.lower():
        return None
    try:
        if not response.content:
            return None
    except RuntimeError:
        return None
    try:
        return response.json()
    except ValueError:
        return None


def show_response(label, response, payload=None):
    print(f"\n{label}: {response.status_code}")
    if payload is None:
        payload = safe_json(response)
    if payload is not None:
        pretty_print_json(payload)
    else:
        content_type = response.headers.get("Content-Type", "")
        if content_type.startswith("text/"):
            try:
                print(response.text)
            except RuntimeError:
                print("(response body already consumed)")
        else:
            summary = content_type or "unknown"
            print(f"(non-JSON response body omitted; Content-Type: {summary})")


def load_operation_yaml(path):
    operation_path = Path(path)
    if not operation_path.exists():
        raise FileNotFoundError(f"Operation file not found: {operation_path}")
    content = operation_path.read_text(encoding="utf-8")
    payload = yaml.safe_load(content)
    if not isinstance(payload, dict):
        raise ValueError("Operation file must be a YAML object")
    for key in ("id", "namespace", "system"):
        if not payload.get(key):
            raise ValueError(f"Operation file missing required field: {key}")
    return content, payload


def get_token():
    global TOKEN
    if not USERNAME or not PASSWORD:
        print("USERNAME and PASSWORD are required")
        sys.exit(1)

    response = requests.post(
        f"{BASE_URL}/api/v1/auth/",
        data={"username": USERNAME, "password": PASSWORD},
        timeout=20,
    )
    content = response.json() if response.content else {}
    if response.status_code == 200:
        TOKEN = content.get("access")
        if not TOKEN:
            raise RuntimeError("No access token in response")
        print("Access token acquired")
        return True

    print(f"Failed to get access token: {content.get('detail')}")
    return False


def ensure_token():
    if TOKEN is None:
        if not get_token():
            raise RuntimeError("Authentication failed")


def get_default_tenant():
    global TENANT

    response = authorized_get(f"{BASE_URL}/api/v1/users/me/tenant", timeout=20)
    payload = safe_json(response)
    if response.status_code != 200:
        detail = None
        if isinstance(payload, dict):
            detail = payload.get("detail")
        message = detail or f"Unexpected status {response.status_code}"
        raise RuntimeError(f"Failed to resolve default tenant: {message}")

    if not isinstance(payload, dict):
        raise RuntimeError("Failed to resolve default tenant: invalid JSON response")

    tenant_name = payload.get("name")
    if not tenant_name:
        raise RuntimeError("Failed to resolve default tenant: missing tenant name")

    TENANT = str(tenant_name)
    return TENANT


def get_tenant():
    if TENANT is None:
        return get_default_tenant()
    return TENANT


def authorized_get(url, **kwargs):
    ensure_token()
    headers = kwargs.pop("headers", {})
    headers["Authorization"] = f"Bearer {TOKEN}"
    response = requests.get(url, headers=headers, **kwargs)
    if response.status_code == 401:
        if get_token():
            headers["Authorization"] = f"Bearer {TOKEN}"
            response = requests.get(url, headers=headers, **kwargs)
    return response


def authorized_post(url, **kwargs):
    ensure_token()
    headers = kwargs.pop("headers", {})
    headers["Authorization"] = f"Bearer {TOKEN}"
    response = requests.post(url, headers=headers, **kwargs)
    if response.status_code == 401:
        if get_token():
            headers["Authorization"] = f"Bearer {TOKEN}"
            response = requests.post(url, headers=headers, **kwargs)
    return response


def authorized_put(url, **kwargs):
    ensure_token()
    headers = kwargs.pop("headers", {})
    headers["Authorization"] = f"Bearer {TOKEN}"
    response = requests.put(url, headers=headers, **kwargs)
    if response.status_code == 401:
        if get_token():
            headers["Authorization"] = f"Bearer {TOKEN}"
            response = requests.put(url, headers=headers, **kwargs)
    return response


def upload_namespace_file(namespace, file_path, dest_path=None):
    file_path = Path(file_path)
    if not file_path.exists():
        raise FileNotFoundError(f"File not found: {file_path}")

    params = {}
    if dest_path:
        params["path"] = dest_path

    tenant = get_tenant()
    with file_path.open("rb") as handle:
        response = authorized_post(
            f"{BASE_URL}/api/v1/{tenant}/namespaces/{namespace}/files",
            params=params,
            files={"file": (file_path.name, handle)},
            timeout=60,
        )

    return response


def download_namespace_file(namespace, remote_path, output_path, revision=None):
    tenant = get_tenant()
    params = {"path": remote_path}
    if revision is not None:
        params["revision"] = revision

    response = authorized_get(
        f"{BASE_URL}/api/v1/{tenant}/namespaces/{namespace}/files",
        params=params,
        timeout=60,
        stream=True,
    )

    if response.status_code != 200:
        return response

    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    with output_path.open("wb") as handle:
        for chunk in response.iter_content(chunk_size=1024 * 1024):
            if chunk:
                handle.write(chunk)

    return response


def list_namespace_directory(namespace, path="/"):
    tenant = get_tenant()
    params = {"path": path}
    response = authorized_get(
        f"{BASE_URL}/api/v1/{tenant}/namespaces/{namespace}/files/directory",
        params=params,
        timeout=60,
    )

    return response


def main():
    operation_yaml, operation_payload = load_operation_yaml(OPERATION_FILE)
    op_id = str(operation_payload.get("id"))
    namespace = str(operation_payload.get("namespace"))
    system = str(operation_payload.get("system"))
    tenant = get_tenant()
    print(f"Resolved tenant: {tenant}")

    # 1) Create namespace (safe to ignore if it already exists)
    create_namespace = authorized_post(
        f"{BASE_URL}/api/v1/{tenant}/namespaces",
        headers={"Content-Type": "application/json"},
        json={"id": namespace, "system": system},
        timeout=30,
    )
    show_response("Create namespace", create_namespace)
    time.sleep(5)

    # 0) Upload files to be uplinked
    file_upload = upload_namespace_file(namespace, FILE_TO_UPLINK)
    show_response("Uploading File", file_upload)
    time.sleep(5)

    # 2) Create operation (YAML file)
    # Only necessary to be executed once
    create_operation = authorized_post(
        f"{BASE_URL}/api/v1/{tenant}/operations",
        headers={"Content-Type": "application/x-yaml"},
        data=operation_yaml,
        timeout=30,
    )
    show_response("Create operation", create_operation)
    if create_operation.status_code != 201:
        update_operation = authorized_put(
            f"{BASE_URL}/api/v1/{tenant}/operations/{namespace}/{op_id}",
            headers={"Content-Type": "application/x-yaml"},
            data=operation_yaml,
            timeout=30,
        )
        show_response("Update operation", update_operation)
    time.sleep(5)

    # 3) List operations
    list_operations = authorized_get(
        f"{BASE_URL}/api/v1/{tenant}/operations/search",
        params={"namespace": namespace, "system": system},
        timeout=30,
    )
    show_response("List operations", list_operations)
    time.sleep(5)

    # 4) Create execution
    schedule_value = (SCHEDULE_DATE or "").strip()
    if not schedule_value:
        schedule_value = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
    exec_params = {"scheduleDate": schedule_value}

    create_execution = authorized_post(
        f"{BASE_URL}/api/v1/{tenant}/executions/{namespace}/{op_id}",
        params=exec_params,
        timeout=30,
    )
    create_execution_payload = safe_json(create_execution) or {}
    show_response(
        "Create execution", create_execution, payload=create_execution_payload
    )
    execution_id = create_execution_payload.get("id")
    if not execution_id:
        print("No execution id returned; listing all executions.")

    time.sleep(5)
    # 5) List executions
    list_executions = authorized_get(
        f"{BASE_URL}/api/v1/{tenant}/executions",
        params={"system": system},
        timeout=30,
    )
    list_executions_payload = safe_json(list_executions)
    if execution_id and list_executions_payload is not None:
        if (
            isinstance(list_executions_payload, dict)
            and "results" in list_executions_payload
        ):
            results = list_executions_payload.get("results") or []
            filtered_results = [
                item for item in results if item.get("id") == execution_id
            ]
            list_executions_payload = {
                **list_executions_payload,
                "count": len(filtered_results),
                "next": None,
                "previous": None,
                "results": filtered_results,
            }
        elif isinstance(list_executions_payload, list):
            list_executions_payload = [
                item
                for item in list_executions_payload
                if item.get("id") == execution_id
            ]

    show_response("List executions", list_executions, payload=list_executions_payload)
    time.sleep(5)

    # Monitor files until the downlink file is available
    found_file = False
    while True:
        files = list_namespace_directory(namespace)
        if files.status_code != 200:
            show_response("List files", files)
            time.sleep(20)
            continue
        for file in files.json():
            if FILE_TO_DOWNLINK in file.get("path", ""):
                print("Found file")
                found_file = True
                break
        if found_file:
            break
        print("Waiting for file to be downlinked...")
        time.sleep(20)

    # Download file
    download_file = download_namespace_file(
        namespace, FILE_TO_DOWNLINK, "downlinked.txt"
    )
    show_response("Download File", download_file)


if __name__ == "__main__":
    main()