Build a Python Async Scraping Pipeline with Backpressure and Continuous Output

Creating thousands of tasks and keeping every response until gather() finishes combines two problems: downloads can outrun parsing, and completed results remain in memory. A pipeline separates downloading, parsing, and writing, using bounded queues between stages.

This tutorial processes three Python documentation pages. Each resolved input produces one JSONL record containing either a page title or a specific failure. No browser, database, or model service is required. Understand how the stages finish before replacing the small input list with your own URLs.

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.
DOC_URL_3: the datastructures.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>'
export DOC_URL_3='<paste-the-actual-doc-url-3>'

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.

Assign a clear responsibility to each stage

URL 清单 → 有界任务队列 → 4 个下载协程
                       ↓
                  有界页面队列 → 1 个解析协程
                                      ↓
                                 有界结果队列 → 1 个文件写入协程

Four download workers share one HTTP client. When a queue fills, its producer waits at await queue.put(). That is backpressure: slower downstream work limits further upstream production.

The example limits concurrent downloads, response size, and queued pages. Queue capacity alone does not limit the size of one response or the memory occupied by the input list. For a large URL file, change the producer to read it incrementally.

Normal completion and failure use different mechanisms. Normal completion sends a sentinel through each stage. An internal exception lets TaskGroup cancel sibling tasks, preventing consumers from waiting forever for a sentinel that can no longer arrive.

Install dependencies and choose a fresh output file

Use Python 3.12 in a new directory:

python -m venv .venv
source .venv/bin/activate
python -m pip install "httpx==0.28.1" "beautifulsoup4==4.13.4"

In Windows PowerShell, activate with .venv\Scripts\Activate.ps1. Save the complete program below as pipeline.py. Exclusive file creation deliberately rejects an existing output file so that separate batches cannot become mixed.

Implement the complete pipeline

import asyncio
import os
import json
from pathlib import Path
 
import httpx
from bs4 import BeautifulSoup
 
URLS = [
    os.environ["DOC_URL_1"],
    os.environ["DOC_URL_2"],
    os.environ["DOC_URL_3"],
]
WORKERS = 4
MAX_BYTES = 2 * 1024 * 1024
OUTPUT = Path("pages.jsonl")
 
 
class PageTooLarge(Exception):
    pass
 
 
async def produce(jobs):
    for url in URLS:
        await jobs.put(url)
    for _ in range(WORKERS):
        await jobs.put(None)
 
 
async def fetch(client, jobs, pages):
    while True:
        url = await jobs.get()
        if url is None:
            return
        try:
            async with client.stream("GET", url) as response:
                response.raise_for_status()
                media = response.headers.get("content-type", "").lower()
                if "text/html" not in media:
                    raise ValueError("unexpected_content_type")
                body = bytearray()
                async for chunk in response.aiter_bytes():
                    if len(body) + len(chunk) > MAX_BYTES:
                        raise PageTooLarge("response_exceeds_2_mib")
                    body.extend(chunk)
                record = {
                    "url": str(response.url),
                    "requested_url": url,
                    "status": "downloaded",
                    "body": bytes(body),
                    "encoding": response.encoding,
                }
        except (httpx.HTTPError, PageTooLarge, ValueError) as exc:
            record = {
                "url": url,
                "requested_url": url,
                "status": "failed",
                "stage": "download",
                "error_type": type(exc).__name__,
                "error": str(exc),
            }
        await pages.put(record)
 
 
async def parse(pages, results):
    while True:
        record = await pages.get()
        if record is None:
            await results.put(None)
            return
        if record["status"] == "failed":
            await results.put(record)
            continue
        soup = BeautifulSoup(
            record.pop("body"),
            "html.parser",
            from_encoding=record.pop("encoding"),
        )
        title = soup.title.get_text(" ", strip=True) if soup.title else ""
        if not title:
            record.update(
                status="failed", stage="parse",
                error_type="MissingTitle", error="title_is_empty",
            )
        else:
            record.update(status="ok", title=title)
        await results.put(record)
 
 
async def write(results):
    written = 0
    with OUTPUT.open("x", encoding="utf-8", newline="\n") as handle:
        try:
            while True:
                record = await results.get()
                if record is None:
                    break
                handle.write(json.dumps(record, ensure_ascii=False) + "\n")
                written += 1
                if written % 10 == 0:
                    handle.flush()
        finally:
            handle.flush()
            print("written_records=" + str(written))
 
 
