Skip to content

Recipes

These recipes use the public API and the behavior covered by aiogzip's test suite. Use open() for streaming; the whole-file read() and write() helpers are intended for payloads that comfortably fit in memory.

Reading JSON Lines

import json

import aiogzip

async with aiogzip.open("events.jsonl.gz", "rt") as f:
    async for line in f:
        event = json.loads(line)
        await process(event)

For large UTF-8 JSON Lines files that are known to use \n terminators, newline="\n" avoids universal-newline translation. Benchmarks in this project use chunk_size=512 * 1024 to reduce async read overhead; tune that value for your memory budget and workload.

For CPU-bound parsing, batched iteration avoids one async transition per input line while retaining a predictable application-level working set:

async with aiogzip.open(
    "events.jsonl.gz",
    "rt",
    newline="\n",
    chunk_size=512 * 1024,
) as f:
    async for batch in f.iter_batches():
        for line in batch:
            event = json.loads(line)
            process(event)

iter_batches(hint) yields non-empty lists of complete lines and is exactly equivalent to calling readlines(hint) in a loop. The hint (default 1 MiB) counts decoded characters and a batch is complete after the whole line that reaches it, so it is an approximate batch target rather than a strict memory limit. Prefer async for line in f when each record is handed to an asynchronous consumer and per-line backpressure is desirable.

Writing JSON Lines

Write records as they become available instead of building the complete file in memory:

import json

import aiogzip

async with aiogzip.open("events.jsonl.gz", "wt", newline="\n") as f:
    async for event in events:
        await f.write(json.dumps(event) + "\n")

If the records already exist as a synchronous iterable of strings, await f.writelines(records) batches small writes while keeping memory use bounded.

Each await f.write(data) deliberately completes that call's codec work and delivers every compressed byte it produced to the sink before returning. This keeps sink failures attached to the call that triggered them, but makes a loop of tiny awaited writes expensive. Prefer writelines() or build explicit bounded batches when per-record sink-failure boundaries are unnecessary:

pending: list[str] = []
pending_chars = 0

async for record in records:
    line = json.dumps(record) + "\n"
    pending.append(line)
    pending_chars += len(line)
    if pending_chars >= 64 * 1024:
        await f.writelines(pending)
        pending.clear()
        pending_chars = 0

await f.writelines(pending)

aiogzip does not buffer across separate write() calls by default. Do not catch a write error and continue the same writer: the incomplete gzip member is poisoned and must be discarded.

Processing untrusted gzip input

Set an application-appropriate decompressed-size ceiling:

import aiogzip

async with aiogzip.open(
    "uploaded-data.gz",
    "rb",
    max_decompressed_size=100 * 1024 * 1024,
) as f:
    while chunk := await f.read(64 * 1024):
        await process(chunk)

The right limit depends on the file format, account quotas, and available resources. Crossing it raises OSError; the inflater is bounded to the remaining allowance plus one byte for overflow detection instead of first allocating the complete expanded payload.

A small file read, such as read(1) or peek(1), can inflate and retain much more plaintext than it returns. The file wrapper drains the compressed input it has already accepted; a highly compressible file can fit in a single source read. Trailer validation and member metadata can therefore advance beyond the logical read position. The streaming codec's output_chunk_size limits each emitted chunk, not the total plaintext retained by a file handle.

For concurrent untrusted inputs, set max_decompressed_size and bound the number of active handles according to available memory. Treat requested read sizes as delivery sizes, not memory budgets. The async-iterable decompress_chunks() API offers consumer-driven output when file seeking and file-layer validation timing are not required.

When the payload is not needed, use verify() to perform the same complete validation scan while discarding decompressed bytes:

result = await aiogzip.verify(
    "uploaded-data.gz",
    max_decompressed_size=100 * 1024 * 1024,
)
print(result.uncompressed_size)

Reproducible gzip output

The gzip header timestamp and embedded filename both affect output bytes. Set mtime and original_filename explicitly when output must be identical even if the destination path changes:

import aiogzip

await aiogzip.write(
    "build/output.gz",
    b"stable payload",
    mtime=0,
    original_filename="payload.bin",
)

Use original_filename="" to omit the filename header field. Keep the compression engine and level fixed as well: default compression uses stdlib zlib at level 6, while fast_compress=True can produce different valid gzip bytes.

Append mode and concatenated gzip members

import aiogzip

async with aiogzip.open("events.gz", "wb") as f:
    await f.write(b"first\n")

async with aiogzip.open("events.gz", "ab") as f:
    await f.write(b"second\n")

assert await aiogzip.read("events.gz") == b"first\nsecond\n"

Append mode adds a new gzip member; it does not rewrite or extend the first deflate stream. aiogzip, gzip.open(), and other standards-compliant readers return the concatenated decompressed contents.

Seeking

import aiogzip

async with aiogzip.open("events.gz", "rb") as f:
    header = await f.read(100)
    await f.seek(0)
    complete_data = await f.read()

Backward seeking rewinds and replays decompression because gzip has no random access index. Prefer sequential processing when possible. If repeated replay is expensive, reopening can make lifecycle intent clearer but does not avoid the decompression cost required to reach a later uncompressed offset.

