Workflow Orders — Python Example

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

export CELLARIOS_HOST="https://your-instance.cellario.cloud"
export CELLARIOS_API_KEY="<your-api-key>"
export CELLARIOS_USER_ID="3fa85f64-5717-4562-b3fc-2c963f66afa6"

DRY_RUN=1 python walkthrough.py

python walkthrough.py

"""End-to-end walkthrough: discover a workflow, read its parameters, create an order.

This mirrors the five-step flow in README.md (§4.4):

    1. List published workflows and pick one by name.
    2. Read the chosen workflow's definition to get its parameters.
    3. Build arguments by parameter name (values as native JSON types).
    4. POST the Workflow Order to /v1/workflow-orders.
    5. Print the created order's id, user_facing_id and state.

Optionally, start the order (POST .../request-start) when START_ORDER=1.

Configuration is read from environment variables so no secrets live in code:

    CELLARIOS_HOST       Environment host, e.g. https://workflows-test.cellario.cloud
                         (the /api/data-access prefix is added automatically)
    CELLARIOS_API_KEY    API key provided by HighRes (sent as x-api-key)
    CELLARIOS_USER_ID    UUID used for the order metadata fields
    CELLARIOS_WORKFLOW   (optional) workflow name to run; default below
    START_ORDER          (optional) set to 1 to start the order after creating it

Run it:

    export CELLARIOS_HOST="https://workflows-test.cellario.cloud"
    export CELLARIOS_API_KEY="<your-api-key>"
    export CELLARIOS_USER_ID="3fa85f64-5717-4562-b3fc-2c963f66afa6"
    python walkthrough.py

NOTE: step 4 creates a real Workflow Order in the target environment. Set
DRY_RUN=1 to print the request body and stop before submitting.
"""

from __future__ import annotations

import json
import os
import sys

from cellarios_client import CellariosClient, CellariosApiError, build_order

# The workflow to run and the values to set, keyed by parameter *name*.
# Adjust these to match a workflow in your environment. Values are native Python
# types that match each parameter's declared type (STRING/BOOLEAN/INT/ARRAY).
DEFAULT_WORKFLOW_NAME = "TestLoopsAndDecisions"

VALUES_BY_NAME = {
    "loopControl": True,   # BOOLEAN
    "LoopCount": 3,        # INT
}


def require_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        sys.exit(f"Missing required environment variable: {name}")
    return value


