How automation_hub scrapes and processes PDFs with Celery, Redis, and Gemini Schema contracts

1. Why Synchronous Scraping Breaks HTTP Event Loops

When an API endpoint tries to launch a headless Chromium browser via Playwright or parse a multi-megabyte PDF inside a synchronous HTTP request cycle, two failure modes appear immediately: event-loop starvation and silent upstream failures.

Launching Chromium and waiting for dynamic DOM hydration easily exceeds reverse-proxy timeouts when multiple requests arrive concurrently. Furthermore, external target sites and LLM APIs enforce rate limits. Without a persistent message broker and retry state machine, a transient 429 Too Many Requests drops the entire job.

In automation_hub, the HTTP layer never executes Playwright or PDF parsing directly. Instead, FastAPI acts strictly as a schema validation and task-dispatch gateway.

2. Non-Blocking FastAPI Ingestion & Redis Broker Queueing

Every incoming scrape or PDF extraction job is validated at the boundary using Pydantic v2 models before being pushed to Redis via Celery's apply_async. The endpoint immediately returns a 202 Accepted response with a deterministic task_id so the API event loop stays unblocked.

from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel, HttpUrl, Field
from app.worker.celery_app import process_document_pipeline

router = APIRouter(prefix="/api/v1/jobs", tags=["jobs"])

class ExtractionJobRequest(BaseModel):
    target_url: HttpUrl | None = None
    pdf_storage_key: str | None = Field(default=None, max_length=256)
    schema_contract: str = Field(default="invoice_v1", pattern=r"^[a-z0-9_]+$")

@router.post("/submit", status_code=status.HTTP_202_ACCEPTED)
async def submit_extraction_job(payload: ExtractionJobRequest):
    if not payload.target_url and not payload.pdf_storage_key:
        raise HTTPException(status_code=400, detail="Provide target_url or pdf_storage_key")

    async_result = process_document_pipeline.apply_async(
        kwargs=payload.model_dump(mode="json"),
        queue="extraction_high_priority",
    )
    return {
        "task_id": async_result.id,
        "status": "PENDING",
        "poll_url": f"/api/v1/jobs/{async_result.id}",
    }

3. Celery Worker Pools with Exponential Backoff & Structured Gemini Contracts

Inside the Celery worker container, tasks pull serialized payloads from Redis. Rather than asking Gemini to return free-form prose from scraped HTML or PDF pages, the worker enforces a strict JSON schema contract validated by Pydantic before persisting the result.

If an upstream target times out or returns a transient connection error, Celery automatically schedules an exponential backoff retry with jitter so workers do not thundering-herd the upstream service.

from celery import shared_task
from pydantic import BaseModel
from app.services.pdf_parser import extract_pdf_text_chunks
from app.services.gemini_extractor import extract_structured_json

class ExtractedDocumentSummary(BaseModel):
    document_title: str
    entities: list[str]
    key_findings: list[str]
    page_count: int

@shared_task(
    bind=True,
    autoretry_for=(TimeoutError, ConnectionError),
    retry_backoff=True,
    retry_jitter=True,
    max_retries=3,
)
def process_document_pipeline(self, target_url=None, pdf_storage_key=None, schema_contract="invoice_v1"):
    raw_chunks = extract_pdf_text_chunks(pdf_storage_key) if pdf_storage_key else []
    structured_payload = extract_structured_json(
        chunks=raw_chunks,
        response_schema=ExtractedDocumentSummary,
    )
    return structured_payload.model_dump()

4. Containerization, Kubernetes HPA & 68 Pytest Verification Tests

To keep the deployment reproducible across environments, automation_hub separates the stateless FastAPI web tier from the stateful Redis broker and Celery worker pods using multi-stage Dockerfiles and Kubernetes Horizontal Pod Autoscaler (k8s/hpa.yaml) manifests.

The entire pipeline is verified across 68 automated pytest unit and integration tests covering Pydantic v2 request validation, Redis queue serialization, worker retry transitions, and Kubernetes readiness/liveness probe endpoints.