Skip to content

Examples

This guide provides practical examples for using aiogzip in various scenarios.

The repository's maintained-example runbook states the Python/dependency requirements and gives exact clean-checkout and wheel-installed commands. Helpers, framing, staging, manifests, and status labels defined by those scripts are application code, not aiogzip API.

Maintained fragmented-transport example

The repository includes examples/fragmented_transport.py, a deterministic, credential-free demonstration of the public GzipEncoder, GzipDecoder, and CodecOperation APIs over a bounded local bidirectional asyncio byte transport. Its application-level two-byte length prefix makes gzip fragmentation explicit instead of incorrectly relying on TCP packet boundaries:

python examples/fragmented_transport.py
python examples/fragmented_transport.py --self-test

The sender exhausts start(), every feed(), every low-latency flush(), and finish() in order. If an operation is abandoned, it calls operation.close(); codec-wide discard() provides idempotent terminal cleanup. The receiver parses JSON Lines incrementally but labels records receiving-provisional until GzipDecoder.finish() proves the final trailer and changes the status to verified.

Calling flush() after each record improves visibility latency at the cost of compression efficiency. One long-lived member preserves compression context but has one final validation boundary; member-per-batch designs validate more often while paying for extra headers, trailers, and compression resets. The example illustrates these lifecycle and framing choices; it is not an official transport abstraction. The repository's examples/README.md has the focused scenario list.

Maintained concurrent JSONL ingest

examples/concurrent_jsonl_ingest.py generates ordinary independent gzip files with the standard library, processes them through bounded asyncio.TaskGroup concurrency, and publishes one dataset only after every shard validates:

python examples/concurrent_jsonl_ingest.py \
  --generate-fixtures ./demo-input \
  --output ./demo-published

One task owns each handle; separate files can overlap without sharing mutable gzip state. The hot path uses iter_batches() rather than one await per JSON line, parses every line, updates counts and digests incrementally, and writes decoded bytes to per-shard files on disk. It retains only one approximate-hint batch per active shard, not the complete dataset or an unbounded queue.

Decoded batches remain provisional until their gzip iterator reaches normal EOF. All output therefore stays in a unique sibling staging directory. After every shard succeeds, the example closes staged outputs, writes the complete manifest, and renames the directory to its final destination once. Corruption, truncation, per-shard or dataset-wide limits, invalid JSON, write failure, and cancellation remove staging and publish nothing.

Per-shard limits bound each gzip expansion independently. The dataset budget atomically counts exact bytes written across all staged outputs while holding its lock only for arithmetic. Structurally valid gzip can still fail the application contract when a decoded line is invalid JSON. These are ordinary independent files, not a custom row-striped format.

CSV Processing with aiocsv

aiogzip pairs perfectly with aiocsv for efficient, asynchronous CSV processing.

Reading a CSV File

import asyncio
import aiocsv
from aiogzip import AsyncGzipTextFile

async def read_csv():
    async with AsyncGzipTextFile("data.csv.gz", "rt", encoding="utf-8", newline="") as f:
        # Use AsyncDictReader for dictionary-based access
        async for row in aiocsv.AsyncDictReader(f):
            print(f"Name: {row['name']}, Age: {row['age']}")

asyncio.run(read_csv())

Writing a CSV File

import asyncio
import aiocsv
from aiogzip import AsyncGzipTextFile

async def write_csv():
    data = [
        {"name": "Alice", "age": 30, "city": "New York"},
        {"name": "Bob", "age": 25, "city": "Los Angeles"},
    ]

    async with AsyncGzipTextFile("output.csv.gz", "wt", encoding="utf-8", newline="") as f:
        writer = aiocsv.AsyncDictWriter(f, fieldnames=["name", "age", "city"])
        await writer.writeheader()
        for row in data:
            await writer.writerow(row)

asyncio.run(write_csv())

JSON Lines (JSONL) Processing

Processing large compressed JSONL files is a common task in data engineering.

Reading JSONL

import asyncio
import json
from aiogzip import AsyncGzipTextFile

async def process_jsonl():
    async with AsyncGzipTextFile("logs.jsonl.gz", "rt") as f:
        async for line in f:
            try:
                record = json.loads(line)
                if record.get("level") == "ERROR":
                    print(f"Error found: {record['message']}")
            except json.JSONDecodeError:
                print(f"Skipping invalid JSON: {line[:50]}...")

asyncio.run(process_jsonl())

Reading JSONL Efficiently

If your data is standard UTF-8 JSONL with \n line endings, configure the text reader explicitly:

import asyncio
import json
from aiogzip import AsyncGzipTextFile

async def process_jsonl_fast():
    async with AsyncGzipTextFile(
        "logs.jsonl.gz",
        "rt",
        newline="\n",
        chunk_size=512 * 1024,
    ) as f:
        async for line in f:
            record = json.loads(line)
            # Process record...

asyncio.run(process_jsonl_fast())

Why this helps:

  • newline="\n" skips universal-newline handling that JSONL does not need.
  • A larger chunk_size reduces async read overhead during line iteration.

This is a good default for large gzipped JSONL files. If memory pressure matters more than throughput, lower the chunk size.

Concurrent File Processing

One of the biggest advantages of aiogzip is the ability to process multiple files concurrently without blocking the event loop.

import asyncio
from aiogzip import AsyncGzipTextFile

async def process_file(filename):
    print(f"Starting {filename}...")
    line_count = 0
    async with AsyncGzipTextFile(filename, "rt") as f:
        async for _ in f:
            line_count += 1
    print(f"Finished {filename}: {line_count} lines")
    return line_count

async def main():
    files = ["data1.gz", "data2.gz", "data3.gz"]

    # Create tasks for all files
    tasks = [process_file(f) for f in files]

    # Run them concurrently
    results = await asyncio.gather(*tasks)

    print(f"Total lines processed: {sum(results)}")

# To run this example, ensure the files exist
# asyncio.run(main())

Error Handling

aiogzip raises standard OSError (or subclasses) for I/O issues, ensuring consistency with Python's built-in file handling.

import asyncio

import aiogzip

async def safe_read():
    try:
        async with aiogzip.open("non_existent.gz", "rb") as f:
            await f.read()
    except FileNotFoundError:
        print("File not found!")
    except OSError as e:
        print(f"An I/O error occurred: {e}")

asyncio.run(safe_read())

Reading Untrusted Files Safely

When reading gzip data from untrusted sources, cap decompressed output to avoid expanding a small compressed file into unbounded memory usage. The inflater is limited to the remaining allowance plus one byte, so the guard does not first allocate the complete expanded payload:

import asyncio

import aiogzip

async def read_untrusted(path):
    async with aiogzip.open(
        path, "rb", max_decompressed_size=100 * 1024 * 1024
    ) as f:
        return await f.read()

asyncio.run(read_untrusted("upload.gz"))

If the input is a non-seekable stream, aiogzip may keep compressed bytes so backward seeks can replay the stream. The cache defaults to 128 MiB; reduce max_rewind_cache_size for memory-sensitive pipelines, or use forward-only reads and leave backward seeking disabled once the cap is exceeded.