Process a directory

Extract structured data from every file in a directory and save one JSON response per file in transform-results. Use the sequential example to test two or three files. For larger directories, use the asynchronous example, which submits up to three files at a time and can resume unfinished jobs after an interruption. Each job tracks one document until processing finishes. The asynchronous example saves each filename, job ID, status, output name, and error so a rerun can continue unfinished work. Set UNSTRUCTURED_API_KEY in your zsh session and install the Python SDK or TypeScript SDK before running an example. The examples use the SDK’s default Transform API endpoint.

The documents list includes every file directly under the input directory. If the directory contains unrelated files, filter that list to the file types you want to process before calling Extract.

These examples call direct Extract with the ACME Financial Highlights schema. It was built from Parse output for the ACME annual report sample. Save the schema beside each script. For other documents, use a schema that matches your files. Define your extraction schema includes an AI prompt for drafting a schema from a CSV or input file.

Python
TypeScript

Process files sequentially

Save this as extract_directory.py. Pass a directory as the first argument. If you omit it, the script reads from Documents:

import json
import sys
from pathlib import Path

from unstructured_transform_client import TransformClient

SCHEMA = json.loads(
    Path(__file__).with_name("acme-financial-highlights-schema.json").read_text(encoding="utf-8")
)

# argv[1] is the directory argument; default to Documents.
input_dir = Path(sys.argv[1] if len(sys.argv) > 1 else "Documents")
output_dir = input_dir / "transform-results"
output_dir.mkdir(exist_ok=True)
documents = sorted(path for path in input_dir.iterdir() if path.is_file())

with TransformClient() as client:
    for document in documents:
        result = client.extract.from_document(input=document, schema=SCHEMA)
        output_file = output_dir / f"{document.name}.json"
        output_file.write_text(result.to_json(), encoding="utf-8")
        print(f"Extracted {document.name} -> {output_file}")

Run it with:

.venv/bin/python extract_directory.py ./Documents

Each call waits for Extract to finish and writes its response before the script submits the next file.

Process files asynchronously

Use this workflow for larger directories or when you may need to stop and resume. It submits up to three files at a time, saves state before and after submission, polls every 30 seconds, and writes each result once. A rerun reuses saved job IDs and skips completed files.The API does not publish request-rate or concurrent-job limits. See supported files and limits, and follow recovery guidance if a request returns rate_limited or quota_exceeded.Save this as extract_directory_async.py. Pass a directory as the first argument. If you omit it, the script reads from Documents:

import json
import os
import sys
import tempfile
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from pathlib import Path

from unstructured_transform_client import TransformClient

SCHEMA = json.loads(
    Path(__file__).with_name("acme-financial-highlights-schema.json").read_text(encoding="utf-8")
)

POLL_SECONDS = 30
MAX_WORKERS = 3
STATE_FILENAME = ".extract-state.json"


def utc_now():
    return datetime.now(timezone.utc).isoformat()


def save_state(state_path, state):
    """Write state atomically so an interruption cannot leave a partial file."""
    with tempfile.NamedTemporaryFile(
        "w",
        dir=state_path.parent,
        prefix=".extract-state.",
        suffix=".tmp",
        delete=False,
        encoding="utf-8",
    ) as temporary:
        json.dump(state, temporary, indent=2)
        temporary.write("\n")
        temporary.flush()
        os.fsync(temporary.fileno())
        temporary_path = Path(temporary.name)
    os.replace(temporary_path, state_path)


def load_state(state_path):
    if not state_path.exists():
        return {"version": 1, "files": {}}
    with state_path.open(encoding="utf-8") as state_file:
        state = json.load(state_file)
    if state.get("version") != 1 or not isinstance(state.get("files"), dict):
        raise RuntimeError(f"Unsupported state file: {state_path}")
    return state


def write_result(output_file, result):
    """Write one result atomically, so a rerun never treats a partial file as done."""
    with tempfile.NamedTemporaryFile(
        "w",
        dir=output_file.parent,
        prefix=f".{output_file.name}.",
        suffix=".tmp",
        delete=False,
        encoding="utf-8",
    ) as temporary:
        temporary.write(result.to_json())
        temporary.write("\n")
        temporary.flush()
        os.fsync(temporary.fileno())
        temporary_path = Path(temporary.name)
    os.replace(temporary_path, output_file)


