Expose a Crawler as a Job API with Progress and Cancellation

A synchronous HTTP wrapper leaves callers waiting for a crawler and obscures its progress. A job API should return identity first, then support status, results, and cancellation.

This tutorial wraps a bounded documentation-title collector. Its state lives in one process's memory: it is not durable scheduling or restart recovery.

Prepare the input addresses

Locate the following pages in your browser, confirm their contents, and copy their full addresses. Obtain service and repository addresses from account integration settings, official setup instructions, or the repository clone menu. Do not derive real addresses from illustrative output.

DOC_URL_1: the Python tutorial introduction.html address.
DOC_URL_2: the controlflow.html address in the same directory.

Set these in the same Bash/WSL terminal used for subsequent commands:

export DOC_URL_1='<paste-the-actual-doc-url-1>'
export DOC_URL_2='<paste-the-actual-doc-url-2>'

Replace each <VARIABLE_NAME> in JSON or task parameters with the copied value before submission; JSON does not expand environment variables. Any example.16yun.cn source in an output illustration shows field structure, not a demo service deployed for this article.

Install and restrict sources

Use Python 3.12:

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install fastapi uvicorn httpx beautifulsoup4
export CRAWL_API_TOKEN='replace-with-your-local-api-token'

The example permits only HTTPS Python tutorial addresses and disables redirects. A real service needs business-specific source controls rather than accepting any parseable URL.

Implement the complete service

Create app.py:

import asyncio
import hashlib
import json
import logging
import os
import secrets
import uuid
from contextlib import asynccontextmanager
from urllib.parse import urlsplit
import httpx
from bs4 import BeautifulSoup
from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, HttpUrl
 
TOKEN = os.environ["CRAWL_API_TOKEN"]
jobs, tasks, keys = {}, {}, {}
slots = asyncio.Semaphore(2)
logger = logging.getLogger(__name__)
 
async def authorize(x_api_key: str = Header()):
    if not secrets.compare_digest(x_api_key, TOKEN):
        raise HTTPException(401, "invalid token")
 
@asynccontextmanager
async def lifespan(app):
    yield
    for task in tasks.values():
        if not task.done():
            task.cancel()
    await asyncio.gather(*tasks.values(), return_exceptions=True)
 
app = FastAPI(lifespan=lifespan, dependencies=[Depends(authorize)])
 
class Submission(BaseModel):
    urls: list[HttpUrl] = Field(min_length=1, max_length=10)
 
def get_job(job_id):
    if job_id not in jobs:
        raise HTTPException(404, "unknown job")
    return jobs[job_id]
 
async def run_job(job_id, urls):
    job = jobs[job_id]
    try:
        async with slots:
            job["state"] = "running"
            async with httpx.AsyncClient(timeout=20, follow_redirects=False) as client:
                for url in urls:
                    try:
                        response = await client.get(url)
                        response.raise_for_status()
                        soup = BeautifulSoup(response.text, "html.parser")
                        title = soup.title.get_text(strip=True) if soup.title else ""
                        if not title:
                            raise ValueError("missing title")
                        job["records"].append({"source_url": url, "title": title})
                    except (httpx.HTTPError, ValueError) as error:
                        job["errors"].append({"source_url": url, "error_type": type(error).__name__})
                    job["processed"] += 1
            job["state"] = "completed_with_errors" if job["errors"] else "completed"
    except asyncio.CancelledError:
        job["state"] = "cancelled"
        raise
    except Exception:
        job["state"] = "failed"
        logger.exception("unexpected job failure")
        raise
 
@app.post("/jobs", status_code=202)
async def submit(body: Submission, idempotency_key: str = Header(min_length=1, max_length=80)):
    urls = list(dict.fromkeys(str(value) for value in body.urls))
    for url in urls:
        parsed = urlsplit(url)
        if (parsed.scheme != "https" or parsed.hostname != "docs.python.org"
            or parsed.port not in {None, 443}
            or not parsed.path.startswith("/3/tutorial/")
            or parsed.username is not None):
            raise HTTPException(400, "source is outside allowed scope")
    digest = hashlib.sha256(json.dumps(urls).encode()).hexdigest()
    if idempotency_key in keys:
        job_id, old_digest = keys[idempotency_key]
        if old_digest != digest:
            raise HTTPException(409, "key already belongs to different inputs")
        return {"job_id": job_id, "state": jobs[job_id]["state"]}
    if len(jobs) >= 100:
        raise HTTPException(503, "teaching instance capacity reached")
    job_id = uuid.uuid4().hex
    jobs[job_id] = {"state": "queued", "total": len(urls), "processed": 0,
                    "records": [], "errors": []}
    keys[idempotency_key] = (job_id, digest)
    tasks[job_id] = asyncio.create_task(run_job(job_id, urls))
    return {"job_id": job_id, "state": "queued"}
 
