Async Batch Processing

Async batch processing is the orchestration layer that lets a lease abstraction system swallow a whole portfolio at once instead of one document at a time. It sits inside Parsing & Extraction Workflows, directly downstream of document acquisition and directly upstream of canonical storage: raw files land in a staging layer, a dispatcher fans them out across isolated workers, each worker runs extraction under a strict resource ceiling, and validated records stream into the property database. The specific challenge this page solves is throughput without collapse — running thousands of OCR-heavy, NLP-heavy extractions concurrently while preventing the out-of-memory crashes, event-loop stalls, and duplicate writes that destroy a rent roll’s integrity.

Executing lease extractions synchronously degrades API latency, exhausts worker memory, and turns an overnight job into a multi-day backlog during peak leasing season. The pattern below decouples ingestion from extraction so a portfolio manager can queue a multi-year set of leases and have structured abstractions ready by morning. Everything here is the engineering of that decoupling: a bounded-concurrency controller, a validation gate that refuses to let malformed data reach the books, and a clean migration path to distributed workers once a single node is no longer enough.

Where this fits in the pipeline

Keeping the batch layer’s responsibilities narrow is what makes a high-volume run auditable:

• Upstream of it: the PDF/DOCX ingestion pipelines that normalize scanned leases, convert embedded tables, and strip proprietary formatting, plus the OCR preprocessing workflows that deskew and clean scanned pages before any text is read.
• This stage: validate file integrity, queue tasks to a broker, pull them under a concurrency cap, run extraction inside an isolated worker, and emit a per-document result carrying a status and a confidence score.
• Downstream of it: validated abstractions feed the canonical lease data models, while low-confidence or malformed results divert through fallback routing logic toward a human review queue rather than being silently committed.

A batch controller that also tries to classify clauses or resolve amendment precedence becomes impossible to reason about under load. It answers one question — “given this file, did extraction produce a valid abstraction, and where does the result go?” — and hands everything else off.

Prerequisites and environment setup

The reference implementation targets Python 3.11+ and a deliberately small dependency set so the controller stays fast and vendorable into a worker image.

Install with pip install "pydantic>=2.6" aioboto3. Two environment assumptions matter. First, every monetary field enters as a typed Decimal or float that the pydantic gate range-checks before it reaches the property management system; raw extraction strings are never committed directly. Second, CPU-bound clause parsing must be wrapped so it does not block the event loop — heavy regex or transformer inference belongs in a worker thread or process pool, while the async layer owns only the I/O waits (storage fetch, OCR API, database commit). Conflating the two is what turns a “concurrent” pipeline into a serialized one that merely looks asynchronous.

Pipeline architecture

The batch lifecycle is event-driven and strictly staged. Each stage has one job, emits structured telemetry, and can be retried or dead-lettered independently of its neighbors.

Primary implementation

Below is a production-grade controller using asyncio, pydantic, and a semaphore-based concurrency limiter. The pattern isolates I/O-bound network calls from CPU-bound parsing while preserving strict error boundaries for property data integrity. The semaphore is the linchpin: by capping active tasks at max_concurrency, you align execution with the available CPU and memory ceiling and prevent the out-of-memory crashes that plague unbounded asyncio.gather over a large portfolio.

import asyncio
import logging
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, Field, ValidationError

# Configure structured logging for production observability
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s"
)
logger = logging.getLogger(__name__)

class LeaseAbstraction(BaseModel):
    property_id: str
    tenant_name: str
    lease_start: datetime
    lease_end: datetime
    base_rent: float
    cam_charges: Optional[float] = None
    extraction_confidence: float = Field(ge=0.0, le=1.0)

class ProcessingResult(BaseModel):
    lease_id: str
    status: str
    payload: Optional[LeaseAbstraction] = None
    error: Optional[str] = None

