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's own bookkeeping is required listing every key in properties and additionalProperties set to false. Omit either and the API fills it in for you rather than rejecting the request, so this is a convenience, not a requirement to hand-author.

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

With no wait_seconds, parse.run and extract.run wait for the job. They return the finished result, or, if the job is still running after the wait, a job handle (HTTP 202) while the job keeps running. The wait is 15 minutes, or less if the deployment sets a lower limit. A handle has id, status and poll_url; it does not have markdown or extracted_data, so check for it before reading a result:

outcome = client.parse.run(input="contract.pdf")
if getattr(outcome, "poll_url", None):
    # Still running after the wait. Follow it to the end.
    for event in client.jobs.stream(outcome.id):
        print(event.event, event.data)
        if event.is_terminal:
            break
else:
    print(outcome.markdown)

Send wait_seconds to choose the wait (the deployment caps it), or wait_seconds=0 to get a job handle back immediately and follow it yourself. If you set timeout_seconds on the client, keep it above the wait you expect, or the request is cut off before the handle arrives.

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.

Waiting out a restarting worker

One refusal is retried on every method, including a submit: 503 with the error code profile_temporarily_unavailable. The service returns it while the worker behind a profile is being replaced, before it has accepted anything, so sending the request again cannot create a second job. A replacement can take well over a minute, longer than the service will hold one request, so the client sends several separate requests instead.

• Each wait starts from the Retry-After the service sent (10 seconds when the header is absent), kept between 2 and 30 seconds, grows with each refusal and is slightly jittered. The 30-second cap always wins, so a larger Retry-After is shortened to 30 seconds.
• A new attempt starts only if its wait still fits inside unavailable_budget_seconds (125 by default), counted from the first attempt, and at most unavailable_max_retries (6) times. A request already in flight is never cancelled, so the total can exceed the budget by one attempt.
• It applies only when the request can be sent again unchanged, and does not change Prefer or the 202 job behavior.
• When the budget runs out, the last response is raised as the usual ServiceException (status == 503, with the service's error body and Retry-After header).

client = TransformClient(
    retries=RetryConfig(unavailable_budget_seconds=300),
)
no_wait = TransformClient(retries=RetryConfig(unavailable_budget_seconds=0))

unavailable_budget_seconds=0 turns only this retry off; retries=None turns all retries off. Every other status, including any other 503, keeps the behavior described above.

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.