def describe_job_error(job):
    return str(job.error or f"job stopped with status {job.status}")


# First non-flag argument is the directory; default to Documents.
directory_argument = next(
    (argument for argument in sys.argv[1:] if not argument.startswith("--")),
    None,
)
input_dir = Path(directory_argument or "Documents")
retry_failed = "--retry-failed" in sys.argv[1:]
output_dir = input_dir / "transform-results"
output_dir.mkdir(parents=True, exist_ok=True)
state_path = output_dir / STATE_FILENAME
documents = sorted(path for path in input_dir.iterdir() if path.is_file())

with TransformClient() as client:
    state = load_state(state_path)
    files = state["files"]

    # Discover files and persist their output names before submitting anything.
    for document in documents:
        record = files.setdefault(
            document.name,
            {"status": "ready", "output_name": f"{document.name}.json"},
        )
        output_file = output_dir / record["output_name"]
        if output_file.exists() and record["status"] not in {
            "completed",
            "completed_with_warnings",
        }:
            record.update(status="completed", completed_at=utc_now())
        if retry_failed and record["status"] in {"failed", "cancelled"}:
            record.update(status="ready")
            record.pop("job_id", None)
            record.pop("error", None)
    save_state(state_path, state)

    # Persist "submitting" before parallel POST requests. If the process stops
    # during submission, reruns will not silently submit the same file twice.
    ready_documents = [
        document for document in documents if files[document.name]["status"] == "ready"
    ]
    for document in ready_documents:
        files[document.name].update(status="submitting", submitted_at=utc_now())
    save_state(state_path, state)

    def submit(document):
        accepted = client.extract.from_document(
            input=document, schema=SCHEMA, wait_seconds=0
        )
        return accepted.id

    with ThreadPoolExecutor(max_workers=min(MAX_WORKERS, max(1, len(ready_documents)))) as executor:
        submissions = {
            executor.submit(submit, document): document for document in ready_documents
        }
        for future in as_completed(submissions):
            document = submissions[future]
            record = files[document.name]
            try:
                record.update(
                    status="queued",
                    job_id=future.result(),
                    submitted_at=utc_now(),
                )
                record.pop("error", None)
                print(f"Submitted {document.name}")
            except Exception as error:
                record.update(
                    status="submission_unknown",
                    error=f"Submission could not be confirmed: {error}",
                )
                print(f"Submission status is unknown for {document.name}: {error}")
            save_state(state_path, state)

    def retrieve(job_id):
        return client.jobs.get(job_id)

    while True:
        pending = [
            (filename, record)
            for filename, record in files.items()
            if record["status"] in {"queued", "processing", "write_error"}
            and record.get("job_id")
        ]
        if not pending:
            break

        # Poll all known jobs in parallel, then persist every state transition.
        with ThreadPoolExecutor(max_workers=min(MAX_WORKERS, len(pending))) as executor:
            jobs = {
                executor.submit(retrieve, record["job_id"]): (filename, record)
                for filename, record in pending
            }
            for future in as_completed(jobs):
                filename, record = jobs[future]
                try:
                    job = future.result()
                    record.update(status=job.status, last_checked_at=utc_now())
                    if job.status in ("completed", "completed_with_warnings"):
                        if job.result is None:
                            record.update(
                                status="result_unavailable",
                                error="The job completed, but its result was unavailable.",
                            )
                        else:
                            try:
                                write_result(
                                    output_dir / record["output_name"],
                                    job.result,
                                )
                            except Exception as error:
                                record.update(
                                    status="write_error",
                                    error=f"Could not write the result: {error}",
                                )
                            else:
                                record.update(
                                    status=job.status,
                                    completed_at=utc_now(),
                                )
                                record.pop("error", None)
                                print(f"Extracted {filename}")
                    elif job.status in ("failed", "cancelled"):
                        record["error"] = describe_job_error(job)
                        print(f"{job.status.title()} {filename}: {record['error']}")
                    elif job.status not in ("queued", "processing"):
                        record.update(
                            status="unexpected_status",
                            error=f"Unexpected job status: {job.status}",
                        )
                except Exception as error:
                    record["last_error"] = f"Polling failed: {error}"
                    print(f"Could not check {filename}: {error}")
                save_state(state_path, state)

        pending_count = sum(
            record["status"] in {"queued", "processing", "write_error"}
            for record in files.values()
        )
        if pending_count:
            print(
                f"{pending_count} job(s) still running. "
                f"Checking again in {POLL_SECONDS} seconds."
            )
            time.sleep(POLL_SECONDS)

