API Reference
aiogzip exposes its supported public API from the top-level package:
AsyncGzipBinaryFile — binary-mode reader/writerAsyncGzipTextFile — text-mode reader/writeropen — recommended factory returning the right class for a mode string (accepts r/w/a/x ops with a b or t suffix)AsyncGzipFile — compatibility name for the same factory behavior; it remains fully supportedread — read and decompress a complete binary stream into memorywrite — compress and write a complete bytes-like payloadEngineInfo and engine_info — immutable diagnostic information about the default compression and active decompression enginesGzipMemberInfo, GzipInfo, and inspect — validated per-member metadata and aggregate sizes from a complete decompression scanVerificationResult and verify — lightweight aggregate counts after complete integrity verificationdecompress_chunks — pull-driven decompression from an AsyncIterable[bytes] with bounded output chunkscompress_chunks — one-member gzip compression from an AsyncIterable[bytes] with bounded output chunksCodecOperation — the closeable iterator returned by codec state changesConcurrentOperationError — an OSError subtype for operations that
overlap on one single-task-owned async file handleGzipEncoder and GzipDecoder — synchronous sans-I/O state machines for
transport-independent, bounded gzip encoding and decodingWithAsyncRead, WithAsyncWrite, WithAsyncReadWrite — runtime-checkable protocols describing the async file objects accepted via fileobj=ZlibEngine — type alias for zlib compressor/decompressor objects (currently Any; the concrete C types are not exposed in type stubs)GZIP_WBITS, GZIP_METHOD_DEFLATE, GZIP_OS_UNKNOWN, and the GZIP_FLAG_FNAME / GZIP_FLAG_FHCRC / GZIP_FLAG_FEXTRA / GZIP_FLAG_FCOMMENT header-flag constants — useful when inspecting gzip headers alongside this library__version__
The synchronous codec is implemented in the public aiogzip.codec module.
Implementation internals live in aiogzip._common, aiogzip._binary,
aiogzip._text, aiogzip._inspection, and aiogzip._streaming. Treat those
modules as private and unstable.
import aiogzip
async with aiogzip.open("events.jsonl.gz", "rt") as stream:
async for line in stream:
print(line)
For small files that fit comfortably in memory, the whole-file helpers avoid
manual lifecycle management:
data = await aiogzip.read("payload.bin.gz")
await aiogzip.write("copy.bin.gz", data, mtime=0)
Both helpers operate in binary mode and load the entire uncompressed payload
into memory. Use open() for streaming large files.
Exact boolean options
Public boolean configuration uses exact built-in booleans. fast_compress,
strict_size, and the codec's collect_member_info accept only True or
False. closefd accepts True, False, or None; None keeps the ownership
default, closing path-opened resources while preserving caller-supplied file
objects. Integer substitutes such as 0 and 1, strings, and custom truthy or
falsy objects raise TypeError with the parameter name.
The contract is identical across AsyncGzipBinaryFile,
AsyncGzipTextFile, open(), AsyncGzipFile, the whole-file helpers, the
async-iterable compressor, and the synchronous codec. Validation happens
before file acquisition, external file methods, source iteration, zlib-ng
warning logic, or codec operation reservation. This contract was tightened
during the alpha series and is frozen for 2.0; it is not a security fix.
For writers, strict_size=False preserves gzip's normal modulo-2³² ISIZE;
strict_size=True rejects a member crossing that field's range.
fast_compress=False keeps stdlib-compatible output, while
fast_compress=True opts that writer into zlib-ng when available and otherwise
emits the documented fallback warning. Explicit closefd=False preserves a
caller-supplied file object; closefd=True closes it with the wrapper.
Live member timestamps
In read mode, AsyncGzipBinaryFile.mtime is None until the shared gzip
parser completes the first valid member header. It then holds that header's
32-bit timestamp. Each later valid header in a concatenated stream replaces
the value, including repeated timestamps and zero.
The property follows parser progress rather than only the bytes returned to
the caller. If one compressed source chunk contains several members, decoder
read-ahead can leave mtime at the last header in that chunk even when the
current read() returned bytes from an earlier member. This means the exact
point at which the property advances can depend on chunk_size and source read
boundaries.
A complete header updates mtime before its DEFLATE body or trailer is
validated, so a later body, CRC, or ISIZE error does not roll the timestamp
back. An invalid or incomplete header does not commit; a rejected first header
therefore leaves mtime=None, while a rejected later header preserves the
last valid value. After a rewind, the old value remains visible until a header
is parsed again. AsyncGzipTextFile.mtime delegates to the same binary reader
state.
Synchronous codec
Use GzipEncoder and GzipDecoder when an application owns a synchronous or
custom transport and needs aiogzip's framing, concatenated-member traversal,
integrity validation, and output limits without any I/O or executor policy.
Each state-changing call returns a lazy CodecOperation that must be exhausted
before another operation begins. Its idempotent close() method provides
deterministic cleanup when a caller exits before exhaustion.
import aiogzip
encoder = aiogzip.GzipEncoder(mtime=0)
wire = b"".join(encoder.start())
wire += b"".join(encoder.feed(b"payload"))
wire += b"".join(encoder.finish())
decoder = aiogzip.GzipDecoder()
payload = b"".join(decoder.feed(wire)) + b"".join(decoder.finish())
Warning: Decoder output can precede its trailer. Integrity is established
only after the operation returned by finish() is exhausted.
The public codec API is frozen for the 2.0 line. See the
stability policy and synchronous codec guide for
the compatibility boundary, constructor validation, immutable input
snapshots, lifecycle hazards, thread safety, and error behavior.
With collect_member_info=True, GzipDecoder.members contains immutable
records only for members whose trailers validated. Those completed records
remain available if a later member fails or the codec is explicitly discarded;
the incomplete member is absent and the codec remains unusable. Retained
records prove only those members, not the entire concatenated stream. Use
decoder.finished as the whole-stream completion indicator. Collection is
opt-in because its memory use grows with the member count.
aiogzip.codec.CodecOperation
Bases: Iterator[bytes], Protocol
A lazy codec operation with deterministic abandonment cleanup.
Exhaust the operation to complete it. Calling :meth:close before
exhaustion releases captured work and makes the owning codec unusable;
closing an exhausted or already closed operation has no effect.
close
Release this operation, abandoning its codec if still incomplete.
Source code in src/aiogzip/codec.py
| def close(self) -> None:
"""Release this operation, abandoning its codec if still incomplete."""
...
|
aiogzip.codec.GzipEncoder
GzipEncoder(*, compresslevel: int = 6, mtime: int | float | None = None, original_filename: str | bytes | None = None, fast_compress: bool = False, strict_size: bool = False, output_chunk_size: int = 256 * 1024)
Bases: _CodecBase
Incrementally encode exactly one complete gzip member.
Methods return bounded lazy iterators. The iterator from each operation
must be exhausted before another method is called. Explicitly closing an
incomplete operation makes the encoder unusable; dropping it leaves the
encoder reserved until :meth:discard is called.
This class performs no I/O or executor offload and is not thread-safe.
Source code in src/aiogzip/codec.py
| def __init__(
self,
*,
compresslevel: int = 6,
mtime: int | float | None = None,
original_filename: str | bytes | None = None,
fast_compress: bool = False,
strict_size: bool = False,
output_chunk_size: int = 256 * 1024,
) -> None:
validated_fast_compress = _validate_bool(fast_compress, "fast_compress")
validated_strict_size = _validate_bool(strict_size, "strict_size")
super().__init__()
_validate_compresslevel(compresslevel)
_validate_chunk_size(output_chunk_size)
self._compresslevel = compresslevel
self._mtime = _normalize_mtime(mtime)
validated_filename = _validate_original_filename(original_filename)
self._filename = _derive_header_filename(validated_filename, None)
self._fast_compress = validated_fast_compress
self._strict_size = validated_strict_size
self._output_chunk_size = output_chunk_size
if self._fast_compress and not _engine.have_fast_engine():
warnings.warn(
"fast_compress=True requested but zlib-ng is not available; "
"falling back to stdlib zlib. Install the extra with "
"'pip install aiogzip[fast]' to enable faster compression.",
stacklevel=2,
)
self._engine: ZlibEngine = _engine.compressobj(
compresslevel,
-_engine.MAX_WBITS,
fast=self._fast_compress,
)
self._crc = 0
self._input_size = 0
self._started = False
self._finished = False
|
crc32
property
CRC-32 of input committed by advancing feed operations.
finished
property
Whether the finish operation completed successfully.
Uncompressed bytes committed by advancing feed operations.
started
property
Whether the header-producing operation has advanced.
feed
feed(data: bytes) -> CodecOperation
Consume one immutable input snapshot and emit compressed chunks.
Source code in src/aiogzip/codec.py
| def feed(self, data: bytes) -> CodecOperation:
"""Consume one immutable input snapshot and emit compressed chunks."""
self._check_encoder_available()
if not self._started:
raise ValueError("gzip encoder must be started before feeding data")
snapshot = _snapshot_bytes_input(data)
size = len(snapshot)
self._check_feed_size(size)
return self._reserve(self._feed(snapshot))
|
finish
finish() -> CodecOperation
Finalize DEFLATE and lazily emit the gzip trailer exactly once.
Source code in src/aiogzip/codec.py
| def finish(self) -> CodecOperation:
"""Finalize DEFLATE and lazily emit the gzip trailer exactly once."""
self._check_encoder_available()
if not self._started:
raise ValueError("gzip encoder must be started before finalizing")
return self._reserve(self._finish())
|
flush
flush() -> CodecOperation
Perform a non-finalizing Z_SYNC_FLUSH operation.
Source code in src/aiogzip/codec.py
| def flush(self) -> CodecOperation:
"""Perform a non-finalizing ``Z_SYNC_FLUSH`` operation."""
self._check_encoder_available()
if not self._started:
raise ValueError("gzip encoder must be started before flushing")
return self._reserve(self._flush())
|
start
start() -> CodecOperation
Start the member and lazily emit its gzip header exactly once.
Source code in src/aiogzip/codec.py
| def start(self) -> CodecOperation:
"""Start the member and lazily emit its gzip header exactly once."""
self._check_encoder_available()
if self._started:
raise ValueError("gzip encoder is already started")
return self._reserve(self._start())
|
aiogzip.codec.GzipDecoder
GzipDecoder(*, output_chunk_size: int = 256 * 1024, max_decompressed_size: int | None = None, collect_member_info: bool = False)
Bases: _CodecBase
Incrementally decode and validate zero or more gzip members.
Payload bytes may be emitted before their member trailer is available, so
integrity is established only after :meth:finish is exhausted. Completed
metadata is retained only when collect_member_info=True. Records for
trailer-validated members survive a later failure or explicit discard;
:attr:finished remains the whole-stream validation indicator.
Methods return bounded lazy iterators that must be exhausted. This class
performs no I/O or executor offload and is not thread-safe.
Source code in src/aiogzip/codec.py
| def __init__(
self,
*,
output_chunk_size: int = 256 * 1024,
max_decompressed_size: int | None = None,
collect_member_info: bool = False,
) -> None:
validated_collect_member_info = _validate_bool(
collect_member_info, "collect_member_info"
)
super().__init__()
_validate_chunk_size(output_chunk_size)
_validate_optional_positive_int(max_decompressed_size, "max_decompressed_size")
self._output_chunk_size = output_chunk_size
self._max_decompressed_size = max_decompressed_size
self._collect_member_info = validated_collect_member_info
self._pending = _InputQueue()
self._inflate_input: bytes | None = None
self._output = _OutputCursor()
self._eof_after_output = False
self._state = "header"
self._engine: ZlibEngine = None
self._header: _ParsedHeader | None = None
self._header_parser: _GzipHeaderParser | None = self._new_header_parser()
# Generation distinguishes repeated timestamps and mtime=0 from no
# completed header. Package-internal file readers observe these two
# scalars without retaining or reparsing header bytes.
self.__header_generation = 0
self.__last_header_mtime: int | None = None
self._members: list[GzipMemberInfo] = []
self._member_count = 0
self._member_offset = 0
self._member_crc = 0
self._member_size = 0
self._compressed_size = 0
self._consumed_size = 0
self._uncompressed_size = 0
self._allow_padding = False
self._finished = False
|
compressed_size
property
Compressed bytes snapshotted by calls to :meth:feed.
finished
property
Whether :meth:finish completed full-stream validation.
member_count
property
Number of completed and trailer-validated members.
members
property
members: tuple[GzipMemberInfo, ...]
Collected completed members, retained after failure or discard.
uncompressed_size
property
Decompressed bytes accounted during operation advancement.
feed
feed(data: bytes) -> CodecOperation
Accept compressed bytes and lazily emit bounded decoded output.
Source code in src/aiogzip/codec.py
| def feed(self, data: bytes) -> CodecOperation:
"""Accept compressed bytes and lazily emit bounded decoded output."""
self._check_decoder_available()
snapshot = _snapshot_bytes_input(data)
operation = self._reserve(self._feed(snapshot))
self._compressed_size += len(snapshot)
return operation
|
finish
finish() -> CodecOperation
Prove that all accepted input forms a complete valid gzip stream.
Source code in src/aiogzip/codec.py
| def finish(self) -> CodecOperation:
"""Prove that all accepted input forms a complete valid gzip stream."""
self._check_decoder_available()
return self._reserve(self._process(finalizing=True))
|
Engine diagnostics
engine_info() reports the default compression implementation and the active
decompression implementation without exposing internal codec objects:
import aiogzip
print(aiogzip.engine_info())
Compression is reported as stdlib-zlib because that is the default even when
zlib-ng is installed. A writer created with fast_compress=True opts that
individual stream into zlib-ng; that per-stream choice is not reflected by
engine_info(). The strings are human-readable diagnostics and are not a
stable machine-readable feature-detection API.
Inspection and verification
inspect() scans and validates the complete gzip stream but discards its
decompressed payload. It returns one immutable GzipMemberInfo per member,
including exact compressed offsets and sizes, actual uncompressed sizes,
literal trailer ISIZE values, and optional header metadata. Header mtime=0
is preserved as 0; FNAME and FCOMMENT use deterministic Latin-1 decoding.
import aiogzip
info = await aiogzip.inspect("events.gz", max_decompressed_size=1024**3)
for member in info.members:
print(member.index, member.compressed_offset, member.uncompressed_size)
verify() performs the same complete structural, deflate, CRC-32, and ISIZE
validation without retaining per-member metadata:
result = await aiogzip.verify("events.gz")
print(result.member_count, result.uncompressed_size)
Successful return means the entire stream is valid. Corruption raises
gzip.BadGzipFile; I/O and decompression-limit failures raise OSError.
Zero-byte input is valid and returns zero members and zero sizes. NUL padding
after a valid member is accepted and included in the aggregate compressed size;
other trailing data is treated as a malformed next member.
inspect() holds every member's record until it returns, so its memory grows
with the number of members; verify() keeps only running totals. While it
parses a header, inspect() (like GzipDecoder with
collect_member_info=True) buffers each FNAME, FCOMMENT and FEXTRA field
whole and keeps it in the member's record. FEXTRA is at most 64 KiB by the
format, but FNAME and FCOMMENT are bounded only by the header limit: any
single gzip header larger than 128 MiB raises gzip.BadGzipFile.
max_decompressed_size does not count header bytes. Scanning untrusted input
with many members or very large header fields can therefore use substantial
memory; prefer verify() when per-member metadata is not needed.
Cancelling inspect() or verify() waits for any native open, read or close
already running in a worker thread to finish before the cancellation
propagates. A native read still queued for a worker is skipped and never
touches the source. A file opened by path is closed once, and only after its last
read; if that close itself fails, the failure is reported as below. A
borrowed fileobj is closed only when closefd=True.
If the scan fails and closing the source then fails too, the scan's error is
raised with the close failure attached as a note. A cancellation or interrupt
that arrives during that close outranks an ordinary scan error and
propagates, with the scan error as its context. If the scan succeeded, a
close failure is raised as before.
Async-iterable decompression
decompress_chunks() accepts only an asynchronous iterable of bytes and
returns an AsyncIterator[bytes]. Empty source chunks are ignored, yielded
chunks are non-empty, and output_chunk_size is a strict upper bound.
async for data in aiogzip.decompress_chunks(
compressed_source(),
output_chunk_size=64 * 1024,
max_decompressed_size=1024**3,
):
await consume(data)
The stream is validated incrementally. Payload can be yielded before its final
CRC and trailer are available, so complete integrity validation occurs only
when iteration ends normally. Corruption raises gzip.BadGzipFile, output
limit violations raise OSError, invalid source items raise TypeError, and
source exceptions and cancellation propagate. See the
streaming guide for backpressure and lifecycle details.
compress_chunks() accepts an asynchronous iterable of uncompressed bytes
and yields exactly one gzip member. It emits the header promptly, before
requesting the first source item. Empty sources therefore produce a valid empty
member rather than zero bytes.
async for data in aiogzip.compress_chunks(
raw_source(),
mtime=0,
output_chunk_size=64 * 1024,
):
await send(data)
The trailer is emitted only after the source ends normally. If the source
raises, compression is cancelled, or output consumption stops early, bytes
already yielded form an incomplete member and must be discarded. Compression
levels, metadata, strict_size, and fast_compress match the file writer.
When writing through an external asynchronous fileobj, its write() method
may accept fewer bytes than requested as long as it returns the accepted byte
count. aiogzip retries short writes until the complete gzip block is written.
A zero-progress or otherwise invalid count raises OSError.
Safety limits
Set max_decompressed_size=<bytes> when reading untrusted gzip data. The
limit applies to cumulative decompressed output for the current pass through
the stream. Each inflate call is restricted to the remaining allowance plus
one byte, so an over-limit member raises OSError without first materializing
its full expansion in memory. Rewinding resets the accounting because the new
read pass starts again at byte zero.
Text encodings
AsyncGzipTextFile keeps one incremental encoder for the lifetime of a write
stream. Multiple write() or writelines() calls therefore produce one
continuous encoded byte stream, including for stateful encodings such as
UTF-16, UTF-32, and ISO-2022-JP. Any final encoder shift sequence is written
before the gzip member's final block and trailer.
Both binary and text writelines() collect small inputs into bounded
chunk_size batches before compression. Inputs larger than a batch are written
directly, so generators remain streaming and memory use stays bounded.
Byte-count and compression tuning parameters are integer-only:
chunk_size, compresslevel, max_decompressed_size, and
max_rewind_cache_size reject floats, strings, and booleans with TypeError.
seek() and tell() in text mode
AsyncGzipBinaryFile.tell() returns the current position as a plain non-negative count of decompressed bytes, and seek(offset) accepts any such offset.
AsyncGzipTextFile.tell() returns one of two things:
- A plain non-negative offset (decompressed bytes) when the stream is at a clean boundary — no buffered text, the decoder holds no partial multibyte sequence, and there is no pending
\r. This is the same value the underlying binary layer reports (await f.buffer.tell()).
- An opaque cookie (a negative integer) otherwise. The cookie encodes the decoder state needed to resume mid-character, mid-line, or mid-
\r\n, so round-tripping seek(await f.tell()) is exact.
seek() accepts both: a non-negative argument is treated as a plain offset (decompression is replayed forward to that byte — from the current position when the decoder is at a clean boundary at or behind the target, from the start of the stream otherwise), and a negative argument is decoded as a cookie.
Cookies are bound to the handle that minted them
Warning. A text tell() cookie is valid only on the same open handle. This differs from io.TextIOWrapper and gzip.open("rt"), whose tell() cookies encode only decoder state and remain usable after re-opening the same file. An aiogzip text cookie embeds a random per-instance nonce, which seek() validates; a cookie from a different handle (or from before the file was re-opened) is rejected with OSError rather than silently restoring the wrong decoder state. Do not persist cookies across processes or re-opens — persist a plain offset instead.
Resumable processing recipe
Because cookies are not portable, checkpoint progress as a plain offset taken at a line boundary, where the decoder is guaranteed clean (\n is single-byte, so it never splits a multibyte character). Drive the binary layer — which splits lines without the text layer's read-ahead — so await f.tell() is an exact decompressed byte offset, then resume in a new process by opening in "rt" and seeking to that offset.
A forward plain seek() is O(n) in the target offset: gzip has no random access, so aiogzip replays decompressed bytes up to the offset. Checkpoint at a coarse granularity if that cost matters. See Resumable text processing for the full worked example.
aiogzip
Async gzip file reader/writer public API.
AsyncGzipBinaryFile
AsyncGzipBinaryFile(filename: Union[str, bytes, Path, None], mode: str = 'rb', chunk_size: int = DEFAULT_CHUNK_SIZE, compresslevel: int = 6, mtime: Optional[Union[int, float]] = None, original_filename: Optional[Union[str, bytes]] = None, fileobj: Optional[Union[WithAsyncRead, WithAsyncWrite, WithAsyncReadWrite]] = None, closefd: Optional[bool] = None, max_decompressed_size: Optional[int] = None, max_rewind_cache_size: Optional[int] = _MAX_CHUNK_SIZE, strict_size: bool = False, fast_compress: bool = False)
An asynchronous gzip file reader/writer for binary data.
This class provides async gzip compression/decompression for binary data,
making it a drop-in replacement for gzip.open() in binary mode.
Features:
- Full compatibility with gzip.open() file format
- Binary mode only (no text encoding/decoding)
- Async context manager support
- Configurable chunk size for performance tuning
Basic Usage
Write binary data
async with AsyncGzipBinaryFile("data.gz", "wb") as f:
await f.write(b"Hello, World!")
Read binary data
async with AsyncGzipBinaryFile("data.gz", "rb") as f:
data = await f.read() # Returns bytes
Imperative lifecycle when async with is impractical
f = AsyncGzipBinaryFile("data.gz", "rb")
await f.open()
try:
data = await f.read()
finally:
await f.close()
Interoperability with gzip.open():
# Files created by AsyncGzipBinaryFile can be read by gzip.open()
async with AsyncGzipBinaryFile("data.gz", "wb") as f:
await f.write(b"data")
with gzip.open("data.gz", "rb") as f:
data = f.read() # Works perfectly!
Source code in src/aiogzip/_binary.py
| def __init__(
self,
filename: Union[str, bytes, Path, None],
mode: str = "rb",
chunk_size: int = DEFAULT_CHUNK_SIZE,
compresslevel: int = 6,
mtime: Optional[Union[int, float]] = None,
original_filename: Optional[Union[str, bytes]] = None,
fileobj: Optional[
Union[WithAsyncRead, WithAsyncWrite, WithAsyncReadWrite]
] = None,
closefd: Optional[bool] = None,
max_decompressed_size: Optional[int] = None,
max_rewind_cache_size: Optional[int] = _MAX_CHUNK_SIZE,
strict_size: bool = False,
fast_compress: bool = False,
) -> None:
validated_closefd = _validate_optional_bool(closefd, "closefd")
validated_strict_size = _validate_bool(strict_size, "strict_size")
validated_fast_compress = _validate_bool(fast_compress, "fast_compress")
# Validate inputs using shared validation functions
_validate_filename(filename, fileobj)
_validate_chunk_size(chunk_size)
_validate_optional_positive_int(max_decompressed_size, "max_decompressed_size")
_validate_optional_positive_int(max_rewind_cache_size, "max_rewind_cache_size")
# Validate mode and derive file characteristics
mode_op, saw_b, saw_t, plus = _parse_mode_tokens(mode)
if saw_t:
raise ValueError("Binary mode cannot include text ('t')")
# _parse_mode_tokens guarantees mode_op is one of r/w/a/x here.
self._filename = filename
self._mode = mode
self._mode_op = mode_op
self._mode_plus = plus
self._writing_mode = mode_op in {"w", "a", "x"}
if self._writing_mode:
_validate_compresslevel(compresslevel)
self._fast_compress = validated_fast_compress
if (
self._writing_mode
and self._fast_compress
and not _engine.have_fast_engine()
):
warnings.warn(
"fast_compress=True requested but zlib-ng is not available; "
"falling back to stdlib zlib. Install the extra with "
"'pip install aiogzip[fast]' to enable faster compression.",
stacklevel=2,
)
self._chunk_size = chunk_size
self._compresslevel = compresslevel
self._header_mtime = _normalize_mtime(mtime)
self._header_filename_override = _validate_original_filename(original_filename)
self._external_file = fileobj
self._closefd = (
validated_closefd if validated_closefd is not None else fileobj is None
)
# Determine the underlying file mode based on gzip mode
file_mode_suffix = "b"
self._file_mode = f"{mode_op}{file_mode_suffix}"
if plus:
self._file_mode += "+"
self._file: Any = None
self._encoder: Optional[GzipEncoder] = None
self._decoder: Optional[GzipDecoder] = None
self._buffer = bytearray() # Use bytearray for efficient buffer growth
self._buffer_offset: int = 0 # Offset to the start of valid data in _buffer
self._opening = False
self._is_closed: bool = False
self._eof: bool = False
self._owns_file: bool = False
self._position: int = 0
self._mtime: Optional[int] = None
self._decoder_header_generation: int = 0
self._compressed_cache = bytearray()
self._source_work: Optional[asyncio.Future[Any]] = None
self._source_native_call: Optional[_NativeSourceCall] = None
self._source_owner: Optional[asyncio.Task[Any]] = None
self._source_abort_requested = False
self._read_aborts = 0
self._pending_compressed_chunk: Optional[bytes] = None
self._replay_offset: Optional[int] = None
self._cache_rewindable_reads: bool = False
self._underlying_seekable: bool = True
self._max_rewind_cache_size: Optional[int] = max_rewind_cache_size
self._write_broken: bool = False
self._write_call_active: bool = False
self._read_call_active: bool = False
self._salvage_read_call = _BinaryReadReservation(
self,
recover_broken=False,
)
self._recovery_read_call = _BinaryReadReservation(
self,
recover_broken=True,
)
self._write_call = _BinaryWriteReservation(self)
self._active_call_waiter: Optional[asyncio.Future[None]] = None
self._close_lock = asyncio.Lock()
self._closed_observer: Optional[Callable[[], None]] = None
self._read_poison_observer: Optional[Callable[[bool], None]] = None
self._read_health: _ReadHealth
self._reset_read_health()
self._max_decompressed_size: Optional[int] = max_decompressed_size
self._strict_size = validated_strict_size
|
closed
property
Return True when this file has been closed.
mtime
property
Return the timestamp from the most recently parsed member header.
The value is None until the first complete valid header is read.
Decoder read-ahead can advance it beyond the bytes returned by the
current read call when concatenated members share one source chunk.
name
property
name: Union[str, bytes, Path, None]
Return the name of the file.
This property provides compatibility with the standard file API.
Returns the filename passed to the constructor, or falls back to the
underlying file object's name attribute when available.
Returns:
| Type |
Description |
Union[str, bytes, Path, None]
|
The filename as str, bytes, or Path, or None if no name is available.
|
__aenter__
async
__aenter__() -> AsyncGzipBinaryFile
Enter the async context manager and initialize resources.
Source code in src/aiogzip/_binary.py
| async def __aenter__(self) -> "AsyncGzipBinaryFile":
"""Enter the async context manager and initialize resources."""
return await self.open()
|
__aexit__
async
__aexit__(exc_type: Optional[type], exc_val: Optional[BaseException], exc_tb: Optional[Any]) -> None
Exit the context manager, flushing and closing the file.
Source code in src/aiogzip/_binary.py
| async def __aexit__(
self,
exc_type: Optional[type],
exc_val: Optional[BaseException],
exc_tb: Optional[Any],
) -> None:
"""Exit the context manager, flushing and closing the file."""
await self._close_on_context_exit(self.close, exc_val)
|
__aiter__
__aiter__() -> AsyncGzipBinaryFile
Make AsyncGzipBinaryFile iterable over newline-delimited chunks.
Source code in src/aiogzip/_binary.py
| def __aiter__(self) -> "AsyncGzipBinaryFile":
"""Make AsyncGzipBinaryFile iterable over newline-delimited chunks."""
return self
|
__anext__
async
Return the next line from the binary stream.
Source code in src/aiogzip/_binary.py
| async def __anext__(self) -> bytes:
"""Return the next line from the binary stream."""
if self._is_closed:
raise StopAsyncIteration
line = await self.readline()
if line == b"":
raise StopAsyncIteration
return line
|
close
async
Flushes any remaining compressed data and closes the file.
Source code in src/aiogzip/_binary.py
| async def close(self) -> None:
"""Flushes any remaining compressed data and closes the file."""
async with self._close_lock:
if self._opening:
raise ConcurrentOperationError(
"close() called while open() is in progress"
)
await self._close_locked()
|
detach
Detach is unsupported to mirror gzip.GzipFile behavior.
Source code in src/aiogzip/_binary.py
| def detach(self) -> Any:
"""Detach is unsupported to mirror gzip.GzipFile behavior."""
raise io.UnsupportedOperation("detach")
|
fileno
Return the underlying file descriptor number.
Source code in src/aiogzip/_binary.py
| def fileno(self) -> int:
"""Return the underlying file descriptor number."""
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
fileno_method = getattr(self._file, "fileno", None)
if fileno_method is None:
raise io.UnsupportedOperation("fileno() not supported by underlying file")
result = fileno_method()
if hasattr(result, "__await__"):
# Dispose of the never-awaited coroutine so it does not emit a
# "coroutine was never awaited" RuntimeWarning.
close_method = getattr(result, "close", None)
if callable(close_method):
close_method()
raise io.UnsupportedOperation(
"fileno() is not awaitable in underlying file"
)
return int(result)
|
flush
async
Flush any buffered compressed data to the file.
In write/append mode, this forces any buffered compressed data to be
written to the underlying file. Note that this does NOT write the gzip
trailer - use close() for that.
In read mode, this is a no-op for compatibility with the file API.
Examples:
async with AsyncGzipBinaryFile("file.gz", "wb") as f:
await f.write(b"Hello")
await f.flush() # Ensure data is written
await f.write(b" World")
Source code in src/aiogzip/_binary.py
| async def flush(self) -> None:
"""
Flush any buffered compressed data to the file.
In write/append mode, this forces any buffered compressed data to be
written to the underlying file. Note that this does NOT write the gzip
trailer - use close() for that.
In read mode, this is a no-op for compatibility with the file API.
Examples:
async with AsyncGzipBinaryFile("file.gz", "wb") as f:
await f.write(b"Hello")
await f.flush() # Ensure data is written
await f.write(b" World")
"""
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if not self._writing_mode:
if self._read_call_active:
self._check_read_call_usable()
return
if self._write_broken:
# Pretending the flush succeeded would tell the caller their
# bytes are safely on disk when the member is already torn.
self._raise_write_broken()
if self._file is not None:
with self._write_call as encoder:
await self._flush_writer(encoder)
|
isatty
Return True if the underlying stream is interactive.
Source code in src/aiogzip/_binary.py
| def isatty(self) -> bool:
"""Return True if the underlying stream is interactive."""
if self._file is None:
return False
isatty_method = getattr(self._file, "isatty", None)
if not callable(isatty_method):
return False
result = isatty_method()
if hasattr(result, "__await__"):
close_method = getattr(result, "close", None)
if callable(close_method):
close_method()
return False
return bool(result)
|
open
async
open() -> AsyncGzipBinaryFile
Open the file for I/O and return self.
This performs the same initialization as entering the async context
manager, for callers who prefer an explicit try/finally over async
with::
f = AsyncGzipBinaryFile("data.gz", "rb")
await f.open()
try:
data = await f.read()
finally:
await f.close()
Raises:
| Type |
Description |
ValueError
|
if the file is already open, or has already been closed
(a closed instance cannot be reopened, matching io objects).
|
Source code in src/aiogzip/_binary.py
| async def open(self) -> "AsyncGzipBinaryFile":
"""Open the file for I/O and return ``self``.
This performs the same initialization as entering the async context
manager, for callers who prefer an explicit try/finally over ``async
with``::
f = AsyncGzipBinaryFile("data.gz", "rb")
await f.open()
try:
data = await f.read()
finally:
await f.close()
Raises:
ValueError: if the file is already open, or has already been closed
(a closed instance cannot be reopened, matching io objects).
"""
if self._opening:
raise ConcurrentOperationError("open() is already in progress")
_check_can_open(self._is_closed, self._file is not None)
self._opening = True
resource = None
owns_file = False
encoder = None
decoder = None
try:
if self._external_file is not None:
resource = cast(Any, self._external_file)
owns_file = False
else:
# __init__'s _validate_filename guarantees a filename exists
# whenever no fileobj was given; assert keeps the narrowing.
assert self._filename is not None
resource = await _acquire_path(
self._filename, self._file_mode, aiofiles.open
)
owns_file = True
# Initialize compression/decompression engine based on mode
if self._writing_mode:
header_filename = self._header_filename_override
if header_filename is None and self._filename is not None:
header_filename = os.fspath(self._filename)
# __init__ already emits the established file-boundary warning
# when fast compression is unavailable. Avoid duplicating it
# when the codec performs its own direct-use warning.
with warnings.catch_warnings():
warnings.filterwarnings(
"ignore",
message="fast_compress=True requested.*",
)
encoder = GzipEncoder(
compresslevel=self._compresslevel,
mtime=self._header_mtime,
original_filename=header_filename,
fast_compress=self._fast_compress,
strict_size=self._strict_size,
output_chunk_size=max(
self._chunk_size, self.DEFAULT_CHUNK_SIZE
),
)
for header_chunk in encoder.start():
await _write_initial_header(resource, header_chunk)
else: # read mode
decoder = GzipDecoder(
# ``chunk_size`` governs transport reads and write
# batching. Preserve the codec's tuned bounded-output
# floor so tiny compatibility-test read sizes do not turn
# a large inflate result into thousands of Python
# operations.
output_chunk_size=max(self._chunk_size, self.DEFAULT_CHUNK_SIZE),
max_decompressed_size=self._max_decompressed_size,
)
self._eof = False
self._read_call_active = False
self._reset_read_health()
self._position = 0
self._mtime = None
self._decoder_header_generation = 0
self._compressed_cache.clear()
self._pending_compressed_chunk = None
self._replay_offset = None
self._underlying_seekable = await _initial_seekable(resource)
self._cache_rewindable_reads = not self._underlying_seekable
# Publication has no suspension: only fully initialized resources
# become reachable through the live handle.
self._file = resource
self._owns_file = owns_file
self._encoder = encoder
self._decoder = decoder
return self
except BaseException as failure:
try:
await self._cleanup_failed_enter(resource, owns_file, encoder, decoder)
except BaseException as cleanup:
# Outside cancellation/interrupts outrank an ordinary failure.
if isinstance(failure, Exception) and not isinstance(
cleanup, Exception
):
if cleanup.__context__ is None:
cleanup.__context__ = failure
raise
failure.add_note(f"Opening cleanup also failed: {cleanup!r}")
raise
finally:
self._opening = False
|
peek
async
peek(size: int = -1) -> bytes
Return up to size bytes without advancing the read position.
Source code in src/aiogzip/_binary.py
| async def peek(self, size: int = -1) -> bytes:
"""Return up to size bytes without advancing the read position."""
if self._mode_op != "r":
raise OSError("File not open for reading")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if self._read_call_active or self._read_health is not _HEALTHY:
self._check_read_call_usable(True)
if size is not None and size > _MAX_CHUNK_SIZE:
raise ValueError(
f"peek size must be <= {_MAX_CHUNK_SIZE} bytes "
f"({_MAX_CHUNK_SIZE // (1024 * 1024)} MiB)"
)
available = len(self._buffer) - self._buffer_offset
target = size
if target is None or target <= 0:
target = available if available > 0 else 1
if available < target and not self._eof:
with self._salvage_read_call:
while available < target and not self._eof:
await self._fill_buffer()
available = len(self._buffer) - self._buffer_offset
if available == 0 and self._eof:
break
self._check_read_call_not_aborted()
end = self._buffer_offset + min(target, available)
return bytes(self._buffer[self._buffer_offset : end])
|
raw
Expose the underlying file object for advanced integrations.
Source code in src/aiogzip/_binary.py
| def raw(self) -> Any:
"""Expose the underlying file object for advanced integrations."""
return self._file
|
read
async
read(size: int = -1) -> bytes
Reads and decompresses binary data from the file.
Parameters:
| Name |
Type |
Description |
Default |
size
|
int
|
Number of bytes to read (-1 for all remaining data)
|
-1
|
Returns:
Examples:
async with AsyncGzipBinaryFile("file.gz", "rb") as f:
data = await f.read() # Returns bytes
partial = await f.read(100) # Returns first 100 bytes
Source code in src/aiogzip/_binary.py
| async def read(self, size: int = -1) -> bytes:
"""
Reads and decompresses binary data from the file.
Args:
size: Number of bytes to read (-1 for all remaining data)
Returns:
bytes
Examples:
async with AsyncGzipBinaryFile("file.gz", "rb") as f:
data = await f.read() # Returns bytes
partial = await f.read(100) # Returns first 100 bytes
"""
if self._mode_op != "r":
raise OSError("File not open for reading")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if self._read_call_active or self._read_health is not _HEALTHY:
self._check_read_call_usable(True)
if size is None:
size = -1
if size < 0:
size = -1
# If size is -1, read all data in chunks to avoid memory issues
if size == -1:
# Return buffered data + read remaining (no recursion)
# Use list + b"".join() — it pre-computes total size and does
# one allocation, which beats bytearray.extend() reallocation.
chunks = []
total_read = 0
buf = self._buffer
offset = self._buffer_offset
if offset < len(buf):
chunk = bytes(memoryview(buf)[offset:])
chunks.append(chunk)
total_read += len(chunk)
# Release the bytearray's allocation before decoding the remainder.
# The immutable prefix above is the sole successful-read copy; it is
# restored only if the failure contract makes those bytes reachable.
del buf[:]
self._buffer_offset = 0
# Append decompressor output directly to the join list. Each piece
# is already a distinct bytes object, so this avoids copying every
# byte through self._buffer (extend + bytes()) only to copy again
# in the final join — the dominant cost for large read-all calls.
if not self._eof:
with self._salvage_read_call:
try:
while not self._eof:
for piece in await self._decompress_next():
chunks.append(piece)
total_read += len(piece)
self._check_read_call_not_aborted()
except BaseException:
# Validation failures expose decoded bytes for explicit
# salvage; transient source errors remain retryable. A
# terminal non-validation poison cannot expose them, so
# avoid copying an arbitrarily large failed read back.
if self._can_restore_failed_read():
# A failing feed can already have deposited its own
# pre-trailer pieces in ``buf``. Restore earlier
# successful output ahead of them, preserving stream
# order across the read-all transactional boundary.
failed_feed = bytes(buf)
del buf[:]
for piece in chunks:
buf.extend(piece)
buf.extend(failed_feed)
raise
self._position += total_read
return b"".join(chunks)
else:
buf = self._buffer
offset = self._buffer_offset
available = len(buf) - offset
# Fast path: buffer already has enough data
if available >= size:
end = offset + size
data_to_return = bytes(memoryview(buf)[offset:end])
self._buffer_offset = end
self._position += size
if end >= len(buf):
del buf[:]
self._buffer_offset = 0
return data_to_return
# Fill until we have enough. Keep the guard through publication so
# another task cannot consume the newly filled buffer first.
with self._salvage_read_call:
available = await self._fill_until(size)
# Return what we have
actual_read_size = min(size, available)
offset = self._buffer_offset
data_to_return = bytes(
memoryview(self._buffer)[offset : offset + actual_read_size]
)
self._buffer_offset += actual_read_size
self._position += actual_read_size
if self._buffer_offset >= len(self._buffer):
del self._buffer[:]
self._buffer_offset = 0
return data_to_return
|
read1
async
read1(size: int = -1) -> bytes
Read up to size bytes with at most one data-producing fill.
A single compressed chunk can decode to nothing (e.g. while consuming
the gzip header), so fills repeat until at least one byte is available;
like stdlib gzip's read1(), an empty result means EOF.
Source code in src/aiogzip/_binary.py
| async def read1(self, size: int = -1) -> bytes:
"""Read up to size bytes with at most one data-producing fill.
A single compressed chunk can decode to nothing (e.g. while consuming
the gzip header), so fills repeat until at least one byte is available;
like stdlib gzip's ``read1()``, an empty result means EOF.
"""
if self._mode_op != "r":
raise OSError("File not open for reading")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if self._read_call_active or self._read_health is not _HEALTHY:
self._check_read_call_usable(True)
if size is None:
size = -1
if size == 0:
return b""
available = len(self._buffer) - self._buffer_offset
if available <= 0 and not self._eof:
with self._salvage_read_call:
available = await self._fill_until(1)
if size is None or size < 0:
actual_read_size = available
else:
actual_read_size = min(size, available)
data_to_return = bytes(
memoryview(self._buffer)[
self._buffer_offset : self._buffer_offset + actual_read_size
]
)
self._buffer_offset += actual_read_size
self._position += actual_read_size
if self._buffer_offset >= len(self._buffer):
del self._buffer[:]
self._buffer_offset = 0
return data_to_return
|
readinto
async
readinto(b: Union[bytearray, memoryview]) -> int
Read bytes directly into a pre-allocated, writable buffer.
Fills the caller's buffer straight from the decompression buffer,
avoiding the intermediate bytes object that delegating to read()
would allocate. Returns the number of bytes written (0 at EOF).
Source code in src/aiogzip/_binary.py
| async def readinto(self, b: Union[bytearray, memoryview]) -> int:
"""Read bytes directly into a pre-allocated, writable buffer.
Fills the caller's buffer straight from the decompression buffer,
avoiding the intermediate ``bytes`` object that delegating to ``read()``
would allocate. Returns the number of bytes written (0 at EOF).
"""
if self._mode_op != "r":
raise OSError("File not open for reading")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if self._read_call_active or self._read_health is not _HEALTHY:
self._check_read_call_usable(True)
view = memoryview(b)
if view.readonly:
raise TypeError("readinto() argument must be writable")
# Accept any writable buffer (e.g. array.array with itemsize > 1) by
# viewing it as bytes, like the stdlib io machinery does.
view = view.cast("B")
total = len(view)
if total == 0:
return 0
# Fill the internal buffer until it can satisfy the whole request (or
# EOF), then copy once. Filling before consuming preserves read()'s
# error semantics: if a refill raises mid-request, the stream position
# and already-buffered data are left intact for the caller to salvage.
with self._salvage_read_call:
await self._fill_until(total)
return self._copy_into_view(view, 0)
|
readinto1
async
readinto1(b: Union[bytearray, memoryview]) -> int
Read directly into the buffer with at most one data-producing fill.
Like read1() but writes straight into the caller's buffer, avoiding
the intermediate bytes object. A single compressed chunk can decode
to nothing (e.g. while consuming the gzip header), so fills repeat
until at least one byte is available; the result is still capped at one
buffer's worth of decoded data. Returns the number of bytes written
(0 only at EOF), so while await f.readinto1(buf): ... is safe.
Source code in src/aiogzip/_binary.py
| async def readinto1(self, b: Union[bytearray, memoryview]) -> int:
"""Read directly into the buffer with at most one data-producing fill.
Like ``read1()`` but writes straight into the caller's buffer, avoiding
the intermediate ``bytes`` object. A single compressed chunk can decode
to nothing (e.g. while consuming the gzip header), so fills repeat
until at least one byte is available; the result is still capped at one
buffer's worth of decoded data. Returns the number of bytes written
(0 only at EOF), so ``while await f.readinto1(buf): ...`` is safe.
"""
if self._mode_op != "r":
raise OSError("File not open for reading")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if self._read_call_active or self._read_health is not _HEALTHY:
self._check_read_call_usable(True)
view = memoryview(b)
if view.readonly:
raise TypeError("readinto1() argument must be writable")
view = view.cast("B")
total = len(view)
if total == 0:
return 0
available = len(self._buffer) - self._buffer_offset
if available <= 0 and not self._eof:
with self._salvage_read_call:
await self._fill_until(1)
return self._copy_into_view(view, 0)
|
readline
async
readline(limit: int = -1) -> bytes
Read and return one line from the binary stream.
Source code in src/aiogzip/_binary.py
| async def readline(self, limit: int = -1) -> bytes:
"""Read and return one line from the binary stream."""
if self._mode_op != "r":
raise OSError("File not open for reading")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if self._read_call_active or self._read_health is not _HEALTHY:
self._check_read_call_usable(True)
if limit is None or limit < 0:
# Any negative limit means "no limit", matching io.IOBase. Values
# below -1 must not reach the arithmetic below, where they would
# move the buffer offset backwards.
limit = -1
if limit == 0:
return b""
# The common short-line path publishes directly from the existing
# buffer. Compaction belongs to the refill-only path below: even its
# per-call threshold check measurably costs line throughput.
buf = self._buffer
start = self._buffer_offset
buf_len = len(buf)
newline_index = buf.find(b"\n", start)
if newline_index != -1:
end = newline_index + 1
if limit != -1:
end = min(end, start + limit)
elif limit != -1 and buf_len - start >= limit:
end = start + limit
elif self._eof:
if self._read_health is not _HEALTHY:
self._check_read_usable()
end = buf_len
else:
with self._salvage_read_call:
return await self._readline_reserved(limit)
result = bytes(memoryview(buf)[start:end])
self._buffer_offset = end
self._position += end - start
if end >= len(buf):
del buf[:]
self._buffer_offset = 0
return result
|
readlines
async
readlines(hint: int = -1) -> List[bytes]
Read and return a list of lines from the binary stream.
Source code in src/aiogzip/_binary.py
| async def readlines(self, hint: int = -1) -> List[bytes]:
"""Read and return a list of lines from the binary stream."""
if self._mode_op != "r":
raise OSError("File not open for reading")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if self._read_call_active or self._read_health is not _HEALTHY:
self._check_read_call_usable(True)
lines: List[bytes] = []
total = 0
position_before = self._position
with self._salvage_read_call:
try:
while True:
line = await self._readline_reserved(-1)
if not line:
break
lines.append(line)
total += len(line)
if hint > 0 and total >= hint:
break
if (
self._read_health is not _HEALTHY
and self._validation_line_salvage_complete()
):
# Publish every complete recovery line exactly once.
# A trailing unterminated span stays buffered so the
# next line call reports the terminal validation error.
break
except BaseException:
# A composite call publishes nothing when it fails. Put complete
# lines consumed by earlier loop iterations back ahead of the
# current unread span for retry or validation-failure salvage.
# This handler is deliberately inside the reservation: a rejected
# overlapping call must never rewrite state owned by its winner.
if self._can_restore_failed_read():
unread = self._buffer[self._buffer_offset :]
restored = bytearray().join((*lines, unread))
self._buffer = restored
self._buffer_offset = 0
self._position = position_before
raise
return lines
|
seek
async
seek(offset: int, whence: int = os.SEEK_SET) -> int
Move to a new file position, mirroring gzip.GzipFile semantics.
Source code in src/aiogzip/_binary.py
| async def seek(self, offset: int, whence: int = os.SEEK_SET) -> int:
"""Move to a new file position, mirroring gzip.GzipFile semantics."""
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if self._writing_mode:
if self._write_broken:
self._raise_write_broken()
if whence == os.SEEK_CUR:
target = self._position + offset
elif whence == os.SEEK_SET:
target = offset
else:
raise ValueError("Seek from end not supported in write mode")
if target < self._position:
raise OSError("Negative seek in write mode")
count = target - self._position
if count > 0:
zero_chunk = b"\x00" * min(self._SEEK_ZERO_CHUNK_SIZE, count)
remaining = count
with self._write_call as encoder:
while remaining > 0:
chunk = (
zero_chunk
if remaining >= len(zero_chunk)
else zero_chunk[:remaining]
)
await self._write_reserved(chunk, encoder)
remaining -= len(chunk)
return self._position
position_before = self._position
rewound = False
with self._recovery_read_call:
try:
# A fresh decoder over the rewound source is the one supported
# recovery path for a poisoned reader. Other seeks cannot safely
# use state from the failed decoder, including relative seeks that
# happen to target 0.
if (
self._read_health is not _HEALTHY
and whence == os.SEEK_SET
and offset == 0
):
await self._rewind_reader()
rewound = True
self._check_read_call_not_aborted()
return 0
self._check_read_usable()
if whence == os.SEEK_SET:
target = offset
elif whence == os.SEEK_CUR:
target = self._position + offset
elif whence == os.SEEK_END:
# Count unread buffered output before checking EOF: a
# peek() can reach EOF while its output is still unread
# (BC12).
while True:
buffered = len(self._buffer) - self._buffer_offset
if buffered > 0:
self._position += buffered
del self._buffer[:]
self._buffer_offset = 0
if self._eof:
break
await self._fill_buffer()
target = self._position + offset
if target < 0:
target = 0
elif target > self._position:
target = self._position
else:
raise ValueError("Invalid whence value")
if target < 0:
raise OSError("Negative seek in read mode")
if target < self._position:
await self._rewind_reader()
rewound = True
await self._consume_bytes(target - self._position)
self._check_read_call_not_aborted()
return self._position
except BaseException:
if not rewound and (
self._is_closed or not self._can_restore_failed_read()
):
self._position = position_before
raise
|
tell
async
Return the current uncompressed file position.
Source code in src/aiogzip/_binary.py
| async def tell(self) -> int:
"""Return the current uncompressed file position."""
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
return self._position
|
truncate
truncate(size: Optional[int] = None) -> int
Truncation is unsupported for gzip-compressed streams.
Source code in src/aiogzip/_binary.py
| def truncate(self, size: Optional[int] = None) -> int:
"""Truncation is unsupported for gzip-compressed streams."""
raise io.UnsupportedOperation("truncate")
|
write
async
write(data: Union[bytes, bytearray, memoryview]) -> int
Compresses and writes binary data to the file.
Parameters:
| Name |
Type |
Description |
Default |
data
|
Union[bytes, bytearray, memoryview]
|
|
required
|
Examples:
async with AsyncGzipBinaryFile("file.gz", "wb") as f:
await f.write(b"Hello, World!") # Bytes input
Source code in src/aiogzip/_binary.py
| async def write(self, data: Union[bytes, bytearray, memoryview]) -> int:
"""
Compresses and writes binary data to the file.
Args:
data: Bytes to write
Examples:
async with AsyncGzipBinaryFile("file.gz", "wb") as f:
await f.write(b"Hello, World!") # Bytes input
"""
if not self._writing_mode:
raise OSError("File not open for writing")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if self._write_broken:
self._raise_write_broken()
# Exact bytes already satisfy the immutable-snapshot contract. Keep
# the overwhelmingly common path here so each small write does not
# pay a second Python method call merely to repeat the type check.
payload = data if type(data) is bytes else self._coerce_byteslike(data)
encoder = self._encoder
if encoder is None:
raise RuntimeError("gzip writer encoder is not initialized")
# Reserve inline so the primitive path avoids the measured context-
# manager cost; the codec/sink work itself is shared with composites.
# Parity: tests/test_parity_binary_write.py
self._check_write_call_available()
self._write_call_active = True
try:
return await self._write_reserved(payload, encoder)
finally:
self._finish_write_call(encoder)
|
writelines
async
writelines(lines: Iterable[bytes]) -> None
Write bytes-like lines in bounded batches without adding separators.
Source code in src/aiogzip/_binary.py
| async def writelines(self, lines: Iterable[bytes]) -> None:
"""Write bytes-like lines in bounded batches without adding separators."""
if not self._writing_mode:
raise OSError("File not open for writing")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if self._write_broken:
self._raise_write_broken()
pending = bytearray()
iterator = iter(lines)
with self._write_call as encoder:
while True:
try:
line = next(iterator)
except StopIteration:
break
except BaseException:
if pending:
await self._write_reserved(bytes(pending), encoder)
raise
try:
data = self._coerce_byteslike(line)
except BaseException:
if pending:
await self._write_reserved(bytes(pending), encoder)
raise
length = len(data)
if length >= self._chunk_size:
if pending:
await self._write_reserved(bytes(pending), encoder)
pending.clear()
await self._write_reserved(data, encoder)
else:
if pending and len(pending) + length > self._chunk_size:
await self._write_reserved(bytes(pending), encoder)
pending.clear()
pending.extend(data)
if pending:
await self._write_reserved(bytes(pending), encoder)
|
AsyncGzipTextFile
AsyncGzipTextFile(filename: Union[str, bytes, Path, None], mode: str = 'rt', chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE, encoding: Optional[str] = 'utf-8', errors: Optional[str] = 'strict', newline: Union[str, None] = None, compresslevel: int = 6, mtime: Optional[Union[int, float]] = None, original_filename: Optional[Union[str, bytes]] = None, fileobj: Optional[Union[WithAsyncRead, WithAsyncWrite, WithAsyncReadWrite]] = None, closefd: Optional[bool] = None, max_decompressed_size: Optional[int] = None, max_rewind_cache_size: Optional[int] = _MAX_CHUNK_SIZE, strict_size: bool = False, fast_compress: bool = False)
An asynchronous gzip file reader/writer for text data.
This class wraps AsyncGzipBinaryFile and provides text mode operations
with proper UTF-8 handling for multi-byte characters.
Features:
- Full compatibility with gzip.open() file format
- Text mode with automatic encoding/decoding
- Proper handling of multi-byte UTF-8 characters
- Line-by-line iteration support
- Async context manager support
Basic Usage
Write text data
async with AsyncGzipTextFile("data.gz", "wt") as f:
await f.write("Hello, World!") # String input
Read text data
async with AsyncGzipTextFile("data.gz", "rt") as f:
text = await f.read() # Returns string
Line-by-line iteration
async with AsyncGzipTextFile("data.gz", "rt") as f:
async for line in f:
print(line.strip())
Imperative lifecycle when async with is impractical
f = AsyncGzipTextFile("data.gz", "rt")
await f.open()
try:
text = await f.read()
finally:
await f.close()
Source code in src/aiogzip/_text.py
| def __init__(
self,
filename: Union[str, bytes, Path, None],
mode: str = "rt",
chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE,
encoding: Optional[str] = "utf-8",
errors: Optional[str] = "strict",
newline: Union[str, None] = None,
compresslevel: int = 6,
mtime: Optional[Union[int, float]] = None,
original_filename: Optional[Union[str, bytes]] = None,
fileobj: Optional[
Union[WithAsyncRead, WithAsyncWrite, WithAsyncReadWrite]
] = None,
closefd: Optional[bool] = None,
max_decompressed_size: Optional[int] = None,
max_rewind_cache_size: Optional[int] = _MAX_CHUNK_SIZE,
strict_size: bool = False,
fast_compress: bool = False,
) -> None:
validated_closefd = _validate_optional_bool(closefd, "closefd")
validated_strict_size = _validate_bool(strict_size, "strict_size")
validated_fast_compress = _validate_bool(fast_compress, "fast_compress")
# Validate inputs using shared validation functions
_validate_filename(filename, fileobj)
_validate_chunk_size(chunk_size)
_validate_optional_positive_int(max_decompressed_size, "max_decompressed_size")
_validate_optional_positive_int(max_rewind_cache_size, "max_rewind_cache_size")
# Validate text-specific parameters
if encoding is None:
encoding = "utf-8"
if not encoding:
raise ValueError("Encoding cannot be empty")
if errors is None:
errors = "strict"
if newline not in {None, "", "\n", "\r", "\r\n"}:
raise ValueError(f"illegal newline value: {newline}")
mode_op, saw_b, saw_t, plus = _parse_mode_tokens(mode)
if saw_b:
raise ValueError("Text mode cannot include binary ('b')")
# _parse_mode_tokens guarantees mode_op is one of r/w/a/x here.
self._filename = filename
self._mode = mode
self._mode_op = mode_op
self._mode_plus = plus
self._writing_mode = mode_op in {"w", "a", "x"}
if self._writing_mode:
_validate_compresslevel(compresslevel)
self._chunk_size = chunk_size
self._encoding = encoding
self._errors = errors
self._newline = newline
self._compresslevel = compresslevel
self._header_mtime = _normalize_mtime(mtime)
self._header_filename_override = _validate_original_filename(original_filename)
self._external_file = fileobj
self._closefd = (
validated_closefd if validated_closefd is not None else fileobj is None
)
# Determine the underlying binary file mode
self._binary_mode = f"{mode_op}b"
if plus:
self._binary_mode += "+"
self._binary_file: Optional[AsyncGzipBinaryFile] = None
self._opening = False
self._is_closed: bool = False
self._close_complete: bool = False
self._close_lock = asyncio.Lock()
self._read_call_active: bool = False
self._pending_read_origin: Optional[_TextBufferOrigin] = None
# One-way hint, not a health copy: set when binary reports poison and
# cleared only when text rewinds. It lets a healthy hot-path call skip
# the binary query; every decision it gates asks binary's authority.
self._read_poison_seen: bool = False
self._write_call_active: bool = False
self._read_call = _TextReadReservation(self)
self._write_call = _TextWriteReservation(self)
# Decoder and buffer state. Constructed eagerly so an invalid
# `encoding` raises LookupError at call site, matching stdlib
# io.TextIOWrapper and codecs.getincrementaldecoder semantics.
self._decoder = codecs.getincrementaldecoder(self._encoding)(
errors=self._errors
)
self._encoder = (
codecs.getincrementalencoder(self._encoding)(errors=self._errors)
if self._writing_mode
else None
)
self._encoder_used: bool = False
self._text_buffer: str = "" # Backing store for buffered decoded text
self._text_buffer_offset: int = 0 # Start of unread text within _text_buffer
self._trailing_cr: bool = False # Track if last decoded chunk ended with \r
self._seen_newline_types: int = 0
# Binary-layer offset of the last byte fed to the incremental decoder.
# Reads made directly on the public `buffer` accessor advance the
# binary position without going through the decoder, so this frontier
# is what plain-position bookkeeping must compare against.
self._decoder_byte_position: int = 0
self._cookie_nonce: int = secrets.randbits(64)
self._buffer_origin = _TextBufferOrigin(
0, self._decoder.getstate(), False, 0, 0
)
self._universal_newlines: bool = newline in {None, ""}
self._max_decompressed_size: Optional[int] = max_decompressed_size
self._max_rewind_cache_size: Optional[int] = max_rewind_cache_size
self._strict_size = validated_strict_size
self._fast_compress = validated_fast_compress
# Batched line iteration: for the single-character terminator modes we
# bulk-split a decoded chunk into whole lines once (C-speed) and serve
# them from _pending_lines, advancing _text_buffer_offset per line so
# tell()/seek() bookkeeping stays identical to consuming one at a time.
# `_line_term` is None for the modes that need cross-boundary handling
# ('' look-ahead, '\r\n'), which keep the per-line buffer-scan path.
if newline in self._FAST_READLINE_NEWLINES:
self._line_term: Optional[str] = "\r" if newline == "\r" else "\n"
# Keep this annotation unsubscripted because the compiled-pattern
# type is an implementation detail and gains nothing from a more
# specific runtime annotation here.
self._line_split_re: Optional[re.Pattern] = (
_LINE_RE_CR if newline == "\r" else _LINE_RE_LF
)
self._splitlines_unsafe: Optional[str] = (
_SPLITLINES_UNSAFE_CR if newline == "\r" else _SPLITLINES_UNSAFE_LF
)
else:
self._line_term = None
self._line_split_re = None
self._splitlines_unsafe = None
self._pending_lines: List[str] = []
self._pending_idx: int = 0
|
buffer
property
buffer: AsyncGzipBinaryFile
Expose the underlying binary gzip stream.
closed
property
Return True when this file has been closed.
encoding
property
Return the configured text encoding.
errors
property
Return the configured text error handler.
mtime
property
Return the latest parsed member timestamp from the binary layer.
name
property
name: Union[str, bytes, Path, None]
Return the name of the file.
This property provides compatibility with the standard file API.
Returns the filename passed to the constructor, or falls back to the
underlying file object's name attribute when available.
Returns:
| Type |
Description |
Union[str, bytes, Path, None]
|
The filename as str, bytes, or Path, or None if no name is available.
|
newlines
property
newlines: Optional[Union[str, Tuple[str, ...]]]
Return newline types observed while reading, like TextIOWrapper.
__aenter__
async
__aenter__() -> AsyncGzipTextFile
Enter the async context manager and initialize resources.
Source code in src/aiogzip/_text.py
| async def __aenter__(self) -> "AsyncGzipTextFile":
"""Enter the async context manager and initialize resources."""
return await self.open()
|
__aexit__
async
__aexit__(exc_type: Optional[type], exc_val: Optional[BaseException], exc_tb: Optional[Any]) -> None
Exit the context manager, flushing and closing the file.
Source code in src/aiogzip/_text.py
| async def __aexit__(
self,
exc_type: Optional[type],
exc_val: Optional[BaseException],
exc_tb: Optional[Any],
) -> None:
"""Exit the context manager, flushing and closing the file."""
binary_file = self._binary_file
if binary_file is None:
if self._opening and exc_val is not None:
# Match binary exceptional exit: preserve the body's failure.
# The concurrent opener still owns its unpublished resource;
# this exit neither releases it nor reports it closed.
return
await self.close()
return
try:
await binary_file._close_on_context_exit(self.close, exc_val)
finally:
self._sync_closed_from_binary()
|
__aiter__
__aiter__() -> AsyncGzipTextFile
Make AsyncGzipTextFile iterable for line-by-line reading.
Source code in src/aiogzip/_text.py
| def __aiter__(self) -> "AsyncGzipTextFile":
"""Make AsyncGzipTextFile iterable for line-by-line reading."""
return self
|
__anext__
async
Return the next line from the file.
Source code in src/aiogzip/_text.py
| async def __anext__(self) -> str:
"""Return the next line from the file."""
if self._is_closed:
raise StopAsyncIteration
if self._read_call_active or self._read_poison_seen:
self._check_text_read_call_usable()
if self._line_term is not None:
# Keep pending-line consumption inline: a helper call in this
# per-line hot path measurably lowers iteration throughput.
# Parity: tests/test_parity_text_inline.py (T4)
idx = self._pending_idx
pending = self._pending_lines
if idx < len(pending):
line = pending[idx]
self._pending_idx = idx + 1
self._text_buffer_offset += len(line)
return line
with self._read_call:
refilled = await self._next_fast_line()
if refilled is None:
raise StopAsyncIteration
return refilled
search_from = 0
while True:
pos, length = self._find_line_terminator(search_from)
if pos != -1:
return self._consume_buffer(pos + length)
# Track how far we've scanned before reading more data.
# Back up 1 char in case a \r at the boundary pairs with a new \n.
buf_len = len(self._text_buffer) - self._text_buffer_offset
with self._read_call:
return await self._anext_buffered_reserved(buf_len)
|
close
async
Closes the file.
Source code in src/aiogzip/_text.py
| async def close(self) -> None:
"""Closes the file."""
# Unlike the binary layer, text close has incremental-encoder state to
# finalize before delegating. Serialize the whole sequence so a second
# close cannot finalize the same encoder while the first awaits its
# trailer write.
async with self._close_lock:
if self._opening:
raise ConcurrentOperationError(
"close() called while open() is in progress"
)
# `_is_closed` mirrors effective binary closure so data methods fail
# immediately after `buffer.close()`. `_close_complete` separately
# records text-layer cleanup: binary closure must not skip encoder
# finalization (and its missing-shift error).
if self._close_complete:
return
binary_file = self._binary_file
if binary_file is None:
self._is_closed = True
self._close_complete = True
return
if self._read_call_active:
self._check_text_read_call_available()
if self._write_call_active:
self._check_text_write_call_available()
if self._writing_mode:
# Reserve text encoder state through final bytes and the binary
# trailer. A concurrent write is rejected before mutation.
binary_file._check_write_call_available()
with self._write_call:
await self._close_text_writer(binary_file)
return
try:
await binary_file.close()
finally:
self._sync_closed_from_binary()
if binary_file.closed:
self._close_complete = True
|
detach
Detach is unsupported to mirror gzip.GzipFile behavior.
Source code in src/aiogzip/_text.py
| def detach(self) -> Any:
"""Detach is unsupported to mirror gzip.GzipFile behavior."""
raise io.UnsupportedOperation("detach")
|
flush
async
Flush any buffered data to the file.
In write/append mode, this forces any buffered text to be encoded
and written to the underlying binary file.
In read mode, this is a no-op for compatibility with the file API.
Examples:
async with AsyncGzipTextFile("file.gz", "wt") as f:
await f.write("Hello")
await f.flush() # Ensure data is written
await f.write(" World")
Source code in src/aiogzip/_text.py
| async def flush(self) -> None:
"""
Flush any buffered data to the file.
In write/append mode, this forces any buffered text to be encoded
and written to the underlying binary file.
In read mode, this is a no-op for compatibility with the file API.
Examples:
async with AsyncGzipTextFile("file.gz", "wt") as f:
await f.write("Hello")
await f.flush() # Ensure data is written
await f.write(" World")
"""
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._binary_file is not None:
if self._writing_mode:
with self._write_call:
await self._binary_file.flush()
else:
if self._read_call_active:
self._check_text_read_call_available()
await self._binary_file.flush()
|
isatty
Return True if the underlying stream is interactive.
Source code in src/aiogzip/_text.py
| def isatty(self) -> bool:
"""Return True if the underlying stream is interactive."""
if self._binary_file is None:
return False
return self._binary_file.isatty()
|
iter_batches
iter_batches(hint: int = DEFAULT_BATCH_HINT) -> AsyncIterator[List[str]]
Iterate over the file in batches of complete lines.
Equivalent to calling readlines(hint) in a loop until it returns an
empty list, but as an async iterator::
async with aiogzip.open("file.gz", "rt") as f:
async for batch in f.iter_batches():
for line in batch:
process(line)
Each yielded batch is a non-empty list of lines (trailing newlines
included, exactly as readlines returns them); iteration ends at
EOF. Batching amortizes the per-line await overhead of async
for line in f and is measurably faster on line-dense files, while
keeping memory bounded by roughly hint characters per batch.
Parameters:
| Name |
Type |
Description |
Default |
hint
|
int
|
Positive batch size hint in characters. A batch is complete
once the lines collected so far total at least hint
characters (the line that crosses the hint is returned whole).
|
DEFAULT_BATCH_HINT
|
Returns:
| Type |
Description |
AsyncIterator[List[str]]
|
AsyncIterator[List[str]]: An async iterator of line batches.
|
Raises:
| Type |
Description |
TypeError
|
If hint is not an integer.
|
ValueError
|
|
Source code in src/aiogzip/_text.py
| def iter_batches(self, hint: int = DEFAULT_BATCH_HINT) -> AsyncIterator[List[str]]:
"""
Iterate over the file in batches of complete lines.
Equivalent to calling ``readlines(hint)`` in a loop until it returns an
empty list, but as an async iterator::
async with aiogzip.open("file.gz", "rt") as f:
async for batch in f.iter_batches():
for line in batch:
process(line)
Each yielded batch is a non-empty list of lines (trailing newlines
included, exactly as ``readlines`` returns them); iteration ends at
EOF. Batching amortizes the per-line ``await`` overhead of ``async
for line in f`` and is measurably faster on line-dense files, while
keeping memory bounded by roughly ``hint`` characters per batch.
Args:
hint: Positive batch size hint in characters. A batch is complete
once the lines collected so far total at least ``hint``
characters (the line that crosses the hint is returned whole).
Returns:
AsyncIterator[List[str]]: An async iterator of line batches.
Raises:
TypeError: If hint is not an integer.
ValueError: If hint is not positive.
"""
# Validate eagerly, at the call, not at the first __anext__ inside the
# generator — matching the strict-parameter conventions elsewhere.
if not isinstance(hint, int) or isinstance(hint, bool):
raise TypeError("iter_batches hint must be a positive integer")
if hint <= 0:
raise ValueError("iter_batches hint must be a positive integer")
return self._iter_batches(hint)
|
open
async
open() -> AsyncGzipTextFile
Open the file for I/O and return self.
This performs the same initialization as entering the async context
manager, for callers who prefer an explicit try/finally over async
with::
f = AsyncGzipTextFile("data.txt.gz", "rt")
await f.open()
try:
text = await f.read()
finally:
await f.close()
Raises:
| Type |
Description |
ValueError
|
if the file is already open, or has already been closed
(a closed instance cannot be reopened, matching io objects).
|
Source code in src/aiogzip/_text.py
| async def open(self) -> "AsyncGzipTextFile":
"""Open the file for I/O and return ``self``.
This performs the same initialization as entering the async context
manager, for callers who prefer an explicit try/finally over ``async
with``::
f = AsyncGzipTextFile("data.txt.gz", "rt")
await f.open()
try:
text = await f.read()
finally:
await f.close()
Raises:
ValueError: if the file is already open, or has already been closed
(a closed instance cannot be reopened, matching io objects).
"""
if self._opening:
raise ConcurrentOperationError("open() is already in progress")
_check_can_open(self._is_closed, self._binary_file is not None)
self._opening = True
try:
filename = os.fspath(self._filename) if self._filename is not None else None
binary_file = AsyncGzipBinaryFile(
filename=filename,
mode=self._binary_mode,
chunk_size=self._chunk_size,
compresslevel=self._compresslevel,
mtime=self._header_mtime,
original_filename=self._header_filename_override,
fileobj=self._external_file,
closefd=self._closefd,
max_decompressed_size=self._max_decompressed_size,
max_rewind_cache_size=self._max_rewind_cache_size,
strict_size=self._strict_size,
fast_compress=self._fast_compress,
)
try:
await binary_file.open()
except BaseException as failure:
try:
await binary_file.close()
except BaseException as cleanup:
# Outside cancellation/interrupts outrank an ordinary failure.
if isinstance(failure, Exception) and not isinstance(
cleanup, Exception
):
if cleanup.__context__ is None:
cleanup.__context__ = failure
raise
failure.add_note(f"Opening cleanup also failed: {cleanup!r}")
raise
binary_file._attach_text_observers(
closed=self._mark_binary_closed,
poisoned=self._mark_binary_read_poisoned,
)
self._binary_file = binary_file
return self
finally:
self._opening = False
|
read
async
read(size: int = -1) -> str
Reads and decodes text data from the file.
Parameters:
| Name |
Type |
Description |
Default |
size
|
int
|
Number of characters to read (-1 for all remaining data)
|
-1
|
Returns:
Examples:
async with AsyncGzipTextFile("file.gz", "rt") as f:
text = await f.read() # Returns string
partial = await f.read(100) # Returns first 100 chars as string
Source code in src/aiogzip/_text.py
| async def read(self, size: int = -1) -> str:
"""
Reads and decodes text data from the file.
Args:
size: Number of characters to read (-1 for all remaining data)
Returns:
str
Examples:
async with AsyncGzipTextFile("file.gz", "rt") as f:
text = await f.read() # Returns string
partial = await f.read(100) # Returns first 100 chars as string
"""
if self._mode_op != "r":
raise OSError("File not open for reading")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._binary_file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if self._read_call_active or self._read_poison_seen:
self._check_text_read_call_usable()
if size is None:
size = -1
if size < 0:
size = -1
# Handle read(0) - should return empty string without draining buffer
if size == 0:
return ""
if size == -1:
# Keep the decoded prefix shared until the one fallible binary read
# succeeds. Concurrent-operation and transient source failures can
# then be retried without losing text already decoded above it.
buf = self._text_buffer
off = self._text_buffer_offset
prefix = buf[off:] if off else buf
bf = self._binary_file
decoder = self._decoder
apply_nl = self._apply_newline_decoding
if prefix and bf._validation_salvage_exhausted():
self._set_buffer("")
return prefix
with self._read_call:
raw_all = await bf.read(-1)
self._decoder_byte_position = bf._position
chunks = [prefix] if prefix else []
if raw_all:
decoded = decoder.decode(raw_all, final=False)
if decoded:
chunks.append(apply_nl(decoded))
if bf._read_is_healthy():
final = decoder.decode(b"", final=True)
if final:
chunks.append(apply_nl(final))
self._finalize_pending_newline_state()
elif not chunks:
# A drained validation salvage that completes no character
# is not a clean EOF; report the broken stream instead.
bf._check_read_usable()
# Draining a validation salvage is not EOF: an incomplete
# trailing character stays undecoded rather than raising.
self._set_buffer("")
return "".join(chunks)
else:
# A fully buffered result cannot overlap another task because it
# has no suspension point. Any refill reserves the text state and
# publishes only after all requested characters are recoverable.
buffered = self._buffered_text_len()
if buffered >= size:
if self._pending_lines:
self._pending_lines = []
self._pending_idx = 0
return self._consume_buffer(size)
if self._pending_lines:
self._pending_lines = []
self._pending_idx = 0
with self._read_call:
return await self._read_sized_reserved(size)
|
readline
async
readline(limit: int = -1) -> str
Read and return one line from the file.
A line is defined as text ending with a newline character ('\n').
If the file ends without a newline, the last line is returned without one.
Parameters:
| Name |
Type |
Description |
Default |
limit
|
int
|
Maximum number of characters to return. Stops at newline,
EOF, or once the limit is reached (matching TextIOBase semantics).
|
-1
|
Returns:
| Name | Type |
Description |
str |
str
|
The next line from the file, including the newline if present.
Returns empty string at EOF.
|
Examples:
async with AsyncGzipTextFile("file.gz", "rt") as f:
line = await f.readline() # Read one line
while line:
print(line.rstrip())
line = await f.readline()
Source code in src/aiogzip/_text.py
| async def readline(self, limit: int = -1) -> str:
"""
Read and return one line from the file.
A line is defined as text ending with a newline character ('\\n').
If the file ends without a newline, the last line is returned without one.
Args:
limit: Maximum number of characters to return. Stops at newline,
EOF, or once the limit is reached (matching TextIOBase semantics).
Returns:
str: The next line from the file, including the newline if present.
Returns empty string at EOF.
Examples:
async with AsyncGzipTextFile("file.gz", "rt") as f:
line = await f.readline() # Read one line
while line:
print(line.rstrip())
line = await f.readline()
"""
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._mode_op != "r":
raise OSError("File not open for reading")
if self._read_call_active or self._read_poison_seen:
self._check_text_read_call_usable()
if limit is None or limit < 0:
# Any negative limit means "no limit", matching io.TextIOBase.
# Values below -1 must not reach _consume_buffer, where they
# would move the buffer offset backwards.
limit = -1
if limit == 0:
return ""
# Unbounded reads in a single-character terminator mode use the batched
# pending-line path. Bounded reads (limit != -1) cap the work at `limit`
# characters anyway, so they stay on the buffer path below.
if limit == -1 and self._line_term is not None:
idx = self._pending_idx
pending = self._pending_lines
if idx < len(pending):
line = pending[idx]
self._pending_idx = idx + 1
self._text_buffer_offset += len(line)
return line
with self._read_call:
refilled = await self._next_fast_line()
return refilled if refilled is not None else ""
# Bounded reads consume the buffer by character count, which would leave
# any batched pending lines pointing at stale offsets; drop them first.
if self._pending_lines:
self._pending_lines = []
self._pending_idx = 0
# Keep the bounded hot path inline: this avoids a helper frame and
# computes the buffered length only once before a refill.
# Parity: tests/test_parity_text_inline.py (T5)
pos, length = self._find_line_terminator(0)
if pos != -1:
end = pos + length
if limit != -1 and end > limit:
end = limit
return self._consume_buffer(end)
buf_len = len(self._text_buffer) - self._text_buffer_offset
if limit != -1 and buf_len >= limit:
return self._consume_buffer(limit)
with self._read_call:
return await self._readline_buffered_reserved(limit, buf_len)
|
readlines
async
readlines(hint: int = -1) -> List[str]
Read and return a list of lines from the file.
Parameters:
| Name |
Type |
Description |
Default |
hint
|
int
|
Optional size hint. If given and greater than 0, reading
stops once the decoded lines returned so far total at least
hint characters (the line that crosses the hint is still
returned whole). If hint is -1 or not given, all remaining
lines are read.
|
-1
|
Returns:
| Type |
Description |
List[str]
|
List[str]: A list of lines from the file, each including any trailing
|
List[str]
|
|
Examples:
async with AsyncGzipTextFile("file.gz", "rt") as f:
lines = await f.readlines() # Read all lines
for line in lines:
print(line.rstrip())
With size hint
async with AsyncGzipTextFile("file.gz", "rt") as f:
lines = await f.readlines(1024) # Read ~1KB of lines
Source code in src/aiogzip/_text.py
| async def readlines(self, hint: int = -1) -> List[str]:
"""
Read and return a list of lines from the file.
Args:
hint: Optional size hint. If given and greater than 0, reading
stops once the decoded lines returned so far total at least
hint characters (the line that crosses the hint is still
returned whole). If hint is -1 or not given, all remaining
lines are read.
Returns:
List[str]: A list of lines from the file, each including any trailing
newline character.
Examples:
async with AsyncGzipTextFile("file.gz", "rt") as f:
lines = await f.readlines() # Read all lines
for line in lines:
print(line.rstrip())
# With size hint
async with AsyncGzipTextFile("file.gz", "rt") as f:
lines = await f.readlines(1024) # Read ~1KB of lines
"""
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._mode_op != "r":
raise OSError("File not open for reading")
if self._read_call_active or self._read_poison_seen:
self._check_text_read_call_usable()
state = self._readlines_rollback_state()
lines: List[str] = []
try:
with self._read_call:
if self._line_term is not None:
return await self._readlines_fast(hint, lines)
total_size = 0
while True:
line = await self._readline_generic_reserved(-1)
if not line:
break
lines.append(line)
total_size += len(line)
if hint > 0 and total_size >= hint:
break
if (
self._read_poison_seen
and self._validation_line_salvage_complete()
):
break
return lines
except BaseException:
self._rollback_readlines(state, lines)
raise
|
truncate
truncate(size: Optional[int] = None) -> int
Truncation is unsupported for gzip-compressed streams.
Source code in src/aiogzip/_text.py
| def truncate(self, size: Optional[int] = None) -> int:
"""Truncation is unsupported for gzip-compressed streams."""
raise io.UnsupportedOperation("truncate")
|
write
async
Encodes and writes text data to the file.
Parameters:
| Name |
Type |
Description |
Default |
data
|
str
|
|
required
|
Examples:
async with AsyncGzipTextFile("file.gz", "wt") as f:
await f.write("Hello, World!") # String input
Source code in src/aiogzip/_text.py
| async def write(self, data: str) -> int:
"""
Encodes and writes text data to the file.
Args:
data: String to write
Examples:
async with AsyncGzipTextFile("file.gz", "wt") as f:
await f.write("Hello, World!") # String input
"""
if not self._writing_mode:
raise OSError("File not open for writing")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._binary_file is None:
raise ValueError("File not opened. Call await open() or use async with.")
if not isinstance(data, str):
raise TypeError("write() argument must be str, not bytes")
# Translate newlines according to Python's text I/O semantics
text_to_encode = data
if self._newline is None:
# Translate \n to os.linesep on write
text_to_encode = text_to_encode.replace("\n", os.linesep)
elif self._newline in ("\n", "\r", "\r\n"):
text_to_encode = text_to_encode.replace("\n", self._newline)
else:
# newline == '' means no translation; any other value treat as no translation
pass
# Keep the primitive encoder/sink body inline: both the reservation
# context manager and a direct helper call exceeded the 5% small-write
# threshold in pinned release measurements.
# Parity: tests/test_parity_text_inline.py (T1)
self._check_text_write_call_available()
self._write_call_active = True
try:
encoder = self._encoder
assert encoder is not None
encoder_state = encoder.getstate()
encoder_was_used = self._encoder_used
self._encoder_used = True
try:
encoded_data = encoder.encode(text_to_encode, final=False)
binary_file = self._binary_file
assert binary_file is not None
await binary_file.write(encoded_data)
except BaseException:
encoder.setstate(encoder_state)
self._encoder_used = encoder_was_used
raise
return len(data)
finally:
self._write_call_active = False
|
writelines
async
writelines(lines: Iterable[str]) -> None
Write a list of lines to the file.
Note that newlines are not added automatically; each string in the
iterable should include its own line terminator if desired.
Parameters:
| Name |
Type |
Description |
Default |
lines
|
Iterable[str]
|
An iterable of strings to write.
|
required
|
Examples:
async with AsyncGzipTextFile("file.gz", "wt") as f:
await f.writelines(["line1\n", "line2\n", "line3\n"])
From a generator
async with AsyncGzipTextFile("file.gz", "wt") as f:
await f.writelines(f"{i}\n" for i in range(100))
Source code in src/aiogzip/_text.py
| async def writelines(self, lines: Iterable[str]) -> None:
"""
Write a list of lines to the file.
Note that newlines are not added automatically; each string in the
iterable should include its own line terminator if desired.
Args:
lines: An iterable of strings to write.
Examples:
async with AsyncGzipTextFile("file.gz", "wt") as f:
await f.writelines(["line1\\n", "line2\\n", "line3\\n"])
# From a generator
async with AsyncGzipTextFile("file.gz", "wt") as f:
await f.writelines(f"{i}\\n" for i in range(100))
"""
if not self._writing_mode:
raise OSError("File not open for writing")
if self._is_closed:
raise ValueError("I/O operation on closed file.")
if self._binary_file is None:
raise ValueError("File not opened. Call await open() or use async with.")
# Like binary writelines(), refuse a torn member even when there is
# nothing to write.
self._binary_file._check_write_usable()
# ``pending`` holds no exact empty ``str``, so its size is bounded by
# the chunk size however many of them arrive (an empty ``str``
# subclass is still stored, as before; see below). ``pending_empty``
# records that an empty string joined the current batch, so the batch
# is still written (an empty write emits a UTF-16/32 BOM, for example)
# exactly when a stored empty string would have produced it.
pending: List[str] = []
pending_chars = 0
pending_empty = False
iterator = iter(lines)
with self._write_call:
while True:
try:
line = next(iterator)
except StopIteration:
break
except BaseException:
if pending or pending_empty:
await self._write_batch_reserved("".join(pending))
raise
if not isinstance(line, str):
if pending or pending_empty:
await self._write_batch_reserved("".join(pending))
await self._write_batch_reserved(line)
continue
length = len(line)
if length >= self._chunk_size:
if pending or pending_empty:
await self._write_batch_reserved("".join(pending))
pending = []
pending_chars = 0
pending_empty = False
await self._write_batch_reserved(line)
elif not length and type(line) is str:
# Only an exact str is known to be empty from its length;
# a subclass may override __len__, so it is stored as before.
pending_empty = True
else:
if pending and pending_chars + length > self._chunk_size:
await self._write_batch_reserved("".join(pending))
pending = []
pending_chars = 0
pending_empty = False
pending.append(line)
pending_chars += length
if pending or pending_empty:
await self._write_batch_reserved("".join(pending))
|
CodecOperation
Bases: Iterator[bytes], Protocol
A lazy codec operation with deterministic abandonment cleanup.
Exhaust the operation to complete it. Calling :meth:close before
exhaustion releases captured work and makes the owning codec unusable;
closing an exhausted or already closed operation has no effect.
close
Release this operation, abandoning its codec if still incomplete.
Source code in src/aiogzip/codec.py
| def close(self) -> None:
"""Release this operation, abandoning its codec if still incomplete."""
...
|
ConcurrentOperationError
Bases: OSError
Raised when overlapping operations use one file handle concurrently.
Await the operation that already owns the handle before retrying the
rejected call. One file handle is intentionally single-task-owned.
EngineInfo
dataclass
EngineInfo(compression: str, decompression: str, crc32: str = 'stdlib-zlib')
Human-readable codec selections used by aiogzip by default.
GzipDecoder
GzipDecoder(*, output_chunk_size: int = 256 * 1024, max_decompressed_size: int | None = None, collect_member_info: bool = False)
Bases: _CodecBase
Incrementally decode and validate zero or more gzip members.
Payload bytes may be emitted before their member trailer is available, so
integrity is established only after :meth:finish is exhausted. Completed
metadata is retained only when collect_member_info=True. Records for
trailer-validated members survive a later failure or explicit discard;
:attr:finished remains the whole-stream validation indicator.
Methods return bounded lazy iterators that must be exhausted. This class
performs no I/O or executor offload and is not thread-safe.
Source code in src/aiogzip/codec.py
| def __init__(
self,
*,
output_chunk_size: int = 256 * 1024,
max_decompressed_size: int | None = None,
collect_member_info: bool = False,
) -> None:
validated_collect_member_info = _validate_bool(
collect_member_info, "collect_member_info"
)
super().__init__()
_validate_chunk_size(output_chunk_size)
_validate_optional_positive_int(max_decompressed_size, "max_decompressed_size")
self._output_chunk_size = output_chunk_size
self._max_decompressed_size = max_decompressed_size
self._collect_member_info = validated_collect_member_info
self._pending = _InputQueue()
self._inflate_input: bytes | None = None
self._output = _OutputCursor()
self._eof_after_output = False
self._state = "header"
self._engine: ZlibEngine = None
self._header: _ParsedHeader | None = None
self._header_parser: _GzipHeaderParser | None = self._new_header_parser()
# Generation distinguishes repeated timestamps and mtime=0 from no
# completed header. Package-internal file readers observe these two
# scalars without retaining or reparsing header bytes.
self.__header_generation = 0
self.__last_header_mtime: int | None = None
self._members: list[GzipMemberInfo] = []
self._member_count = 0
self._member_offset = 0
self._member_crc = 0
self._member_size = 0
self._compressed_size = 0
self._consumed_size = 0
self._uncompressed_size = 0
self._allow_padding = False
self._finished = False
|
compressed_size
property
Compressed bytes snapshotted by calls to :meth:feed.
finished
property
Whether :meth:finish completed full-stream validation.
member_count
property
Number of completed and trailer-validated members.
members
property
members: tuple[GzipMemberInfo, ...]
Collected completed members, retained after failure or discard.
uncompressed_size
property
Decompressed bytes accounted during operation advancement.
feed
feed(data: bytes) -> CodecOperation
Accept compressed bytes and lazily emit bounded decoded output.
Source code in src/aiogzip/codec.py
| def feed(self, data: bytes) -> CodecOperation:
"""Accept compressed bytes and lazily emit bounded decoded output."""
self._check_decoder_available()
snapshot = _snapshot_bytes_input(data)
operation = self._reserve(self._feed(snapshot))
self._compressed_size += len(snapshot)
return operation
|
finish
finish() -> CodecOperation
Prove that all accepted input forms a complete valid gzip stream.
Source code in src/aiogzip/codec.py
| def finish(self) -> CodecOperation:
"""Prove that all accepted input forms a complete valid gzip stream."""
self._check_decoder_available()
return self._reserve(self._process(finalizing=True))
|
GzipEncoder
GzipEncoder(*, compresslevel: int = 6, mtime: int | float | None = None, original_filename: str | bytes | None = None, fast_compress: bool = False, strict_size: bool = False, output_chunk_size: int = 256 * 1024)
Bases: _CodecBase
Incrementally encode exactly one complete gzip member.
Methods return bounded lazy iterators. The iterator from each operation
must be exhausted before another method is called. Explicitly closing an
incomplete operation makes the encoder unusable; dropping it leaves the
encoder reserved until :meth:discard is called.
This class performs no I/O or executor offload and is not thread-safe.
Source code in src/aiogzip/codec.py
| def __init__(
self,
*,
compresslevel: int = 6,
mtime: int | float | None = None,
original_filename: str | bytes | None = None,
fast_compress: bool = False,
strict_size: bool = False,
output_chunk_size: int = 256 * 1024,
) -> None:
validated_fast_compress = _validate_bool(fast_compress, "fast_compress")
validated_strict_size = _validate_bool(strict_size, "strict_size")
super().__init__()
_validate_compresslevel(compresslevel)
_validate_chunk_size(output_chunk_size)
self._compresslevel = compresslevel
self._mtime = _normalize_mtime(mtime)
validated_filename = _validate_original_filename(original_filename)
self._filename = _derive_header_filename(validated_filename, None)
self._fast_compress = validated_fast_compress
self._strict_size = validated_strict_size
self._output_chunk_size = output_chunk_size
if self._fast_compress and not _engine.have_fast_engine():
warnings.warn(
"fast_compress=True requested but zlib-ng is not available; "
"falling back to stdlib zlib. Install the extra with "
"'pip install aiogzip[fast]' to enable faster compression.",
stacklevel=2,
)
self._engine: ZlibEngine = _engine.compressobj(
compresslevel,
-_engine.MAX_WBITS,
fast=self._fast_compress,
)
self._crc = 0
self._input_size = 0
self._started = False
self._finished = False
|
crc32
property
CRC-32 of input committed by advancing feed operations.
finished
property
Whether the finish operation completed successfully.
Uncompressed bytes committed by advancing feed operations.
started
property
Whether the header-producing operation has advanced.
feed
feed(data: bytes) -> CodecOperation
Consume one immutable input snapshot and emit compressed chunks.
Source code in src/aiogzip/codec.py
| def feed(self, data: bytes) -> CodecOperation:
"""Consume one immutable input snapshot and emit compressed chunks."""
self._check_encoder_available()
if not self._started:
raise ValueError("gzip encoder must be started before feeding data")
snapshot = _snapshot_bytes_input(data)
size = len(snapshot)
self._check_feed_size(size)
return self._reserve(self._feed(snapshot))
|
finish
finish() -> CodecOperation
Finalize DEFLATE and lazily emit the gzip trailer exactly once.
Source code in src/aiogzip/codec.py
| def finish(self) -> CodecOperation:
"""Finalize DEFLATE and lazily emit the gzip trailer exactly once."""
self._check_encoder_available()
if not self._started:
raise ValueError("gzip encoder must be started before finalizing")
return self._reserve(self._finish())
|
flush
flush() -> CodecOperation
Perform a non-finalizing Z_SYNC_FLUSH operation.
Source code in src/aiogzip/codec.py
| def flush(self) -> CodecOperation:
"""Perform a non-finalizing ``Z_SYNC_FLUSH`` operation."""
self._check_encoder_available()
if not self._started:
raise ValueError("gzip encoder must be started before flushing")
return self._reserve(self._flush())
|
start
start() -> CodecOperation
Start the member and lazily emit its gzip header exactly once.
Source code in src/aiogzip/codec.py
| def start(self) -> CodecOperation:
"""Start the member and lazily emit its gzip header exactly once."""
self._check_encoder_available()
if self._started:
raise ValueError("gzip encoder is already started")
return self._reserve(self._start())
|
GzipInfo
dataclass
GzipInfo(members: tuple[GzipMemberInfo, ...], compressed_size: int, uncompressed_size: int)
Aggregate information for a completely validated gzip stream.
member_count
property
Return the number of gzip members in stream order.
GzipMemberInfo
dataclass
GzipMemberInfo(index: int, compressed_offset: int, compressed_size: int, uncompressed_size: int, mtime: int, original_filename: str | None, comment: str | None, extra: bytes | None, flags: int, crc32: int, trailer_isize: int)
Validated metadata and sizes for one gzip member.
mtime preserves the literal unsigned header value, including zero.
Filename and comment fields are decoded one-to-one with Latin-1; absent
fields are None, while present empty fields are empty strings.
VerificationResult
dataclass
VerificationResult(member_count: int, compressed_size: int, uncompressed_size: int)
Aggregate counts returned after successful integrity verification.
WithAsyncRead
Bases: Protocol
Protocol for async file-like objects that can be read.
WithAsyncReadWrite
Bases: Protocol
Protocol for async file-like objects that can be read and written.
WithAsyncWrite
Bases: Protocol
Protocol for async file-like objects that can be written.
AsyncGzipFile
AsyncGzipFile(filename: _Filename, mode: _TextMode, *, chunk_size: int = ..., encoding: Optional[str] = ..., errors: Optional[str] = ..., newline: Optional[str] = ..., compresslevel: int = ..., mtime: Optional[Union[int, float]] = ..., original_filename: Optional[Union[str, bytes]] = ..., fileobj: _FileObj = ..., closefd: Optional[bool] = ..., max_decompressed_size: Optional[int] = ..., max_rewind_cache_size: Optional[int] = ..., strict_size: bool = ..., fast_compress: bool = ...) -> AsyncGzipTextFile
AsyncGzipFile(filename: _Filename, mode: _BinaryMode = ..., *, chunk_size: int = ..., compresslevel: int = ..., mtime: Optional[Union[int, float]] = ..., original_filename: Optional[Union[str, bytes]] = ..., fileobj: _FileObj = ..., closefd: Optional[bool] = ..., max_decompressed_size: Optional[int] = ..., max_rewind_cache_size: Optional[int] = ..., strict_size: bool = ..., fast_compress: bool = ...) -> AsyncGzipBinaryFile
AsyncGzipFile(filename: _Filename, mode: str, **kwargs: Any) -> Union[AsyncGzipBinaryFile, AsyncGzipTextFile]
AsyncGzipFile(filename: _Filename, mode: str = 'rb', **kwargs: Any) -> Union[AsyncGzipBinaryFile, AsyncGzipTextFile]
Factory function that returns the appropriate AsyncGzip class based on mode.
This provides backward compatibility with the original AsyncGzipFile interface
while using the new separated binary and text file classes.
Parameters:
| Name |
Type |
Description |
Default |
filename
|
_Filename
|
|
required
|
mode
|
str
|
File mode; any of 'r', 'w', 'a', 'x' with an optional 'b'
(binary, the default) or 't' (text) suffix
|
'rb'
|
**kwargs
|
Any
|
Additional arguments passed to the appropriate class
|
{}
|
Returns:
Source code in src/aiogzip/__init__.py
| def AsyncGzipFile(
filename: _Filename, mode: str = "rb", **kwargs: Any
) -> Union[AsyncGzipBinaryFile, AsyncGzipTextFile]:
"""
Factory function that returns the appropriate AsyncGzip class based on mode.
This provides backward compatibility with the original AsyncGzipFile interface
while using the new separated binary and text file classes.
Args:
filename: Path to the file
mode: File mode; any of 'r', 'w', 'a', 'x' with an optional 'b'
(binary, the default) or 't' (text) suffix
**kwargs: Additional arguments passed to the appropriate class
Returns:
AsyncGzipBinaryFile for binary modes ('rb', 'wb', 'ab', 'xb')
AsyncGzipTextFile for text modes ('rt', 'wt', 'at', 'xt')
"""
if not isinstance(mode, str):
raise TypeError("mode must be a string")
text_mode = "t" in mode
if not text_mode:
for arg_name in ("encoding", "errors", "newline"):
if kwargs.get(arg_name) is not None:
raise ValueError(f"Argument '{arg_name}' not supported in binary mode")
kwargs = {
key: value
for key, value in kwargs.items()
if key not in {"encoding", "errors", "newline"}
}
if text_mode:
return AsyncGzipTextFile(filename, mode, **kwargs)
else:
return AsyncGzipBinaryFile(filename, mode, **kwargs)
|
compress_chunks
compress_chunks(source: AsyncIterable[bytes], *, compresslevel: int = 6, mtime: Optional[Union[int, float]] = None, original_filename: Optional[Union[str, bytes]] = None, fast_compress: bool = False, strict_size: bool = False, output_chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE) -> AsyncIterator[bytes]
Incrementally compress an asynchronous byte iterable as one gzip member.
The header is emitted before the first source item is requested. Output
chunks are non-empty and no larger than output_chunk_size.
Parameters:
| Name |
Type |
Description |
Default |
source
|
AsyncIterable[bytes]
|
Asynchronous iterable yielding uncompressed bytes.
|
required
|
compresslevel
|
int
|
Compression level from -1 through 9.
|
6
|
mtime
|
Optional[Union[int, float]]
|
Optional gzip header timestamp. Use zero for reproducibility.
|
None
|
original_filename
|
Optional[Union[str, bytes]]
|
Optional filename stored in the gzip header.
|
None
|
fast_compress
|
bool
|
Opt into zlib-ng compression when available.
|
False
|
strict_size
|
bool
|
Reject payloads exceeding gzip's 32-bit ISIZE field.
|
False
|
output_chunk_size
|
int
|
Strict upper bound for each yielded chunk.
|
DEFAULT_CHUNK_SIZE
|
Returns:
| Type |
Description |
AsyncIterator[bytes]
|
A single-consumer asynchronous iterator containing one gzip member.
|
Raises:
| Type |
Description |
TypeError
|
If call-time arguments or a source item have invalid types.
|
ValueError
|
If an option is outside its supported range.
|
OSError
|
If compression fails or strict_size rejects the payload.
|
Source code in src/aiogzip/__init__.py
| def compress_chunks(
source: AsyncIterable[bytes],
*,
compresslevel: int = 6,
mtime: Optional[Union[int, float]] = None,
original_filename: Optional[Union[str, bytes]] = None,
fast_compress: bool = False,
strict_size: bool = False,
output_chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE,
) -> AsyncIterator[bytes]:
"""Incrementally compress an asynchronous byte iterable as one gzip member.
The header is emitted before the first source item is requested. Output
chunks are non-empty and no larger than ``output_chunk_size``.
Args:
source: Asynchronous iterable yielding uncompressed ``bytes``.
compresslevel: Compression level from ``-1`` through ``9``.
mtime: Optional gzip header timestamp. Use zero for reproducibility.
original_filename: Optional filename stored in the gzip header.
fast_compress: Opt into zlib-ng compression when available.
strict_size: Reject payloads exceeding gzip's 32-bit ``ISIZE`` field.
output_chunk_size: Strict upper bound for each yielded chunk.
Returns:
A single-consumer asynchronous iterator containing one gzip member.
Raises:
TypeError: If call-time arguments or a source item have invalid types.
ValueError: If an option is outside its supported range.
OSError: If compression fails or ``strict_size`` rejects the payload.
"""
return _compress_chunks(
source,
compresslevel=compresslevel,
mtime=mtime,
original_filename=original_filename,
fast_compress=fast_compress,
strict_size=strict_size,
output_chunk_size=output_chunk_size,
)
|
decompress_chunks
decompress_chunks(source: AsyncIterable[bytes], *, output_chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE, max_decompressed_size: Optional[int] = None) -> AsyncIterator[bytes]
Incrementally decompress gzip bytes from an asynchronous iterable.
Output chunks are non-empty and no larger than output_chunk_size.
Complete CRC and trailer validation occurs only when the returned iterator
is consumed to completion.
Parameters:
| Name |
Type |
Description |
Default |
source
|
AsyncIterable[bytes]
|
Asynchronous iterable yielding compressed bytes.
|
required
|
output_chunk_size
|
int
|
Strict upper bound for each yielded chunk.
|
DEFAULT_CHUNK_SIZE
|
max_decompressed_size
|
Optional[int]
|
Optional cumulative decompressed-byte limit.
|
None
|
Returns:
| Type |
Description |
AsyncIterator[bytes]
|
A single-consumer asynchronous iterator of decompressed bytes.
|
Raises:
| Type |
Description |
TypeError
|
If call-time arguments or a source item have invalid types.
|
ValueError
|
If a size argument is outside its supported range.
|
BadGzipFile
|
If the consumed gzip stream is malformed or corrupt.
|
OSError
|
If the cumulative output limit is exceeded.
|
Source code in src/aiogzip/__init__.py
| def decompress_chunks(
source: AsyncIterable[bytes],
*,
output_chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE,
max_decompressed_size: Optional[int] = None,
) -> AsyncIterator[bytes]:
"""Incrementally decompress gzip bytes from an asynchronous iterable.
Output chunks are non-empty and no larger than ``output_chunk_size``.
Complete CRC and trailer validation occurs only when the returned iterator
is consumed to completion.
Args:
source: Asynchronous iterable yielding compressed ``bytes``.
output_chunk_size: Strict upper bound for each yielded chunk.
max_decompressed_size: Optional cumulative decompressed-byte limit.
Returns:
A single-consumer asynchronous iterator of decompressed ``bytes``.
Raises:
TypeError: If call-time arguments or a source item have invalid types.
ValueError: If a size argument is outside its supported range.
gzip.BadGzipFile: If the consumed gzip stream is malformed or corrupt.
OSError: If the cumulative output limit is exceeded.
"""
return _decompress_chunks(
source,
output_chunk_size=output_chunk_size,
max_decompressed_size=max_decompressed_size,
)
|
engine_info
engine_info() -> EngineInfo
Return the default compression and active decompression engines.
Compression remains on stdlib zlib unless a writer opts into zlib-ng with
fast_compress=True. The returned strings are intended for diagnostics,
not machine-readable feature detection.
Source code in src/aiogzip/_engine.py
| def engine_info() -> EngineInfo:
"""Return the default compression and active decompression engines.
Compression remains on stdlib zlib unless a writer opts into zlib-ng with
``fast_compress=True``. The returned strings are intended for diagnostics,
not machine-readable feature detection.
"""
return EngineInfo(
compression="stdlib-zlib",
decompression="zlib-ng" if _HAVE_ZNG else "stdlib-zlib",
crc32="zlib-ng" if crc32 is not zlib.crc32 else "stdlib-zlib",
)
|
inspect
async
inspect(filename: _Filename, *, chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE, fileobj: _ReadFileObj = None, closefd: Optional[bool] = None, max_decompressed_size: Optional[int] = None) -> GzipInfo
Inspect and validate every member in a complete gzip stream.
This performs a full decompression scan while discarding payload bytes.
mtime preserves the literal header value, including zero; filename and
comment metadata use Latin-1 decoding.
Source code in src/aiogzip/__init__.py
| async def inspect(
filename: _Filename,
*,
chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE,
fileobj: _ReadFileObj = None,
closefd: Optional[bool] = None,
max_decompressed_size: Optional[int] = None,
) -> GzipInfo:
"""Inspect and validate every member in a complete gzip stream.
This performs a full decompression scan while discarding payload bytes.
``mtime`` preserves the literal header value, including zero; filename and
comment metadata use Latin-1 decoding.
"""
result = await _scan_gzip(
filename,
fileobj=fileobj,
closefd=closefd,
max_decompressed_size=max_decompressed_size,
chunk_size=chunk_size,
collect_members=True,
)
return GzipInfo(
members=result.members,
compressed_size=result.compressed_size,
uncompressed_size=result.uncompressed_size,
)
|
open
open(filename: _Filename, mode: _TextMode, *, chunk_size: int = ..., encoding: Optional[str] = ..., errors: Optional[str] = ..., newline: Optional[str] = ..., compresslevel: int = ..., mtime: Optional[Union[int, float]] = ..., original_filename: Optional[Union[str, bytes]] = ..., fileobj: _FileObj = ..., closefd: Optional[bool] = ..., max_decompressed_size: Optional[int] = ..., max_rewind_cache_size: Optional[int] = ..., strict_size: bool = ..., fast_compress: bool = ...) -> AsyncGzipTextFile
open(filename: _Filename, mode: _BinaryMode = ..., *, chunk_size: int = ..., compresslevel: int = ..., mtime: Optional[Union[int, float]] = ..., original_filename: Optional[Union[str, bytes]] = ..., fileobj: _FileObj = ..., closefd: Optional[bool] = ..., max_decompressed_size: Optional[int] = ..., max_rewind_cache_size: Optional[int] = ..., strict_size: bool = ..., fast_compress: bool = ...) -> AsyncGzipBinaryFile
open(filename: _Filename, mode: str, **kwargs: Any) -> Union[AsyncGzipBinaryFile, AsyncGzipTextFile]
open(filename: _Filename, mode: str = 'rb', **kwargs: Any) -> Union[AsyncGzipBinaryFile, AsyncGzipTextFile]
Open a gzip stream in binary or text mode.
This is the recommended public entry point. AsyncGzipFile remains
available and has identical behavior.
Source code in src/aiogzip/__init__.py
| def open(
filename: _Filename, mode: str = "rb", **kwargs: Any
) -> Union[AsyncGzipBinaryFile, AsyncGzipTextFile]:
"""Open a gzip stream in binary or text mode.
This is the recommended public entry point. ``AsyncGzipFile`` remains
available and has identical behavior.
"""
return AsyncGzipFile(filename, mode, **kwargs)
|
read
async
read(filename: _Filename, *, chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE, fileobj: _ReadFileObj = None, closefd: Optional[bool] = None, max_decompressed_size: Optional[int] = None, max_rewind_cache_size: Optional[int] = _MAX_CHUNK_SIZE) -> bytes
Read and decompress an entire gzip stream into memory.
Source code in src/aiogzip/__init__.py
| async def read(
filename: _Filename,
*,
chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE,
fileobj: _ReadFileObj = None,
closefd: Optional[bool] = None,
max_decompressed_size: Optional[int] = None,
max_rewind_cache_size: Optional[int] = _MAX_CHUNK_SIZE,
) -> bytes:
"""Read and decompress an entire gzip stream into memory."""
async with open(
filename,
"rb",
chunk_size=chunk_size,
fileobj=fileobj,
closefd=closefd,
max_decompressed_size=max_decompressed_size,
max_rewind_cache_size=max_rewind_cache_size,
) as stream:
return await stream.read()
|
verify
async
verify(filename: _Filename, *, chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE, fileobj: _ReadFileObj = None, closefd: Optional[bool] = None, max_decompressed_size: Optional[int] = None) -> VerificationResult
Validate a complete gzip stream and return aggregate counts.
Successful return means every header, deflate payload, CRC, and ISIZE
was valid. Invalid input and resource-limit failures raise instead.
Source code in src/aiogzip/__init__.py
| async def verify(
filename: _Filename,
*,
chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE,
fileobj: _ReadFileObj = None,
closefd: Optional[bool] = None,
max_decompressed_size: Optional[int] = None,
) -> VerificationResult:
"""Validate a complete gzip stream and return aggregate counts.
Successful return means every header, deflate payload, CRC, and ``ISIZE``
was valid. Invalid input and resource-limit failures raise instead.
"""
result = await _scan_gzip(
filename,
fileobj=fileobj,
closefd=closefd,
max_decompressed_size=max_decompressed_size,
chunk_size=chunk_size,
collect_members=False,
)
return VerificationResult(
member_count=result.member_count,
compressed_size=result.compressed_size,
uncompressed_size=result.uncompressed_size,
)
|
write
async
write(filename: _Filename, data: Union[bytes, bytearray, memoryview], *, chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE, compresslevel: int = 6, mtime: Optional[Union[int, float]] = None, original_filename: Optional[Union[str, bytes]] = None, fileobj: _WriteFileObj = None, closefd: Optional[bool] = None, strict_size: bool = False, fast_compress: bool = False) -> None
Compress and write an entire bytes-like payload to a gzip stream.
Source code in src/aiogzip/__init__.py
| async def write(
filename: _Filename,
data: Union[bytes, bytearray, memoryview],
*,
chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE,
compresslevel: int = 6,
mtime: Optional[Union[int, float]] = None,
original_filename: Optional[Union[str, bytes]] = None,
fileobj: _WriteFileObj = None,
closefd: Optional[bool] = None,
strict_size: bool = False,
fast_compress: bool = False,
) -> None:
"""Compress and write an entire bytes-like payload to a gzip stream."""
async with open(
filename,
"wb",
chunk_size=chunk_size,
compresslevel=compresslevel,
mtime=mtime,
original_filename=original_filename,
fileobj=fileobj,
closefd=closefd,
strict_size=strict_size,
fast_compress=fast_compress,
) as stream:
await stream.write(data)
|