def main() -> int:
    host = require_env("CELLARIOS_HOST")
    api_key = require_env("CELLARIOS_API_KEY")
    user_id = require_env("CELLARIOS_USER_ID")
    workflow_name = os.environ.get("CELLARIOS_WORKFLOW", DEFAULT_WORKFLOW_NAME)
    dry_run = os.environ.get("DRY_RUN") == "1"

    client = CellariosClient(host=host, api_key=api_key)

    try:
        # 1. Discover the workflow by name.
        print(f"[1/5] Looking up published workflow '{workflow_name}'...")
        workflow = client.find_workflow_by_name(workflow_name)
        if workflow is None:
            available = sorted(w.get("name", "") for w in client.list_workflows())
            sys.exit(
                f"Workflow '{workflow_name}' not found. "
                f"{len(available)} workflows available; first few: {available[:10]}"
            )
        workflow_id = workflow["id"]
        version = workflow["version"]
        print(f"      -> id={workflow_id} version={version}")

        # 2. Read its parameters (name -> ParameterInfo map).
        print("[2/5] Reading workflow definition parameters...")
        parameter_map = client.get_parameter_map(workflow_id, version)
        for pname, pinfo in parameter_map.items():
            flag = " (required)" if pinfo.required else ""
            print(f"      -> {pname}: type={pinfo.type} id={pinfo.parameter_id}{flag}")

        # 3. Build the order body from parameter names.
        print("[3/5] Building the Workflow Order...")
        order = build_order(
            workflow_id=workflow_id,
            workflow_version=version,
            name=f"{workflow_name} - demo order",
            user_id=user_id,
            parameter_map=parameter_map,
            values_by_name=VALUES_BY_NAME,
            description="Created by walkthrough.py",
            tags=["demo"],
        )
        print(json.dumps(order, indent=2))

        if dry_run:
            print("[4/5] DRY_RUN=1 set — not submitting. Done.")
            return 0

        # 4. Submit the order.
        print("[4/5] Submitting POST /v1/workflow-orders...")
        created = client.create_workflow_order(order)

        # 5. Report the result.
        print("[5/5] Order created:")
        print(f"      id             = {created.get('id')}")
        print(f"      user_facing_id = {created.get('user_facing_id')}")
        print(f"      state          = {created.get('state')}")

        # 6. Optionally start the order (opt-in, so we don't run it by accident).
        #    For workflows with file parameters, upload files first (see README §4.7).
        if os.environ.get("START_ORDER") == "1":
            print("[6/6] Starting order (POST .../request-start)...")
            result = client.start_order(created["id"])
            print(
                f"      {result.get('previous_state')} -> {result.get('state')}"
            )
        else:
            print("      (set START_ORDER=1 to start the order via request-start)")
        return 0

    except CellariosApiError as exc:
        print(f"\nAPI error: {exc}", file=sys.stderr)
        if exc.body:
            print(f"Response body: {exc.body}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

"""A minimal client for creating CellarioOS Workflow Orders over the REST API.

This module wraps the three endpoints needed to go from "which workflows exist?"
to "create an order":

    * list_workflows()            -> GET  {BASE_URL}/v2/workflows/latest-published
    * get_workflow_definition()   -> GET  {BASE_URL}/v2/workflows/id/{id}/version/{version}
    * create_workflow_order()     -> POST {BASE_URL}/v1/workflow-orders

where BASE_URL is your environment host plus the "/api/data-access" prefix, e.g.
"https://workflows-test.cellario.cloud/api/data-access".

Every request is authenticated with an API key sent in the ``x-api-key`` header.
HighRes will provide you with that key; keep it secret (load it from an
environment variable, never commit it).

Verified against the live API in July 2026. Key facts baked into this client:
  * Discovery is on /v2; Workflow Orders are on /v1 (there is no /v2/workflow-orders).
  * A parameter's order id is its ``parameter_id`` (NOT the top-level ``id``);
    its name/type live in ``parameter_type_descriptor``.
  * An argument ``value`` is a native JSON type matching the parameter's type
    (string / boolean / number / array), not a stringified value.

See README.md for the full API documentation. Only ``requests`` is required.
"""

from __future__ import annotations

import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any

import requests


# ---------------------------------------------------------------------------
# Configuration & client
# ---------------------------------------------------------------------------

# The Data Access API is mounted under this path on the environment host.
API_PREFIX = "/api/data-access"


class CellariosApiError(RuntimeError):
    """Raised when the CellarioOS API returns a non-success HTTP status."""

    def __init__(self, message: str, *, status_code: int | None = None, body: str | None = None):
        super().__init__(message)
        self.status_code = status_code
        self.body = body


@dataclass
class CellariosClient:
    """Thin HTTP client for the CellarioOS Data Access API.

    Args:
        host: Environment host, e.g. ``https://workflows-test.cellario.cloud``.
            The ``/api/data-access`` prefix is added automatically unless the host
            you pass already includes it.
        api_key: The API key provided by HighRes, sent as ``x-api-key``.
        timeout: Per-request timeout in seconds.
    """

    host: str
    api_key: str
    timeout: float = 30.0
    session: requests.Session = field(default_factory=requests.Session)

    def __post_init__(self) -> None:
        base = self.host.rstrip("/")
        if not base.endswith(API_PREFIX):
            base = base + API_PREFIX
        self.base_url = base
        # Attach auth + JSON headers once for every request made by this client.
        self.session.headers.update(
            {
                "x-api-key": self.api_key,
                "Accept": "application/json",
            }
        )

    # -- internal helpers ---------------------------------------------------

    def _url(self, path: str) -> str:
        return f"{self.base_url}/{path.lstrip('/')}"

    def _request(self, method: str, path: str, *, json_body: Any | None = None) -> Any:
        url = self._url(path)
        response = self.session.request(
            method,
            url,
            json=json_body,
            timeout=self.timeout,
        )
        if not response.ok:
            raise CellariosApiError(
                f"{method} {url} failed with HTTP {response.status_code}",
                status_code=response.status_code,
                body=response.text,
            )
        if not response.content:
            return None
        return response.json()

    # -- discovery ----------------------------------------------------------

    def list_workflows(self) -> list[dict[str, Any]]:
        """Return the latest published version of every Workflow definition.

        GET /v2/workflows/latest-published

        Each item includes ``id``, ``name``, ``version``, ``status`` and more.
        Use ``id`` and ``version`` to fetch the full definition and to create an
        order.
        """
        return self._request("GET", "/v2/workflows/latest-published")

    def find_workflow_by_name(self, name: str) -> dict[str, Any] | None:
        """Return the published workflow summary whose ``name`` matches (case-insensitive)."""
        target = name.strip().lower()
        for workflow in self.list_workflows():
            if str(workflow.get("name", "")).strip().lower() == target:
                return workflow
        return None

    def get_workflow_definition(self, workflow_id: str, version: str) -> dict[str, Any]:
        """Return the full definition for one workflow version, including ``parameters``.

        GET /v2/workflows/id/{id}/version/{version}
        """
        return self._request(
            "GET", f"/v2/workflows/id/{workflow_id}/version/{version}"
        )

    def get_parameters(self, workflow_id: str, version: str) -> list["ParameterInfo"]:
        """Return a list of :class:`ParameterInfo` for a workflow version.

        Flattens each raw parameter (``parameter_id`` + ``parameter_type_descriptor``)
        into a simple object exposing ``name``, ``type``, ``parameter_id``,
        ``required`` and ``default_value``.
        """
        definition = self.get_workflow_definition(workflow_id, version)
        return [ParameterInfo.from_raw(p) for p in definition.get("parameters", [])]

    def get_parameter_map(self, workflow_id: str, version: str) -> dict[str, "ParameterInfo"]:
        """Return a ``{parameter_name: ParameterInfo}`` map for a workflow version.

        Convenience wrapper over :meth:`get_parameters` so callers can build
        arguments by parameter *name* instead of copying UUIDs by hand.
        """
        return {p.name: p for p in self.get_parameters(workflow_id, version)}

    def file_parameters(self, workflow_id: str, version: str) -> list["ParameterInfo"]:
        """Return the file parameters of a workflow version.

        A file parameter is identified by ``type == "STRING"`` and
        ``validation.format == "FILE"`` (see :attr:`ParameterInfo.is_file`). These
        are populated by uploading a file to the order (see
        :meth:`upload_order_file`), not via order arguments.
        """
        return [p for p in self.get_parameters(workflow_id, version) if p.is_file]

    # -- order creation -----------------------------------------------------

    def create_workflow_order(self, order: dict[str, Any]) -> dict[str, Any]:
        """Create a Workflow Order.

        POST /v1/workflow-orders

        Args:
            order: A CreateWorkflowOrderRequest body. Use :func:`build_order`
                to construct one safely.

        Returns:
            The created order, including ``id``, ``user_facing_id`` and ``state``.
        """
        return self._request("POST", "/v1/workflow-orders", json_body=order)

    def list_workflow_orders(self) -> list[dict[str, Any]]:
        """List Workflow Orders. GET /v1/workflow-orders"""
        return self._request("GET", "/v1/workflow-orders")

    def start_order(self, order_id: str) -> dict[str, Any]:
        """Request that a created order start running.

        POST /v1/workflow-orders/{order_id}/request-start

        A newly created order stays ``PENDING`` until you start it. For workflows
        with file parameters, upload all files (see :meth:`upload_order_file`)
        BEFORE calling this. Takes no request body.

        Returns:
            A WorkflowOrderStateResponse dict with ``order_id``, ``previous_state``
            and ``state`` (e.g. ``{"previous_state": "PENDING", "state": "STARTING"}``).
        """
        return self._request("POST", f"/v1/workflow-orders/{order_id}/request-start")

    def cancel_order(self, order_id: str, reason: str | None = None) -> Any:
        """Request that a pending or running order be cancelled.

        POST /v1/workflow-orders/{order_id}/request-cancel

        Args:
            order_id: The ``id`` of the order to cancel.
            reason: Optional free-text reason recorded with the cancellation.

        Returns:
            None — the endpoint responds with 200 OK and no body on success.
        """
        body = {"cancellation_reason": reason} if reason is not None else {}
        return self._request(
            "POST", f"/v1/workflow-orders/{order_id}/request-cancel", json_body=body
        )

    def upload_order_file(
        self,
        order_id: str,
        parameter_id: str,
        file_path: str,
        *,
        content_type: str | None = None,
        field_name: str = "file",
    ) -> Any:
        """Upload a file for a ``FILE`` parameter of an existing order.

        POST /v1/workflow-orders/{order_id}/files?parameter_id={parameter_id}

        This must be done AFTER creating the order and BEFORE starting it. Call
        once per ``FILE`` parameter, each with that parameter's ``parameter_id``.

        Args:
            order_id: The ``id`` of the created order.
            parameter_id: The ``parameter_id`` of the FILE parameter.
            file_path: Path to the local file to upload.
            content_type: Optional MIME type for the file part.
            field_name: Multipart form field name for the file part. Defaults to
                ``"file"``; change it if your environment expects a different name.

        Returns:
            The parsed JSON response, or None if the response has no body.
        """
        url = self._url(f"/v1/workflow-orders/{order_id}/files")
        filename = os.path.basename(file_path)
        with open(file_path, "rb") as handle:
            file_part = (filename, handle, content_type) if content_type else (filename, handle)
            # Do not set Content-Type manually: requests adds the multipart boundary.
            response = self.session.post(
                url,
                params={"parameter_id": parameter_id},
                files={field_name: file_part},
                timeout=self.timeout,
            )
        if not response.ok:
            raise CellariosApiError(
                f"POST {url} (parameter_id={parameter_id}) failed with HTTP {response.status_code}",
                status_code=response.status_code,
                body=response.text,
            )
        return response.json() if response.content else None


# ---------------------------------------------------------------------------
# Parameter model
# ---------------------------------------------------------------------------


@dataclass
class ParameterInfo:
    """A flattened view of one workflow parameter.

    Attributes:
        name: Parameter name (from ``parameter_type_descriptor.name``).
        type: Base value type (``TaskParameterType``): ``STRING``, ``INT``,
            ``FLOAT``, ``BOOLEAN``, ``ARRAY``, ``OBJECT`` or ``DICTIONARY``.
        parameter_id: The id to send as ``workflow_parameter_id`` in an order.
        required: Whether an argument is mandatory.
        default_value: The default value string, if any.
        format: The validation format (``parameter_type_descriptor.validation.format``,
            i.e. ``ParameterDescriptorFormat``): ``NONE``, ``DATE``, ``TIME``,
            ``DATE_TIME``, ``FILE`` or ``URL``. It refines a ``STRING`` into a more
            specific type — see :attr:`resolved_type`.
    """

    name: str
    type: str
    parameter_id: str
    required: bool = False
    default_value: str | None = None
    format: str | None = None

    @property
    def is_file(self) -> bool:
        """True if this is a file parameter (``type`` STRING with ``format`` FILE)."""
        return self.type == "STRING" and self.format == "FILE"

    @property
    def resolved_type(self) -> str:
        """The effective parameter type, combining ``type`` and ``format``.

        Mirrors the platform's ``ResolvedParameterType`` mapping. A ``STRING``
        with a ``DATE``/``TIME``/``DATE_TIME``/``FILE``/``URL`` format resolves to
        that format; ``INT``/``FLOAT``/``BOOLEAN`` pass through; anything else
        resolves to ``STRING``. Returns one of: ``STRING``, ``INT``, ``FLOAT``,
        ``BOOLEAN``, ``DATE``, ``TIME``, ``DATE_TIME``, ``FILE``, ``URL``.
        """
        if self.type == "STRING":
            if self.format in ("DATE", "TIME", "DATE_TIME", "FILE", "URL"):
                return self.format
            return "STRING"
        if self.type in ("INT", "FLOAT", "BOOLEAN"):
            return self.type
        return "STRING"

    @classmethod
    def from_raw(cls, raw: dict[str, Any]) -> "ParameterInfo":
        descriptor = raw.get("parameter_type_descriptor", {}) or {}
        validation = descriptor.get("validation", {}) or {}
        return cls(
            name=descriptor.get("name", ""),
            type=descriptor.get("type", ""),
            parameter_id=raw["parameter_id"],
            required=bool(descriptor.get("required", False)),
            default_value=descriptor.get("default_value"),
            format=validation.get("format"),
        )


# ---------------------------------------------------------------------------
# Payload builders
# ---------------------------------------------------------------------------


def build_metadata(user_id: str, *, tags: list[str] | None = None) -> dict[str, Any]:
    """Build the ``metadata`` block, stamping created/updated timestamps as now (UTC)."""
    now = datetime.now(timezone.utc).isoformat()
    return {
        "created_by_user_id": user_id,
        "created_at": now,
        "last_updated_by_user_id": user_id,
        "last_update": now,
        "assigned_to_user_id": user_id,
        "tags": tags or [],
    }


def build_order(
    *,
    workflow_id: str,
    workflow_version: str,
    name: str,
    user_id: str,
    parameter_map: dict[str, ParameterInfo],
    values_by_name: dict[str, Any],
    description: str = "",
    tags: list[str] | None = None,
) -> dict[str, Any]:
    """Assemble a CreateWorkflowOrderRequest body from parameter *names*.

    Values are sent as native JSON types (string/boolean/number/array). Match the
    Python value to the parameter's declared type — e.g. pass ``True`` for a
    ``BOOLEAN`` parameter, ``3`` for an ``INT``, a list for an ``ARRAY``.

    Args:
        workflow_id: Workflow definition ID (from list_workflows).
        workflow_version: Version to run (from list_workflows).
        name: A name for this order.
        user_id: UUID used for the metadata user fields.
        parameter_map: ``{parameter_name: ParameterInfo}`` (from get_parameter_map).
        values_by_name: ``{parameter_name: value}`` you want to set.
        description: Optional order description.
        tags: Optional metadata tags.

    Raises:
        KeyError: If a name in ``values_by_name`` is not a parameter of the workflow.
        ValueError: If a parameter marked ``required`` in the definition has no
            value supplied (missing from ``values_by_name``, or supplied as
            ``None``) *and* has no ``default_value`` to fall back on.
    """
    arguments = []
    for param_name, value in values_by_name.items():
        if param_name not in parameter_map:
            raise KeyError(
                f"'{param_name}' is not a parameter of this workflow. "
                f"Available: {sorted(parameter_map)}"
            )
        arguments.append(
            {
                "workflow_parameter_id": parameter_map[param_name].parameter_id,
                "value": value,  # native JSON type — matches the parameter's type
            }
        )

    # Every required parameter must have a value the server can use: either one
    # supplied here, or a default_value in the definition to fall back on. A
    # missing key or an explicit None both count as "not supplied"; a parameter
    # with a non-empty default_value is satisfied even if omitted.
    missing_required = sorted(
        name
        for name, info in parameter_map.items()
        if info.required
        and values_by_name.get(name) is None
        and not info.default_value
    )
    if missing_required:
        raise ValueError(
            "Missing values for required parameter(s) with no default: "
            f"{missing_required}. Supply a value for each in values_by_name."
        )

    return {
        "workflow_id": workflow_id,
        "workflow_version": workflow_version,
        "name": name,
        "description": description,
        "arguments": arguments,
        "metadata": build_metadata(user_id, tags=tags),
    }

requests>=2.31,<3