Partition quickstart

Complete the requirements before you begin. You can learn about Unstructured’s partitioning before you begin.

Create the job

Replace INPUT_DIR with the path to your local directory of files to process. The response includes the job ID.

Each local-file processing job can include up to 10 files, and each file can be up to 50 MB.These local-file processing jobs are on-demand jobs. A job enters the queue if the Workspace is running its allowed number of on-demand jobs or an earlier job is queued. By default, send at most one queued submission per second per Workspace. Another queued submission within that second returns HTTP 429. Jobs that start without entering the queue do not count toward this limit.For Unstructured SaaS accounts, up to five on-demand jobs can run at once per Workspace. Accepted queued jobs remain scheduled until capacity is available. Your deployment settings can change both this capacity and the queued-submission rate.For other Pipelines API limits, see Pipelines API rate limits.

curl
Python

Save and run this script:

#!/usr/bin/env bash

INPUT_DIR="/full/path/to/your/directory"

form_args=()
for filepath in "$INPUT_DIR"/*; do
    [ -f "$filepath" ] || continue
    filename=$(basename "$filepath")
    mimetype=$(file --mime-type -b "$filepath")
    form_args+=(--form "input_files=@${filepath};filename=${filename};type=${mimetype}")
done

response=$(curl --request POST --location \
  "$UNSTRUCTURED_API_URL/jobs/" \
  --header "accept: application/json" \
  --header "unstructured-api-key: $UNSTRUCTURED_API_KEY" \
  --form 'request_data={"job_nodes":[{"name":"Partitioner","type":"partition","subtype":"vlm","settings":{"is_dynamic":true,"allow_fast":true}}]}' \
  "${form_args[@]}")

JOB_ID=$(echo "$response" | jq -r '.id')
echo "Job ID: $JOB_ID"

This script requires jq to parse the JSON response.

import mimetypes
import os
import json

from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateJobRequest
from unstructured_client.models.shared import BodyCreateJob, InputFiles

INPUT_DIR = "/full/path/to/your/directory"

client = UnstructuredClient(
    api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
    server_url=os.getenv("UNSTRUCTURED_API_URL")
)

input_files = []
for filename in os.listdir(INPUT_DIR):
    full_path = os.path.join(INPUT_DIR, filename)
    if not os.path.isfile(full_path):
        continue
    content_type, _ = mimetypes.guess_type(full_path)
    input_files.append(
        InputFiles(
            content=open(full_path, "rb"),
            file_name=filename,
            content_type=content_type or "application/octet-stream"
        )
    )

response = client.jobs.create_job(
    request=CreateJobRequest(
        body_create_job=BodyCreateJob(
            request_data=json.dumps({
                "job_nodes": [
                    {
                        "name": "Partitioner",
                        "type": "partition",
                        "subtype": "vlm",
                        "settings": {
                            "is_dynamic": True,
                            "allow_fast": True
                        }
                    }
                ]
            }),
            input_files=input_files
        )
    )
)

job_info = response.job_information
print(f"Job ID: {job_info.id}")

Poll for job status

Replace JOB_ID with the job ID from the previous step. This script polls every 10 seconds and stops when the job completes.

curl
Python

Save and run this script:

#!/usr/bin/env bash

JOB_ID="<job-id>"

while true; do
    job=$(curl --request GET --silent --location \
      "$UNSTRUCTURED_API_URL/jobs/$JOB_ID" \
      --header "accept: application/json" \
      --header "unstructured-api-key: $UNSTRUCTURED_API_KEY")

    status=$(echo "$job" | jq -r '.status')
    echo "Job status: $status"

    if [ "$status" = "COMPLETED" ]; then
        echo "Job completed."
        echo "Output node file IDs: $(echo "$job" | jq -c '[.output_node_files[].file_id]')"
        break
    elif [ "$status" = "FAILED" ] || [ "$status" = "STOPPED" ]; then
        echo "Job did not complete successfully: $status"
        exit 1
    fi

    sleep 10
done

This script requires jq to parse the JSON response.

import os
import time

from unstructured_client import UnstructuredClient

JOB_ID = "<job-id>"

client = UnstructuredClient(
    api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
    server_url=os.getenv("UNSTRUCTURED_API_URL")
)

while True:
    response = client.jobs.get_job(request={"job_id": JOB_ID})
    job_info = response.job_information
    status = job_info.status

    print(f"Job status: {status.value}")

    if status == "COMPLETED":
        print("Job completed.")
        print(f"Output node file IDs: {[f.file_id for f in (job_info.output_node_files or [])]}")
        break
    elif status in ("FAILED", "STOPPED"):
        raise RuntimeError(f"Job did not complete successfully: {status}")

    time.sleep(10)

Download the job output

Replace JOB_ID, OUTPUT_FILE_IDS, and OUTPUT_DIR with your values from the previous steps.

curl
Python

Save and run this script:

#!/usr/bin/env bash

JOB_ID="<job-id>"
OUTPUT_FILE_IDS=("<output-file-id>" "<output-file-id>") # From Step 1
OUTPUT_DIR="/full/path/to/your/output/directory"

mkdir -p "$OUTPUT_DIR"

for file_id in "${OUTPUT_FILE_IDS[@]}"; do
    curl --request GET --silent --location \
      "$UNSTRUCTURED_API_URL/jobs/$JOB_ID/download?file_id=$file_id" \
      --header "unstructured-api-key: $UNSTRUCTURED_API_KEY" \
      --output "$OUTPUT_DIR/$file_id.json"
    echo "Saved: $OUTPUT_DIR/$file_id.json"
done

import json
import os

from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import DownloadJobOutputRequest

JOB_ID = "<job-id>"
OUTPUT_FILE_IDS = ["<output-file-id>", "<output-file-id>"] # From Step 1
OUTPUT_DIR = "/full/path/to/your/output/directory"

client = UnstructuredClient(
    api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
    server_url=os.getenv("UNSTRUCTURED_API_URL")
)

os.makedirs(OUTPUT_DIR, exist_ok=True)

for file_id in OUTPUT_FILE_IDS:
    response = client.jobs.download_job_output(
        request=DownloadJobOutputRequest(job_id=JOB_ID, file_id=file_id)
    )
    output_path = os.path.join(OUTPUT_DIR, f"{file_id}.json")
    with open(output_path, "w") as f:
        json.dump(response.any, f, indent=4)
    print(f"Saved: {output_path}")

Complete end-to-end script

Replace INPUT_DIR and OUTPUT_DIR with your directory paths, then save and run this script.

curl
Python

This script requires jq to parse JSON responses.

#!/usr/bin/env bash

INPUT_DIR="/full/path/to/your/input/directory"
OUTPUT_DIR="/full/path/to/your/output/directory"

# Step 1: Create the job.
form_args=()
for filepath in "$INPUT_DIR"/*; do
    [ -f "$filepath" ] || continue
    filename=$(basename "$filepath")
    mimetype=$(file --mime-type -b "$filepath")
    form_args+=(--form "input_files=@${filepath};filename=${filename};type=${mimetype}")
done

response=$(curl --request POST --location \
  "$UNSTRUCTURED_API_URL/jobs/" \
  --header "accept: application/json" \
  --header "unstructured-api-key: $UNSTRUCTURED_API_KEY" \
  --form 'request_data={"job_nodes":[{"name":"Partitioner","type":"partition","subtype":"vlm","settings":{"is_dynamic":true,"allow_fast":true}}]}' \
  "${form_args[@]}")

JOB_ID=$(echo "$response" | jq -r '.id')
echo "Job ID: $JOB_ID"

# Step 2: Poll until the job completes.
output_file_ids=()

while true; do
    job=$(curl --request GET --silent --location \
      "$UNSTRUCTURED_API_URL/jobs/$JOB_ID" \
      --header "accept: application/json" \
      --header "unstructured-api-key: $UNSTRUCTURED_API_KEY")

    status=$(echo "$job" | jq -r '.status')
    echo "Job status: $status"

    if [ "$status" = "COMPLETED" ]; then
        echo "Job completed."

        while IFS= read -r id; do
            output_file_ids+=("$id")
        done < <(echo "$job" | jq -r '.output_node_files[].file_id')

        break
    elif [ "$status" = "FAILED" ] || [ "$status" = "STOPPED" ]; then
        echo "Job did not complete successfully: $status"
        exit 1
    fi

    sleep 10
done

# Step 3: Download the job output.
mkdir -p "$OUTPUT_DIR"

for file_id in "${output_file_ids[@]}"; do
    curl --request GET --silent --location \
      "$UNSTRUCTURED_API_URL/jobs/$JOB_ID/download?file_id=$file_id" \
      --header "unstructured-api-key: $UNSTRUCTURED_API_KEY" \
      --output "$OUTPUT_DIR/$file_id.json"
    echo "Saved: $OUTPUT_DIR/$file_id.json"
done

import json
import mimetypes
import os
import time

from unstructured_client import UnstructuredClient
from unstructured_client.models.operations import CreateJobRequest, DownloadJobOutputRequest
from unstructured_client.models.shared import BodyCreateJob, InputFiles

INPUT_DIR = "/full/path/to/your/input/directory"
OUTPUT_DIR = "/full/path/to/your/output/directory"

client = UnstructuredClient(
    api_key_auth=os.getenv("UNSTRUCTURED_API_KEY"),
    server_url=os.getenv("UNSTRUCTURED_API_URL")
)

# Step 1: Create the job.
input_files = []
for filename in os.listdir(INPUT_DIR):
    full_path = os.path.join(INPUT_DIR, filename)
    if not os.path.isfile(full_path):
        continue
    content_type, _ = mimetypes.guess_type(full_path)
    input_files.append(
        InputFiles(
            content=open(full_path, "rb"),
            file_name=filename,
            content_type=content_type or "application/octet-stream"
        )
    )

response = client.jobs.create_job(
    request=CreateJobRequest(
        body_create_job=BodyCreateJob(
            request_data=json.dumps({
                "job_nodes": [
                    {
                        "name": "Partitioner",
                        "type": "partition",
                        "subtype": "vlm",
                        "settings": {
                            "is_dynamic": True,
                            "allow_fast": True
                        }
                    }
                ]
            }),
            input_files=input_files
        )
    )
)

job_id = response.job_information.id
print(f"Job ID: {job_id}")

# Step 2: Poll until the job completes.
while True:
    response = client.jobs.get_job(request={"job_id": job_id})
    job_info = response.job_information
    status = job_info.status

    print(f"Job status: {status.value}")

    if status == "COMPLETED":
        print("Job completed.")
        break
    elif status in ("FAILED", "STOPPED"):
        raise RuntimeError(f"Job did not complete successfully: {status}")

    time.sleep(10)

output_node_file_ids = [f.file_id for f in (job_info.output_node_files or [])]

# Step 3: Download the job output.
os.makedirs(OUTPUT_DIR, exist_ok=True)

for file_id in output_node_file_ids:
    response = client.jobs.download_job_output(
        request=DownloadJobOutputRequest(job_id=job_id, file_id=file_id)
    )
    output_path = os.path.join(OUTPUT_DIR, f"{file_id}.json")
    with open(output_path, "w") as f:
        json.dump(response.any, f, indent=4)
    print(f"Saved: {output_path}")

What’s next?

Create local-file processing jobs that add enrichments, add embeddings, or perform extractions.
Learn about the Unstructured API.

Developer guide

API reference

Troubleshooting

Complete end-to-end script

What’s next?

​Complete end-to-end script

​What’s next?

Complete end-to-end script

What’s next?