# ---------------------------------------------------------------------------
# Simulated External Dependencies (Replace with actual S3, OCR, and PMS APIs)
# ---------------------------------------------------------------------------
async def fetch_and_parse_document(file_path: str) -> str:
    """Simulates async I/O: object storage fetch, OCR, and text extraction."""
    await asyncio.sleep(0.1)
    return f"Raw lease text extracted from {file_path}"

async def run_clause_extraction(raw_text: str) -> dict:
    """Simulates CPU/NLP-bound clause parsing and field mapping."""
    await asyncio.sleep(0.05)
    return {
        "lease_id": "L-1001",
        "property_id": "PROP-55",
        "tenant_name": "Acme Commercial LLC",
        "lease_start": "2023-01-01T00:00:00",
        "lease_end": "2028-12-31T00:00:00",
        "base_rent": 15000.00,
        "cam_charges": 2500.00,
        "extraction_confidence": 0.94
    }

async def commit_to_pms(result: ProcessingResult) -> None:
    """Simulates idempotent write to Yardi, RealPage, or custom PMS."""
    await asyncio.sleep(0.02)
    if result.status == "success":
        logger.info("Successfully committed %s to PMS", result.lease_id)
    else:
        logger.warning("Failed to commit %s to PMS: %s", result.lease_id, result.error)

# ---------------------------------------------------------------------------
# Core Async Batch Controller
# ---------------------------------------------------------------------------
async def process_single_lease(file_path: str, semaphore: asyncio.Semaphore) -> ProcessingResult:
    async with semaphore:
        try:
            raw_text = await fetch_and_parse_document(file_path)
            extracted = await run_clause_extraction(raw_text)

            # Strict pydantic validation prevents malformed data from hitting the PMS
            abstraction = LeaseAbstraction(**extracted)
            return ProcessingResult(lease_id=extracted["lease_id"], status="success", payload=abstraction)

        except ValidationError as e:
            logger.error("Schema validation failed for %s: %s", file_path, e)
            return ProcessingResult(lease_id="unknown", status="validation_error", error=str(e))
        except Exception as e:
            logger.exception("Unexpected error processing %s", file_path)
            return ProcessingResult(lease_id="unknown", status="processing_error", error=str(e))

async def run_batch_pipeline(file_paths: List[str], max_concurrency: int = 10) -> List[ProcessingResult]:
    semaphore = asyncio.Semaphore(max_concurrency)
    tasks = [process_single_lease(fp, semaphore) for fp in file_paths]

    # Gather results without failing the entire batch on a single exception
    raw_results = await asyncio.gather(*tasks, return_exceptions=True)

    results = []
    for r in raw_results:
        if isinstance(r, Exception):
            results.append(ProcessingResult(lease_id="unknown", status="gather_error", error=str(r)))
        else:
            results.append(r)

    # Commit successful extractions to the Property Management System
    for res in results:
        await commit_to_pms(res)

    return results

# ---------------------------------------------------------------------------
# Execution Entry Point
# ---------------------------------------------------------------------------
async def main():
    lease_files = [
        "s3://leases-bucket/2023/lease_001.pdf",
        "s3://leases-bucket/2023/lease_002.docx",
        "s3://leases-bucket/2023/lease_003.pdf"
    ]
    results = await run_batch_pipeline(lease_files, max_concurrency=5)
    success_count = sum(1 for r in results if r.status == "success")
    logger.info("Batch complete: %d/%d leases processed successfully", success_count, len(results))

if __name__ == "__main__":
    asyncio.run(main())

Within each worker, regex & NLP clause extraction logic runs synchronously but is wrapped in an async context, ensuring the event loop stays unblocked while waiting on external OCR services or vector-database queries. For genuinely CPU-heavy parsing, replace the inline call with await loop.run_in_executor(pool, run_clause_extraction, raw_text) so transformer inference executes off the event-loop thread.

Validation and quality gates

