Parse Documents Concurrently

Parse many documents concurrently by submitting each one as a Parse Job and polling all of them at the same time. Use this pattern in applications and services that already run asynchronous code, or when you have more documents than you want to wait on one at a time. For a plain script that works through a folder in order, see Parse a Directory of Documents. The scripts save each document’s Markdown to an output folder as <name>_<extension>.md (such as invoice_pdf.md), so files that share a name, like invoice.pdf and invoice.png, don’t overwrite each other. A Parse Job returns its job_id as soon as the upload finishes, and the server queues the parsing. Concurrency in your code controls how many documents upload and poll at the same time, not how fast the server parses them. The scripts submit jobs on the standard tier, which consumes half the credits of the priority tier and of synchronous requests, with a slower turnaround. standard jobs count against your plan’s hourly page limit. When you exceed a rate limit, the API returns a 429 error. See Rate Limits.

These examples require the Python or TypeScript client library. Before running a script, set your API key and install the library and any required dependencies.

Python

Use AsyncLandingAIADE with asyncio.gather() and a semaphore

TypeScript

Use Promise.allSettled() with p-limit

Python

The AsyncLandingAIADE client exposes the same client.v2.parse_jobs methods as the synchronous client, as coroutines. The script runs one task per document with asyncio.gather(), and an asyncio.Semaphore caps how many uploads run at the same time. A failed document is reported at the end and doesn’t stop the others.

import asyncio
from pathlib import Path
from landingai_ade import AsyncLandingAIADE

# Initialize client (uses the API key from the VISION_AGENT_API_KEY environment variable)
client = AsyncLandingAIADE()

# Limit how many documents upload at the same time (adjust as needed)
upload_slots = asyncio.Semaphore(5)


async def parse_document(filepath: Path, output_folder: Path) -> None:
    # Submit the Parse Job on the standard tier
    async with upload_slots:
        created = await client.v2.parse_jobs.create(
            document=filepath,
            model="dpt-3-pro-latest",
            service_tier="standard",
        )
    print(f"Submitted {filepath.name}: {created.job_id}")

    # Poll until the job finishes, for up to an hour (timeout is in seconds)
    job = await client.v2.parse_jobs.wait(created.job_id, timeout=3600)
    if job.status != "completed":
        raise RuntimeError(f"job {job.job_id} ended {job.status}: {job.error.message if job.error else ''}")

    # Save Markdown output (useful if you plan to run Extract on the Markdown)
    # Keep the extension in the output name, so invoice.pdf and invoice.png don't overwrite each other
    output_name = f"{filepath.stem}_{filepath.suffix[1:]}"
    (output_folder / f"{output_name}.md").write_text(job.result.markdown, encoding="utf-8")
    print(f"Parsed {filepath.name}")


async def main() -> None:
    # Replace "data/" with your folder path
    data_folder = Path("data/")
    output_folder = Path("output/")
    output_folder.mkdir(exist_ok=True)

    # File types the Parse API accepts (adjust as needed for your use case)
    supported_types = {
        ".pdf", ".png", ".jpg", ".jpeg", ".gif", ".bmp", ".tif", ".tiff", ".webp",
        ".docx", ".odt", ".rtf", ".pptx", ".xlsx", ".csv",
    }
    # Skip folders and file types the Parse API doesn't accept
    files = [p for p in sorted(data_folder.iterdir()) if p.is_file() and p.suffix.lower() in supported_types]

    # Parse all documents concurrently, collecting errors instead of stopping at the first one
    results = await asyncio.gather(*(parse_document(p, output_folder) for p in files), return_exceptions=True)

    failed = [(p, r) for p, r in zip(files, results) if isinstance(r, Exception)]
    for filepath, error in failed:
        print(f"Failed to parse {filepath.name}: {error}")
    print(f"\nCompleted: {len(files) - len(failed)} successful, {len(failed)} failed")


asyncio.run(main())

TypeScript

The TypeScript library’s methods already return promises. The script starts one task per document and waits for all of them with Promise.allSettled(), so a failed document doesn’t stop the others. The p-limit library caps how many uploads run at the same time:

npm install p-limit

import LandingAIADE from "landingai-ade";
import fs from "fs";
import path from "path";
import pLimit from "p-limit";

// Initialize client (uses the API key from the VISION_AGENT_API_KEY environment variable)
const client = new LandingAIADE();

// Limit how many documents upload at the same time (adjust as needed)
const limit = pLimit(5);

async function parseDocument(filepath: string, outputFolder: string): Promise<void> {
  // Submit the Parse Job on the standard tier
  const created = await limit(() =>
    client.v2.parseJobs.create({
      document: fs.createReadStream(filepath),
      model: "dpt-3-pro-latest",
      service_tier: "standard",
    }),
  );
  console.log(`Submitted ${path.basename(filepath)}: ${created.job_id}`);

  // Poll until the job finishes, for up to an hour (timeout is in milliseconds)
  const job = await client.v2.parseJobs.wait(created.job_id, { timeout: 3_600_000 });
  if (job.status !== "completed") {
    throw new Error(`job ${job.job_id} ended ${job.status}: ${job.error?.message ?? ""}`);
  }

  // Save Markdown output (useful if you plan to run Extract on the Markdown)
  const result = job.result as LandingAIADE.V2ParseResponse;
  // Keep the extension in the output name, so invoice.pdf and invoice.png don't overwrite each other
  const extension = path.extname(filepath);
  const outputName = `${path.basename(filepath, extension)}_${extension.slice(1)}`;
  fs.writeFileSync(path.join(outputFolder, `${outputName}.md`), result.markdown ?? "", "utf-8");
  console.log(`Parsed ${path.basename(filepath)}`);
}

async function main(): Promise<void> {
  // Replace with your folder path containing documents to parse
  const dataFolder = "data/";
  const outputFolder = "output/";
  fs.mkdirSync(outputFolder, { recursive: true });

  // File types the Parse API accepts (adjust as needed for your use case)
  const supportedTypes = new Set([
    ".pdf", ".png", ".jpg", ".jpeg", ".gif", ".bmp", ".tif", ".tiff", ".webp",
    ".docx", ".odt", ".rtf", ".pptx", ".xlsx", ".csv",
  ]);
  // Skip folders and file types the Parse API doesn't accept
  const files = fs
    .readdirSync(dataFolder)
    .sort()
    .filter((filename) => supportedTypes.has(path.extname(filename).toLowerCase()))
    .map((filename) => path.join(dataFolder, filename))
    .filter((filepath) => fs.statSync(filepath).isFile());

  // Parse all documents concurrently, collecting errors instead of stopping at the first one
  const results = await Promise.allSettled(files.map((filepath) => parseDocument(filepath, outputFolder)));

  let failed = 0;
  results.forEach((result, i) => {
    if (result.status === "rejected") {
      failed++;
      console.error(`Failed to parse ${path.basename(files[i])}: ${result.reason}`);
    }
  });
  console.log(`\nCompleted: ${files.length - failed} successful, ${failed} failed`);
}

main();

Overview

Parse Only

Parse & Extract

Parse Documents Concurrently

Python

TypeScript

Python

TypeScript

Python

TypeScript

​Python

​TypeScript

Python

TypeScript