async def finish_downloads(tasks, pages):
    await asyncio.gather(*tasks)
    await pages.put(None)
 
 
async def main():
    jobs = asyncio.Queue(maxsize=16)
    pages = asyncio.Queue(maxsize=8)
    results = asyncio.Queue(maxsize=16)
    limits = httpx.Limits(
        max_connections=WORKERS, max_keepalive_connections=WORKERS
    )
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(20.0),
        limits=limits,
        follow_redirects=True,
        headers={"User-Agent": "DocumentPipelineTutorial/1.0"},
    ) as client:
        async with asyncio.TaskGroup() as group:
            group.create_task(produce(jobs))
            downloads = [
                group.create_task(fetch(client, jobs, pages))
                for _ in range(WORKERS)
            ]
            group.create_task(parse(pages, results))
            group.create_task(write(results))
            group.create_task(finish_downloads(downloads, pages))
 
 
if __name__ == "__main__":
    asyncio.run(main())

Run:

python pipeline.py

There is no queue.join(): the producer and sentinels define completion. A coordinator waits for all four download tasks before sending one page sentinel. The parser processes preceding pages, then notifies the single writer. Sending a parser sentinel from each individual downloader could close the downstream stage while other downloaders still have results.

Parsing runs on the event loop here and suits short pages with a response-size limit. Expensive cleanup or CPU-heavy transformations need a separate bounded processing pool. An async function does not make CPU work parallel.

Interpret and check the output

This is an illustrative record; the actual title comes from the page:

{"url":"https://example.16yun.cn/3/tutorial/introduction.html","requested_url":"https://example.16yun.cn/3/tutorial/introduction.html","status":"ok","title":"An Informal Introduction to Python — Python documentation"}

A network failure also produces a record, with status=failed and stage=download, instead of silently reducing the row count. Three rows are expected only when the run finishes normally and resolves all three inputs. Three rows do not imply three successes.

Inspect the file rather than relying solely on the printed write count:

import json
from collections import Counter
 
with open("pages.jsonl", encoding="utf-8") as handle:
    rows = [json.loads(line) for line in handle]
print("records:", len(rows))
print("statuses:", dict(Counter(row["status"] for row in rows)))
print("inputs:", len({row["requested_url"] for row in rows}))
for row in rows:
    if row["status"] != "ok":
        print(row["requested_url"], row["stage"], row["error_type"])

Archive the file before starting another batch. This program does not treat an old file as a checkpoint. To implement recovery, subtract successfully processed requested_url values from the input list; failed records must remain eligible for retry.

Distinguish page failures from broken processing

HTTP errors, timeouts, oversized responses, and unexpected content types belong to individual URLs. Record them and let other pages continue.

Misspelled internal fields, parser defects, and disk write failures must not masquerade as ordinary webpage failures. These exceptions escape the worker, allowing TaskGroup to cancel related tasks and fail the process. Otherwise, a run could claim completion after writing only part of its output. Do not add except Exception: pass to every stage.

Ctrl+C also enters the writer's finally block during cooperative cancellation. Records already written to the Python file buffer are flushed. Data still downloading or waiting in either queue may not reach the file. This is different from power-loss durability: flush() does not promise persistence through a power failure, nor an intact final line after a hard kill.

Troubleshoot and adjust deliberately

Symptom	Check first	Change and confirm
More concurrency makes the batch slower	Rate limiting, timeout records, connection count	Reduce WORKERS to 2 and compare success count and elapsed time for identical inputs
Memory keeps growing	Response sizes, fully loaded input lists, retained soup objects	Keep the response limit, stream the URL list, and avoid collecting all HTML before parsing
Slow writing pauses downloads	A full results queue or congested disk	This is backpressure working; inspect disk performance before increasing buffers
FileExistsError	Existing output file	Archive it or choose a new filename instead of appending unrelated batches
Stopping leaves fewer records than inputs	Cancellation and unconsumed queues	Compare successful requested_url values with inputs and retry unfinished work
ExceptionGroup appears	The first unexpected exception within the group	Fix that cause instead of catching the whole group and printing “complete”

The writer flushes every ten records and at normal exit. Time-based flushing requires the writer to manage a timer and its buffer together. If a record should count as complete only after durable confirmation, use transactional storage and update completion state after commit. More concurrency cannot replace a persistence contract.

See Python task groups and cancellation and asyncio queues for the underlying behavior.