@app.get("/jobs/{job_id}")
async def progress(job_id: str):
    job = get_job(job_id)
    return {**{key: job[key] for key in ("state", "total", "processed")},
            "accepted": len(job["records"]), "failed": len(job["errors"])}
 
@app.get("/jobs/{job_id}/result")
async def result(job_id: str):
    job = get_job(job_id)
    if job["state"] in {"queued", "running", "cancelling"}:
        raise HTTPException(409, "job is not terminal")
    return JSONResponse(job, headers={
        "Content-Disposition": f'attachment; filename="{job_id}.json"'
    })
 
@app.post("/jobs/{job_id}/cancel")
async def cancel(job_id: str):
    job = get_job(job_id)
    task = tasks[job_id]
    if not task.done():
        job["state"] = "cancelling"
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            job["state"] = "cancelled"
    return {"job_id": job_id, "state": job["state"]}

Start one worker:

uvicorn app:app --host 127.0.0.1 --port 8008 --workers 1

Multiple workers cannot share these dictionaries and may report one another's tasks as unknown. Durable or distributed operation needs external state and coordination.

Submit, inspect, and retrieve

Configure the same token and both DOC_URL_1 and DOC_URL_2 in another terminal:

python -c 'import json,os; print(json.dumps({"urls":[os.environ["DOC_URL_1"],os.environ["DOC_URL_2"]]}))' > submission.json
curl --fail-with-body -X POST http://localhost:8008/jobs   -H "X-API-Key: $CRAWL_API_TOKEN"   -H 'Idempotency-Key: tutorial-batch-1'   -H 'Content-Type: application/json'   -d @submission.json   > submitted.json

Read the returned identity and query it:

CRAWL_JOB_ID=$(python -c 'import json;print(json.load(open("submitted.json"))["job_id"])')
curl --fail-with-body -H "X-API-Key: $CRAWL_API_TOKEN"   "http://localhost:8008/jobs/$CRAWL_JOB_ID"
curl --fail-with-body -H "X-API-Key: $CRAWL_API_TOKEN"   "http://localhost:8008/jobs/$CRAWL_JOB_ID/result" -o result.json

A nonterminal result request returns 409. Poll status with a bounded, deliberate interval and retain the ID for continuation.

Processed counts resolved inputs, not accepted rows. A completed-with-errors state is not full success.

Interpret idempotency and cancellation

An identical key and input return the same task. A reused key with different input conflicts. This holds only while the process retains its state.

Cancel with:

curl --fail-with-body -X POST   -H "X-API-Key: $CRAWL_API_TOKEN"   "http://localhost:8008/jobs/$CRAWL_JOB_ID/cancel"

A short job may already be terminal. Queued work can also cancel, including before its coroutine starts. The endpoint waits for cancellation handling rather than merely setting a label.

Previously completed records remain available as partial cancelled output. Cancellation does not undo external side effects. This collector only reads; write-oriented jobs need separate compensation rules. See asyncio cancellation.

Adapt an existing collector

Replace one input's collection inside run_job while preserving outcome accounting, progress meaning, and exception behavior.

Blocking libraries need an appropriate execution strategy, and browser-backed tasks need task-owned cleanup.

FastAPI's ordinary background-task facility does not automatically supply durable identity, status, and cancellation. See the official background-task guidance. Explicit Tasks here demonstrate the protocol rather than a production scheduler.

Troubleshoot

A 202 job is immediately complete: clients must support that valid transition without requiring an observed running state.

Cancelled output contains records: those are completed partial results, not cancellation failure.

IDs vanish after restart: in-memory state was lost; add persistence before promising recovery.

A reused key returns old work: use a new key for a new business batch and the same key only for retrying one submission.

A source is rejected: inspect the allowed range rather than removing all controls.

Progress stalls: distinguish request timeout, unexpected task error, and blocking execution through actual state and server diagnostics.

A job API needs clear identity, progress, result, and cancellation semantics before persistence or more workers can improve it.