Data integrity is enforced at the schema boundary, not after the fact. The pydantic model acts as a strict gate that rejects malformed dates, out-of-range financial figures, and missing mandatory fields before they reach the property database. Three gates work together:

• Schema enforcement. LeaseAbstraction coerces and range-checks every field. A base_rent that arrives as "$15,000/mo" or an extraction_confidence above 1.0 raises a ValidationError that routes the document to validation_error instead of the PMS. This is the same typed-decimal contract that metadata normalization standards require at the canonical boundary, applied early so bad data never travels.
• Confidence scoring. Every result carries an aggregate extraction_confidence. Documents below the review threshold should never auto-commit; instead they divert to the human review queue via fallback routing. A sensible default is 0.80, tuned against a labeled holdout — high enough that a wrong base rent never silently posts money.
• Idempotent commits. Because retries and re-uploads guarantee a document will be processed more than once, the PMS write must be keyed on the document content hash with a unique constraint, so replaying a task is a no-op rather than a double-count.

Failures that are not recoverable — a fundamentally unparseable file, a persistent schema mismatch — belong in a dead-letter queue with their original payload, stack trace, and a diagnostic summary, so operations teams can triage without halting the batch. The classification and backoff machinery for that lives in error handling & retry logic.

Troubleshooting

Performance and scale notes

Batch sizing is a function of three numbers: per-task peak memory, per-task wall-clock latency, and node memory. Set max_concurrency from the memory budget, then size the batch (the number of files handed to one run_batch_pipeline call) so a single failure window is recoverable — a few thousand documents per batch keeps checkpointing cheap without flooding the broker. Stream results as they complete with asyncio.as_completed rather than blocking on a full gather when you want partial progress to commit incrementally.

A single-node asyncio pipeline has a ceiling: one machine’s CPU and memory. When portfolio volumes exceed tens of thousands of documents, or you need work to survive a worker restart, the same bounded-concurrency and validation semantics migrate onto a distributed task queue across many worker nodes — preserving the per-task cap, the confidence gate, and idempotent commits while adding broker-level retries and dead-letter queues. That evolution, including Redis broker configuration, idempotent task routing, and distributed rate limiting, is covered in depth in Scaling Async Lease Parsing Pipelines with Celery and Redis.

Frequently asked questions

What confidence threshold should trigger manual review? Start at 0.80 for the aggregate extraction score and tune against a labeled holdout. Because a wrong base rent or escalation posts real money, the batch gate should hold this threshold higher than a pure classification step would, and any document that fails schema validation should divert to review regardless of its score.

How do I handle lease amendments that override base clauses in a batch? Process the amendment as its own document with its own hash, then resolve precedence downstream in the lease data models using effective dates — never let the batch layer overwrite a base lease in place. Appending amendment versions keeps the audit trail and lets you reconstruct what was billed in any period.

Why not just use asyncio.gather over the whole portfolio? Unbounded gather schedules every coroutine at once, so memory scales with portfolio size and a 30,000-lease run exhausts the node. The semaphore caps the live working set so memory stays flat regardless of how many files are queued.

Where do non-recoverable failures go? To a dead-letter queue carrying the original payload, stack trace, and a diagnostic summary, so operations can triage offline. The batch never blocks on them, and they never reach the property management system.

Related

• Scaling Async Lease Parsing Pipelines with Celery and Redis — distributing this controller across worker nodes with a Redis broker and idempotent task routing.
• Error Handling & Retry Logic — exponential backoff, failure classification, and dead-letter routing for the per-task failures this pipeline emits.
• PDF/DOCX Ingestion Pipelines — the upstream stage that normalizes raw lease files before they enter the batch queue.
• Field Mapping Strategies — turning the validated abstractions this stage produces into canonical schema fields.
• Fallback Routing Logic — where low-confidence and invalid results divert for human review instead of auto-committing.

← Back to Parsing & Extraction Workflows

← Back to Parsing & Extraction Workflows