Python SDK — Usage

Construction

1import os
2from onepin import OnePinClient
3
4# Explicit token
5client = OnePinClient(token="op_live_...")
6
7# From environment variable (explicit read — the SDK does not auto-read env vars)
8client = OnePinClient(token=os.environ["ONEPIN_API_KEY"])
9
10# Custom base URL
11client = OnePinClient(
12 token="op_live_...",
13 base_url="https://api.onepin.ai",
14)

Client lifecycle

Construct the client once and reuse it for the lifetime of your process — it holds an underlying httpx connection pool. The generated client is not a context manager (no with / async with); just build it and use it.

Note the async pager pattern: list() on the async client is a coroutine that returns an AsyncPager, so await the call first, then async for over the result.

1client = OnePinClient(token="op_live_...")
2for wf in client.workflows.list():
3 print(wf.id)

Workflows

1# List — iterate the pager directly; pages are fetched automatically
2for wf in client.workflows.list(limit=50, offset=0):
3 print(wf.id, wf.name, wf.status)
4
5# Get one — response is an envelope; the workflow is at .data
6result = client.workflows.get(workflow_id="3fa85f64-5717-4562-b3fc-2c963f66afa6")
7wf = result.data
8print(wf.id, wf.name)
9
10# Start a run — envelope; run at .data
11result = client.workflows.runs.start("3fa85f64-5717-4562-b3fc-2c963f66afa6")
12run = result.data
13print(run.id, run.status)
14
15# List run history — iterate the pager directly
16for r in client.workflows.runs.list("3fa85f64-5717-4562-b3fc-2c963f66afa6", limit=20):
17 print(r.id, r.status, r.created_at)

Voices

1for v in client.voices.list(limit=50):
2 print(v.id, v.display_name, v.locale)
3
4result = client.voices.get(voice_id="...")
5voice = result.data
6
7result = client.voices.similar(voice_id="...")
8similar = result.data

Templates

1# List gallery templates — iterate the pager directly
2for t in client.templates.list(limit=50):
3 print(t.id, t.name)
4
5# Get one — envelope; template at .data
6result = client.templates.get(template_id="3fa85f64-5717-4562-b3fc-2c963f66afa6")
7tmpl = result.data
8
9# Clone into a workflow in your workspace (requires workflows:write scope)
10# The new workflow is at .data
11result = client.templates.clone(
12 template_id="3fa85f64-5717-4562-b3fc-2c963f66afa6",
13 name="My production pipeline", # optional; defaults to "<template name> (copy)"
14)
15new_wf = result.data
16print(new_wf.id, new_wf.name)

Uploads

Uploads use a three-step presigned URL flow: create → PUT bytes to S3 → confirm.

1import httpx
2
3# Step 1: create the upload record and get a presigned URL
4result = client.uploads.create(
5 filename="my-script.txt",
6 category="script", # "script" (.txt/.docx) or "dictionary" (.csv)
7)
8upload_id = result.data.upload.id
9upload_url = result.data.upload_url
10
11# Step 2: PUT the file bytes directly to S3 (not via the SDK — raw HTTP)
12with open("my-script.txt", "rb") as f:
13 httpx.put(upload_url, content=f.read()) # must return 200
14
15# Step 3: confirm the upload and bind it to a workflow
16confirmed = client.uploads.confirm(
17 upload_id=upload_id,
18 context_type="workflow",
19 context_id="3fa85f64-5717-4562-b3fc-2c963f66afa6",
20)
21print(confirmed.data.id, confirmed.data.status)
22
23# Delete an upload
24client.uploads.delete(upload_id)

Categories: category must be "script" (.txt, .docx) or "dictionary" (.csv). Audio file upload uses a separate endpoint not available via API key.


Pagination

Every list() call returns a pager that you iterate directly. Pages are fetched automatically as you consume them — there is no separate page object, no offset cursor to track manually, and no helper method needed.

1# Sync — iterate the pager; fetches pages on demand
2for wf in client.workflows.list(limit=50):
3 process(wf)
async
1pager = await client.workflows.list(limit=50)
2async for wf in pager:
3 process(wf)

Use the offset and limit parameters to start from a known position:

1for wf in client.workflows.list(offset=100, limit=50):
2 process(wf)

Single-object responses (from get(), runs.get(), runs.status(), voices.get(), etc.) are envelopes with a .data attribute holding the object and a .meta attribute with request metadata.


Error handling

1import os
2from onepin import OnePinClient
3from onepin.core.api_error import ApiError
4from onepin.errors import NotFoundError
5
6client = OnePinClient(token=os.environ["ONEPIN_API_KEY"])
7
8try:
9 result = client.workflows.get(workflow_id="nonexistent-id")
10except NotFoundError:
11 print("workflow not found")
12except ApiError as e:
13 print(f"API error {e.status_code}: {e.body}")

See Errors for the full exception hierarchy and retry patterns.