print(f"State saved to {state_path}. Rerun this command to resume unfinished work.")

wait_seconds=0 returns a job handle immediately. The state file records each handle before polling. If the process stops while polling or the connection fails, rerunning the command uses the saved job IDs instead of submitting completed files again. A job with status failed or cancelled stays in the state file with its error so you can inspect it or rerun with --retry-failed.An interruption during submission can leave a record as submitting without a job ID. A request error can leave it as submission_unknown. The script does not retry either state automatically because the API may already have accepted the document.To recover either state:

Stop all copies of the script. In the selected input directory, open transform-results/.extract-state.json (for example, ./Documents/transform-results/.extract-state.json).
Check your job history for the document and submission time. If you identify the accepted job, add its ID as job_id and change the record’s status to queued. Keep the other fields.
If you confirm that no job was accepted, change the record’s status to ready and remove any job_id. The next run submits the document again.
Save the state file and rerun the command. If you cannot confirm whether a job was accepted, leave the record unchanged and investigate before retrying. Retrying an uncertain submission can create a duplicate job.

Run the asynchronous example with:

.venv/bin/python extract_directory_async.py ./Documents

To retry only jobs that the API reported as failed or cancelled:

.venv/bin/python extract_directory_async.py ./Documents --retry-failed

Process files sequentially

Save this as extract-directory.mjs. Pass a directory as the first argument. If you omit it, the script reads from Documents:

import { mkdir, readFile, readdir, writeFile } from "node:fs/promises";
import { join } from "node:path";

import { TransformClient, isAccepted } from "unstructured-transform-client";

// argv[2] is the directory argument; default to Documents.
const inputDir = process.argv[2] ?? "Documents";
const schema = JSON.parse(
  await readFile(new URL("./acme-financial-highlights-schema.json", import.meta.url), "utf8"),
);
const outputDir = join(inputDir, "transform-results");
await mkdir(outputDir, { recursive: true });
const documents = (await readdir(inputDir, { withFileTypes: true }))
  .filter((entry) => entry.isFile())
  .map((entry) => entry.name)
  .sort();

const client = new TransformClient();

for (const filename of documents) {
  const bytes = await readFile(join(inputDir, filename));
  const outcome = await client.extract.fromDocument({
    input: new File([bytes], filename),
    schema,
  });
  if (isAccepted(outcome)) {
    throw new Error(`Extract is still running as job ${outcome.body.id}`);
  }
  const outputFile = join(outputDir, `${filename}.json`);
  await writeFile(outputFile, JSON.stringify(outcome.body, null, 2));
  console.log(`Extracted ${filename} -> ${outputFile}`);
}

Run it with:

node extract-directory.mjs ./Documents

Each call waits for Extract to finish and writes its response before the script submits the next file.

Process files asynchronously

Use this workflow for larger directories or when you may need to stop and resume. It submits up to three files at a time, saves state before and after submission, polls every 30 seconds, and writes each result once. A rerun reuses saved job IDs and skips completed files.The API does not publish request-rate or concurrent-job limits. See supported files and limits, and follow recovery guidance if a request returns rate_limited or quota_exceeded.Save this as extract-directory-async.mjs. Pass a directory as the first argument. If you omit it, the script reads from Documents:

import {
  access,
  mkdir,
  readFile,
  readdir,
  rename,
  writeFile,
} from "node:fs/promises";
import { join } from "node:path";

import { TransformClient, isAccepted } from "unstructured-transform-client";

const POLL_MILLISECONDS = 30_000;
const MAX_WORKERS = 3;
const STATE_FILENAME = ".extract-state.json";