A binary reader's seek(offset, SEEK_END) decompresses the rest of the stream (validating every trailer) to find the end, then moves to end + offset, clamped to the range from zero to the end. Output already buffered by a peek() or partial read counts exactly once, so the result is the same whether or not the reader had already reached EOF.

A read-mode seek() whose target is negative, such as seek(-5) or a SEEK_CUR offset that goes before the start, raises OSError("Negative seek in read mode"). The stdlib gzip.GzipFile instead clamps it and returns 0. Clamp the target yourself if you depend on that behavior. In text mode a negative argument is a tell() cookie, not an offset (see seek() and tell() in text mode).

Cancellation

Large compression and decompression calls may run in an executor. A worker thread cannot be stopped after its awaiting task is cancelled, so cancellation waits until the worker has finished before it propagates; the codec state has then advanced.

If a read is cancelled while decompression runs in an executor, the reader is broken: later reads raise OSError. A successful seek(0) recovers it, either because the source can rewind or because a non-seekable source's replay cache still holds the compressed input from the start. If seek(0) fails, close the reader and open a new one. If an executor-backed write is cancelled, discard that incomplete output member and create a new writer rather than continuing on the broken stream. A cancelled source read has different consequences, described under Source failures and cancellation.

import asyncio

import aiogzip

async def read_payload(path):
    try:
        return await aiogzip.read(path)
    except asyncio.CancelledError:
        # aiogzip.read() closes its handle before propagating cancellation.
        raise

External async file objects

The fileobj argument accepts objects with asynchronous read() or write() methods. For example, an existing aiofiles handle can be wrapped without transferring ownership:

import aiofiles

import aiogzip

async with aiofiles.open("payload.gz", "rb") as raw:
    async with aiogzip.open(None, "rb", fileobj=raw, closefd=False) as f:
        payload = await f.read()

With closefd=False, closing the gzip wrapper leaves the external object open. Pass closefd=True only when the wrapper should close it. Non-seekable readers use a bounded compressed-input replay cache for backward seeks.

Source failures and cancellation

An unmodified binary aiofiles source is allowed to finish an outstanding read before cancellation propagates. The gzip reader retains the completed bytes for its next read, so cancellation cannot silently skip part of the compressed stream. Cancellation can therefore wait for blocked native I/O.

Other custom sources must finish all source access before read() or seek() returns or raises, including cancellation. They must not leave hidden background work accessing the source. After an error or cancellation, the gzip reader becomes broken unless an unchanged, trustworthy synchronous tell() proves that no input was consumed. That cursor must account for consumption even when the method raises; a success-only byte counter does not establish safe retry. Async tell() methods do not provide this checkpoint.

The custom-source checkpoint calls synchronous tell() once before each physical read (and seek), plus once after a failure when a valid initial checkpoint exists. These are new calls into source code: tell() must be observational and inexpensive. A successful read does not require a second checkpoint.

A broken source-read path does not expose buffered data as validation salvage. Recover with a successfully completed physical seek(0), or reopen the source from the beginning. A non-seekable replay cache cannot recover input that the source consumed without returning. Treat this as a compatibility correction in 2.0.0b2: an OSError alone no longer implies that retrying a custom source is safe. Do not access a borrowed source concurrently through another owner.

Gzip over S3 / fsspec

The same fileobj mechanism streams gzip straight from object storage: anything exposing an async read() (for reading) or async write() (for writing) satisfies the protocol. With fsspec's async filesystems (e.g. s3fs):

import json

import aiogzip
import s3fs

fs = s3fs.S3FileSystem(asynchronous=True)

async def read_events(bucket_key):
    session = await fs.set_session()
    raw = await fs.open_async(bucket_key, "rb")
    try:
        async with aiogzip.open(None, "rt", fileobj=raw, closefd=False) as f:
            async for batch in f.iter_batches():
                for line in batch:
                    process(json.loads(line))
    finally:
        await raw.close()
        await session.close()

Notes:

  • Pass filename=None when supplying fileobj.
  • Object-store readers are typically non-seekable; forward reads stream normally, while backward seek() relies on the bounded replay cache (max_rewind_cache_size). For pure forward iteration, no tuning is needed.
  • Writing works symmetrically: open the remote object for binary write and pass it as fileobj to aiogzip.open(None, "wt", fileobj=...).
  • For untrusted buckets, combine this with max_decompressed_size — see Processing untrusted gzip input.

Opening ownership

Await open() before sharing a gzip handle with other tasks. A concurrent open() or close() during acquisition or initialization raises ConcurrentOperationError. Cancellation of a path-backed open waits for native acquisition and cleanup; it cannot abandon a file that the executor opens later. A handle becomes available only after its codec and initial header or seekability probe are ready. Ordinary repeated close and already-open errors are unchanged.

An external file object remains caller-owned if opening fails, preserving failed-open retry behavior. After opening succeeds, closefd controls whether closing the gzip handle also closes that object. Custom async sources/sinks must finish their resource access before returning or raising, including cancellation.