Using the API from Python

This example uses only the Python standard library (urllib.request and json), so there is nothing to install. It needs Python 3.8 or later, an API token, and the COCOCO_URL and COCOCO_TOKEN environment variables described in Calling the CoCoCo API.

The operations

query ListJobs($first: Int, $after: String) {
  listJobs(first: $first, after: $after) {
    totalCount
    pageInfo { hasNextPage endCursor }
    edges { node { id name quantity status dueAt } }
  }
}

mutation CreateJob($input: CreateJobInput!) {
  createJob(input: $input) {
    job { id name status }
    errors { path message }
  }
}

The program

Save this as jobs.py. python3 jobs.py lists every job; python3 jobs.py create creates one.

import json
import os
import sys
import urllib.request

URL = os.environ["COCOCO_URL"].rstrip("/") + "/graphql"
TOKEN = os.environ["COCOCO_TOKEN"]

LIST_JOBS = """query ListJobs($first: Int, $after: String) {
  listJobs(first: $first, after: $after) {
    totalCount
    pageInfo { hasNextPage endCursor }
    edges { node { id name quantity status dueAt } }
  }
}"""

CREATE_JOB = """mutation CreateJob($input: CreateJobInput!) {
  createJob(input: $input) {
    job { id name status }
    errors { path message }
  }
}"""


def graphql(query, variables=None):
    """POST one operation and return its data. Raises on any top-level error."""
    request = urllib.request.Request(
        URL,
        data=json.dumps({"query": query, "variables": variables or {}}).encode(),
        headers={
            "Authorization": f"Bearer {TOKEN}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    # urlopen raises urllib.error.HTTPError for 4xx/5xx responses.
    with urllib.request.urlopen(request, timeout=30) as response:
        result = json.load(response)
    # Authentication, permission and query errors arrive here with HTTP 200.
    if result.get("errors"):
        raise RuntimeError("; ".join(e["message"] for e in result["errors"]))
    return result["data"]


def list_jobs():
    """Yield every job, 100 per request."""
    after = None
    while True:
        page = graphql(LIST_JOBS, {"first": 100, "after": after})["listJobs"]
        for edge in page["edges"]:
            yield edge["node"]
        if not page["pageInfo"]["hasNextPage"]:
            return
        after = page["pageInfo"]["endCursor"]


def create_job(name, quantity, external_id):
    payload = graphql(CREATE_JOB, {
        "input": {"name": name, "quantity": quantity, "externalId": external_id},
    })["createJob"]
    # Validation problems come back in the payload, not as top-level errors.
    if payload["errors"]:
        for error in payload["errors"]:
            print(f"{error['path']}: {error['message']}", file=sys.stderr)
        sys.exit(1)
    return payload["job"]


if __name__ == "__main__":
    if sys.argv[1:] == ["create"]:
        job = create_job("Business cards", 500, "ERP-10042")
        print(f"Created {job['id']} ({job['status']})")
    else:
        for job in list_jobs():
            print(job["id"], job["status"], job["quantity"], job["name"])

Notes

• graphql() raises for both kinds of failure: an HTTP error status, and a response whose errors array is not empty.
• list_jobs() follows pageInfo.endCursor until hasNextPage is false. The API returns at most 100 items per page.
• externalId must be unique in your tenant. Running create twice with the same value returns a validation error in createJob.errors.