Build an API client

export ION_BASE_URL="https://api.buildwithion.com"
export ION_TOKEN="<your-access-token>"

python -m venv .venv
source .venv/bin/activate
pip install "httpx>=0.27" "tenacity>=8.0"

import os
import httpx
from typing import Any
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class IonError(RuntimeError):
    """Raised when ION returns errors[] in the response body."""
    def __init__(self, errors: list[dict[str, Any]]):
        self.errors = errors
        super().__init__("; ".join(e.get("message", "") for e in errors))


class IonClient:
    def __init__(self, base_url: str | None = None, token: str | None = None):
        self.base_url = (base_url or os.environ["ION_BASE_URL"]).rstrip("/")
        self.token = token or os.environ["ION_TOKEN"]
        self._http = httpx.Client(
            base_url=self.base_url,
            headers={"Authorization": f"Bearer {self.token}"},
            timeout=30.0,
        )

    @retry(
        retry=retry_if_exception_type((httpx.HTTPError,)),
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        reraise=True,
    )
    def _post_graphql(self, query: str, variables: dict[str, Any] | None = None) -> dict[str, Any]:
        r = self._http.post(
            "/graphql",
            json={"query": query, "variables": variables or {}},
        )
        r.raise_for_status()
        payload = r.json()
        if payload.get("errors"):
            raise IonError(payload["errors"])
        return payload["data"]

    def query(self, query: str, **variables) -> dict[str, Any]:
        return self._post_graphql(query, variables)

    def mutate(self, mutation: str, **variables) -> dict[str, Any]:
        return self._post_graphql(mutation, variables)

    def close(self):
        self._http.close()

ME = """
query CurrentUser {
  me {
    id
    name
    email
    organization { id domain }
  }
}
"""

with IonClient() as ion:
    me = ion.query(ME)["me"]
    print(f"Authenticated as {me['name']} ({me['email']}) in org {me['organization']['domain']}")

LIST_PARTS = """
query Parts($filters: PartsInputFilters, $first: Int) {
  parts(filters: $filters, first: $first) {
    edges {
      node {
        id
        partNumber
        description
        revision
        partType
        status
        partSubtypes { id name }
      }
    }
  }
}
"""

with IonClient() as ion:
    edges = ion.query(LIST_PARTS, filters={}, first=20)["parts"]["edges"]
    for edge in edges:
        p = edge["node"]
        subtypes = ", ".join(s["name"] for s in p["partSubtypes"]) or "—"
        print(f"{p['partNumber']}\t{p['description']}\t{subtypes}")

CREATE_RUN = """
mutation CreateRun($input: CreateRunInput!) {
  createRun(input: $input) {
    run {
      id
      title
      status
      procedure { id title }
      partInventory { id serialNumber }
    }
  }
}
"""

with IonClient() as ion:
    new_run = ion.mutate(
        CREATE_RUN,
        input={
            "procedureId": 12,
            "partInventoryId": 9876,
            "title": "Build #2026-04-26 Unit 1",
        },
    )["createRun"]["run"]
    print(f"Created run {new_run['id']}: {new_run['title']}")

SUBMIT_STEP = """
mutation SubmitStep($input: UpdateRunStepInput!) {
  updateRunStep(input: $input) {
    runStep {
      id
      status
      endTime
      fields { id name value }
    }
  }
}
"""

with IonClient() as ion:
    result = ion.mutate(
        SUBMIT_STEP,
        input={
            "id": 4567,
            "etag": "abc123",        # required for optimistic concurrency
            "status": "COMPLETE",
        },
    )["updateRunStep"]["runStep"]
    print(f"Step {result['id']} → {result['status']} at {result['endTime']}")

import os

REQUEST_UPLOAD = """
query Upload($filename: String!) {
  requestFileUploadUrl(filename: $filename) {
    uploadUrl
    s3Key
    contentType
  }
}
"""

GET_RUN_STEP_ENTITY = """
query StepEntity($id: ID!) {
  runStep(id: $id) {
    entity { id }
  }
}
"""

CREATE_FILE_ATTACHMENT = """
mutation CreateAttachment($input: CreateFileAttachmentInput!) {
  createFileAttachment(input: $input) {
    fileAttachment { id filename downloadUrl }
  }
}
"""

def attach_to_run_step(ion: IonClient, run_step_id: int, file_path: str) -> int:
    filename = os.path.basename(file_path)

    # 1. Get a signed upload URL
    upload = ion.query(REQUEST_UPLOAD, filename=filename)["requestFileUploadUrl"]

    # 2. PUT the file to S3, capture the ETag
    with open(file_path, "rb") as f:
        s3_response = httpx.put(
            upload["uploadUrl"],
            content=f.read(),
            headers={"Content-Type": upload["contentType"]},
            timeout=120.0,
        )
    s3_response.raise_for_status()
    s3_etag = s3_response.headers["ETag"].strip('"')

    # 3. Resolve the run step's entity.id (NOT the run step's own id)
    entity_id = ion.query(GET_RUN_STEP_ENTITY, id=run_step_id)["runStep"]["entity"]["id"]

    # 4. Register the attachment
    attachment = ion.mutate(
        CREATE_FILE_ATTACHMENT,
        input={
            "filename": filename,
            "s3Key": upload["s3Key"],
            "s3Etag": s3_etag,
            "entityId": entity_id,
            "validateUpload": True,
        },
    )["createFileAttachment"]["fileAttachment"]
    return attachment["id"]


with IonClient() as ion:
    attachment_id = attach_to_run_step(ion, run_step_id=4567, file_path="./inspection.jpg")
    print(f"Attached file id={attachment_id}")

from httpx import HTTPStatusError, ConnectError

def safely_create_run(ion: IonClient, **inputs) -> int | None:
    try:
        result = ion.mutate(CREATE_RUN, input=inputs)
    except IonError as e:
        # GraphQL errors[] — usually 4xx auth/permission/validation
        # Surface to the user; don't retry blindly
        print(f"ION rejected the request: {e}")
        return None
    except HTTPStatusError as e:
        # 5xx after retries exhausted
        print(f"ION 5xx after retries: {e.response.status_code}")
        return None
    except ConnectError:
        # Network down
        print("Cannot reach ION")
        return None
    return result["createRun"]["run"]["id"]

FRAGMENTS = """
fragment PartFields on Part {
  id
  partNumber
  description
  revision
  partType
  status
  partSubtypes { id name }
}

fragment RunFields on Run {
  id
  title
  status
  procedure { id title }
  partInventory { id serialNumber part { ...PartFields } }
}
"""

LIST_RUNS_WITH_PARTS = FRAGMENTS + """
query Runs($filters: RunsInputFilters) {
  runs(filters: $filters) {
    edges {
      node {
        ...RunFields
      }
    }
  }
}
"""