function errorMessage(error) {
  return error instanceof Error ? error.message : String(error);
}

async function exists(path) {
  try {
    await access(path);
    return true;
  } catch {
    return false;
  }
}

async function saveState(statePath, state) {
  // Write a complete temporary file before replacing the state file.
  const temporaryPath = statePath + ".tmp";
  await writeFile(temporaryPath, JSON.stringify(state, null, 2) + "\n");
  await rename(temporaryPath, statePath);
}

async function loadState(statePath) {
  if (!(await exists(statePath))) {
    return { version: 1, files: {} };
  }
  const state = JSON.parse(await readFile(statePath, "utf8"));
  if (state.version !== 1 || typeof state.files !== "object") {
    throw new Error("Unsupported state file: " + statePath);
  }
  return state;
}

async function writeResult(outputFile, result) {
  // Atomic output prevents a partial file from looking complete after a crash.
  const temporaryPath = outputFile + ".tmp";
  await writeFile(temporaryPath, JSON.stringify(result, null, 2) + "\n");
  await rename(temporaryPath, outputFile);
}

async function mapWithConcurrency(items, workerCount, action) {
  const results = new Array(items.length);
  let nextIndex = 0;
  const workers = Array.from(
    { length: Math.min(workerCount, items.length) },
    async () => {
      while (nextIndex < items.length) {
        const index = nextIndex++;
        results[index] = await action(items[index]);
      }
    },
  );
  await Promise.all(workers);
  return results;
}

const args = process.argv.slice(2);
// First non-flag argument is the directory; default to Documents.
const directoryArgument = args.find((argument) => !argument.startsWith("--"));
const inputDir = directoryArgument ?? "Documents";
const retryFailed = args.includes("--retry-failed");
const outputDir = join(inputDir, "transform-results");
await mkdir(outputDir, { recursive: true });
const statePath = join(outputDir, STATE_FILENAME);
const documents = (await readdir(inputDir, { withFileTypes: true }))
  .filter((entry) => entry.isFile())
  .map((entry) => entry.name)
  .sort();

const schema = JSON.parse(
  await readFile(new URL("./acme-financial-highlights-schema.json", import.meta.url), "utf8"),
);
const client = new TransformClient();
const state = await loadState(statePath);

// Discover files and persist their output names before submitting anything.
for (const filename of documents) {
  const record = (state.files[filename] ??= {
    status: "ready",
    outputName: filename + ".json",
  });
  const outputFile = join(outputDir, record.outputName);
  if (
    (await exists(outputFile)) &&
    record.status !== "completed" &&
    record.status !== "completed_with_warnings"
  ) {
    record.status = "completed";
    record.completedAt = new Date().toISOString();
  }
  if (retryFailed && (record.status === "failed" || record.status === "cancelled")) {
    record.status = "ready";
    delete record.jobId;
    delete record.error;
  }
}
await saveState(statePath, state);

// Persist "submitting" before parallel POST requests. If the process stops
// during submission, reruns will not silently submit the same file twice.
const readyDocuments = documents.filter(
  (filename) => state.files[filename].status === "ready",
);
for (const filename of readyDocuments) {
  state.files[filename].status = "submitting";
}
await saveState(statePath, state);

async function submit(filename) {
  const bytes = await readFile(join(inputDir, filename));
  const accepted = await client.extract.fromDocument({
    input: new File([bytes], filename),
    schema,
    waitSeconds: 0,
  });
  if (!isAccepted(accepted)) {
    throw new Error("Extract did not return a job handle for " + filename);
  }
  return accepted.body.id;
}

// Submit all ready files in parallel, then persist each returned job ID.
const submissions = await mapWithConcurrency(
  readyDocuments,
  MAX_WORKERS,
  async (filename) => {
    try {
      return { filename, jobId: await submit(filename) };
    } catch (error) {
      return { filename, error: errorMessage(error) };
    }
  },
);
for (const submission of submissions) {
  const record = state.files[submission.filename];
  if ("jobId" in submission) {
    record.status = "queued";
    record.jobId = submission.jobId;
    delete record.error;
    console.log("Submitted " + submission.filename);
  } else {
    record.status = "submission_unknown";
    record.error = "Submission could not be confirmed: " + submission.error;
    console.log(
      "Submission status is unknown for " +
        submission.filename +
        ": " +
        submission.error,
    );
  }
  await saveState(statePath, state);
}

