unstructured-transform-client

Python client for the Unstructured Transform v2 API. One document in, structured output back.

from unstructured_transform_client import TransformClient

with TransformClient() as client, open("invoice.pdf", "rb") as f:
    result = client.parse.run(input=f)
    print(result.markdown)

Install

pip install unstructured-transform-client
# export UNSTRUCTURED_API_KEY="your-key-here"

The three flows

Parse a document.

with open("invoice.pdf", "rb") as f:
    result = client.parse.run(input=f)

A path works too, and closes itself:

result = client.parse.run(input="invoice.pdf")

Extract fields from a document in one call. Supplying a schema adds an extraction step to the same job, so the document is parsed once.

result = client.parse.run(
    input="invoice.pdf",
    schema={
        "type": "object",
        "properties": {"invoice_number": {"type": "string"}},
        "required": ["invoice_number"],
        "additionalProperties": False,
    },
)
print(result.extracted_data)

The engine requires required to list every key in properties and additionalProperties to be false. That constraint is not expressible in OpenAPI, so it is not enforced by the types — a schema without it is rejected at request time rather than by your editor.

Extract against a parse you already have. The document is not parsed again.

extraction = client.extract.run(parse_id=result.id, schema=schema)

An extraction is a job like any other. If that call returns an accepted job, follow it to a terminal status before reading result.extracted_data. A single jobs.get right after submitting can still land on queued or processing, with result still None:

import time

accepted = client.extract.run(parse_id=result.id, schema=schema, wait_seconds=0)
deadline = time.monotonic() + 300
job = client.jobs.get(accepted.id)
while job.status in ("queued", "processing"):
    if time.monotonic() >= deadline:
        raise TimeoutError(f"extraction {accepted.id} is still {job.status}")
    time.sleep(2)
    job = client.jobs.get(accepted.id)
print(job.status, job.result.extracted_data if job.result else None)

Give the wait a bound, as above. A job can stay queued or processing, and a loop without a deadline polls forever. jobs.stream is the alternative: it ends on its own at the terminal event, and is shown under Long-running jobs below.

Long-running jobs

Send wait_seconds to block, or wait_seconds=0 to get a job handle back immediately and follow it yourself.

job = client.parse.run(input="contract.pdf", wait_seconds=0)

for event in client.jobs.stream(job.id):
    print(event.event, event.data)
    if event.is_terminal:
        break

stream yields status events as the job progresses, then exactly one terminal result or error. Polling with client.jobs.get(job.id) returns the same resource if you would rather not hold a connection open.

To scan jobs across every page, use the cursor-following iterator. list() still returns one page when you need page boundaries; its total_count is the number of matching jobs across all pages, or None when the service cannot count. Both take status and operation ("parse" or "extract") filters. Jobs whose result has expired are not listed.

for job in client.jobs.iterate(status="completed", operation="parse"):
    print(job.id, job.status)

Retries

Retries are enabled by default for transient failures. Configure them per client, or pass retries=None to disable them:

from unstructured_transform_client import RetryConfig, TransformClient

client = TransformClient(
    retries=RetryConfig(max_attempts=5, max_elapsed_seconds=20),
)
without_retries = TransformClient(retries=None)

GET and DELETE are retried on any transient failure. Everything else — a parse, extract or upload submit, and any other write — is retried only when the failure proves the request never reached the service, such as a refused connection or a DNS failure. A read timeout or a 5xx on a write is not retried, because the service may have acted on it already and a second attempt could duplicate the effect.

Credentials

Set UNSTRUCTURED_API_KEY for the default API-key authentication path:

export UNSTRUCTURED_API_KEY="your-key-here"

Then create the client without passing a key:

client = TransformClient()

An explicit api_key= argument takes precedence over the environment variable. For bearer authentication, pass bearer_token= explicitly. If neither an explicit credential nor UNSTRUCTURED_API_KEY is available, construction raises an error naming UNSTRUCTURED_API_KEY.

client = TransformClient(api_key="...")

What this client tells us about your environment

Every request carries a User-Agent and a set of X-Unstructured-Client-* headers: this package's version, the language, your Python version, your OS family and release, and your CPU architecture. That is what makes a support conversation start from facts.

It never sends your hostname, username, working directory, arbitrary environment variables, IP address, or anything read from your documents. When no explicit credential is provided, UNSTRUCTURED_API_KEY is used only as the API key sent for authentication.

To send only the User-Agent:

client = TransformClient(send_host_headers=False)

or set UNSTRUCTURED_TRANSFORM_DISABLE_HOST_HEADERS=1.

Versioning

This client's version tracks the Transform API revision it was generated from, so 0.18.x of this package describes 0.18.x of the API.

Anything importable from unstructured_transform_client is public. Modules with a leading underscore are not — _generated in particular is produced from the OpenAPI contract at build time and may be reorganised without notice.