while (true) {
  const pending = documents
    .map((filename) => ({ filename, record: state.files[filename] }))
    .filter(
      ({ record }) =>
        (record.status === "queued" ||
          record.status === "processing" ||
          record.status === "write_error") &&
        record.jobId !== undefined,
    );
  if (pending.length === 0) {
    break;
  }

  // Poll all known jobs in parallel, then persist every state transition.
  const results = await mapWithConcurrency(
    pending,
    MAX_WORKERS,
    async ({ filename, record }) => {
      try {
        return {
          filename,
          job: await client.jobs.get(record.jobId),
        };
      } catch (error) {
        return { filename, error: errorMessage(error) };
      }
    },
  );

  for (const result of results) {
    const record = state.files[result.filename];
    if ("error" in result) {
      record.lastError = "Polling failed: " + result.error;
      console.log("Could not check " + result.filename + ": " + result.error);
      continue;
    }

    const job = result.job;
    record.status = job.status;
    if (job.status === "completed" || job.status === "completed_with_warnings") {
      if (job.result === undefined) {
        record.status = "result_unavailable";
        record.error = "The job completed, but its result was unavailable.";
      } else {
        try {
          await writeResult(join(outputDir, record.outputName), job.result);
          record.completedAt = new Date().toISOString();
          delete record.error;
          console.log("Extracted " + result.filename);
        } catch (error) {
          record.status = "write_error";
          record.error = "Could not write the result: " + errorMessage(error);
        }
      }
    } else if (job.status === "failed" || job.status === "cancelled") {
      record.error = String(job.error ?? "job stopped with status " + job.status);
      console.log(job.status + " " + result.filename + ": " + record.error);
    } else if (job.status !== "queued" && job.status !== "processing") {
      record.status = "unexpected_status";
      record.error = "Unexpected job status: " + job.status;
    }
  }
  await saveState(statePath, state);

  const pendingCount = documents.filter((filename) => {
    const status = state.files[filename].status;
    return status === "queued" || status === "processing" || status === "write_error";
  }).length;
  if (pendingCount > 0) {
    console.log(
      pendingCount + " job(s) still running. Checking again in 30 seconds.",
    );
    await new Promise((resolve) => setTimeout(resolve, POLL_MILLISECONDS));
  }
}

console.log("State saved to " + statePath + ". Rerun this command to resume unfinished work.");

waitSeconds: 0 returns a job handle immediately. The state file records each handle before polling. If the process stops while polling or the connection fails, rerunning the command uses the saved job IDs instead of submitting completed files again. A job with status failed or cancelled stays in the state file with its error so you can inspect it or rerun with --retry-failed.An interruption during submission can leave a record as submitting without a job ID. A request error can leave it as submission_unknown. The script does not retry either state automatically because the API may already have accepted the document.To recover either state:

Stop all copies of the script. In the selected input directory, open transform-results/.extract-state.json (for example, ./Documents/transform-results/.extract-state.json).
Check your job history for the document and submission time. If you identify the accepted job, add its ID as jobId and change the record’s status to queued. Keep the other fields.
If you confirm that no job was accepted, change the record’s status to ready and remove any jobId. The next run submits the document again.
Save the state file and rerun the command. If you cannot confirm whether a job was accepted, leave the record unchanged and investigate before retrying. Retrying an uncertain submission can create a duplicate job.

Run the asynchronous example with:

node extract-directory-async.mjs ./Documents

To retry only jobs that the API reported with status failed or cancelled:

node extract-directory-async.mjs ./Documents --retry-failed

For more ways to define fields, see Define your extraction schema. Direct Extract accepts one document and one schema per request.

Integrate

API Reference

API endpoints

Process files sequentially

Process files asynchronously

Process files sequentially

Process files asynchronously

​Process files sequentially

​Process files asynchronously

​Process files sequentially

​Process files asynchronously

Process files sequentially

Process files asynchronously

Process files sequentially

Process files asynchronously