Skip to content

API Reference

aiogzip exposes its supported public API from the top-level package:

  • AsyncGzipBinaryFile — binary-mode reader/writer
  • AsyncGzipTextFile — text-mode reader/writer
  • open — 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 supported
  • read — read and decompress a complete binary stream into memory
  • write — compress and write a complete bytes-like payload
  • EngineInfo and engine_info — immutable diagnostic information about the default compression and active decompression engines
  • GzipMemberInfo, GzipInfo, and inspect — validated per-member metadata and aggregate sizes from a complete decompression scan
  • VerificationResult and verify — lightweight aggregate counts after complete integrity verification
  • decompress_chunks — pull-driven decompression from an AsyncIterable[bytes] with bounded output chunks
  • compress_chunks — one-member gzip compression from an AsyncIterable[bytes] with bounded output chunks
  • WithAsyncRead, 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__

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.

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.

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 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
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:
    # 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 = bool(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 = closefd if 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._engine: ZlibEngine = 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._is_closed: bool = False
    self._eof: bool = False
    self._owns_file: bool = False
    self._crc: int = 0
    self._input_size: int = 0
    self._position: int = 0
    self._mtime: Optional[int] = None
    self._header_probe_buffer = bytearray()
    self._compressed_cache = bytearray()
    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._read_broken: bool = False
    self._max_decompressed_size: Optional[int] = max_decompressed_size
    self._decompressed_total: int = 0
    self._strict_size: bool = bool(strict_size)
    self._saw_compressed_data: bool = False

closed property 

closed: bool

Return True when this file has been closed.

mtime property 

mtime: Optional[int]

Return the gzip member mtime after the header has been read.

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 
302
303
304
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 
306
307
308
309
310
311
312
313
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()

__aiter__ 

__aiter__() -> AsyncGzipBinaryFile

Make AsyncGzipBinaryFile iterable over newline-delimited chunks.

Source code in src/aiogzip/_binary.py 
1455
1456
1457
def __aiter__(self) -> "AsyncGzipBinaryFile":
    """Make AsyncGzipBinaryFile iterable over newline-delimited chunks."""
    return self

__anext__ async 

__anext__() -> bytes

Return the next line from the binary stream.

Source code in src/aiogzip/_binary.py 
1459
1460
1461
1462
1463
1464
1465
1466
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 

close() -> None

Flushes any remaining compressed data and closes the file.

Source code in src/aiogzip/_binary.py 
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
async def close(self) -> None:
    """Flushes any remaining compressed data and closes the file."""
    if self._is_closed:
        return

    # Mark as closed immediately to prevent concurrent close attempts
    self._is_closed = True

    close_file = (
        self._file
        if self._file is not None and (self._owns_file or self._closefd)
        else None
    )
    write_failed = False
    try:
        if self._writing_mode and self._file is not None and not self._write_broken:
            # Flush the compressor to write the gzip trailer. Skipped on
            # a broken writer because the member is already torn and a
            # trailer would lie about the bytes actually on disk.
            remaining_data = self._engine.flush()
            if remaining_data:
                await self._write_all(remaining_data)
            trailer = _build_gzip_trailer(self._crc, self._input_size)
            await self._write_all(trailer)
    except BaseException:
        write_failed = True
        raise
    finally:
        if close_file is not None:
            # Close only if we own it or closefd=True. Preserve a prior
            # final-write exception if close() also fails.
            close_method = getattr(close_file, "close", None)
            if callable(close_method):
                try:
                    result = close_method()
                    if hasattr(result, "__await__"):
                        await result
                except BaseException:
                    if not write_failed:
                        raise

detach 

detach() -> Any

Detach is unsupported to mirror gzip.GzipFile behavior.

Source code in src/aiogzip/_binary.py 
480
481
482
def detach(self) -> Any:
    """Detach is unsupported to mirror gzip.GzipFile behavior."""
    raise io.UnsupportedOperation("detach")

fileno 

fileno() -> int

Return the underlying file descriptor number.

Source code in src/aiogzip/_binary.py 
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
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() -> 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")

Source code in src/aiogzip/_binary.py 
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
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 self._writing_mode and self._write_broken:
        # Pretending the flush succeeded would tell the caller their
        # bytes are safely on disk when the member is already torn.
        raise OSError(
            "write stream is broken after a prior write failure; "
            "the gzip member is unusable"
        )

    if self._writing_mode and self._file is not None:
        # Flush any buffered compressed data (but not the final trailer)
        # Using Z_SYNC_FLUSH allows us to flush without ending the stream
        try:
            flushed_data = self._engine.flush(_engine.Z_SYNC_FLUSH)
            if flushed_data:
                try:
                    await self._write_all(flushed_data)
                except BaseException:
                    # Z_SYNC_FLUSH advanced the compressor. If its output did
                    # not reach the sink, no later bytes can repair the member.
                    self._write_broken = True
                    raise

            # Also flush the underlying file if it has a flush method
            flush_method = getattr(self._file, "flush", None)
            if callable(flush_method):
                result = flush_method()
                if hasattr(result, "__await__"):
                    await result
        except _engine.ZLIB_ERRORS as e:
            raise OSError(f"Error flushing compressed data: {e}") from e
        except OSError:
            raise
        except Exception as e:
            raise OSError(f"Unexpected error during flush: {e}") from e

isatty 

isatty() -> bool

Return True if the underlying stream is interactive.

Source code in src/aiogzip/_binary.py 
465
466
467
468
469
470
471
472
473
474
475
476
477
478
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 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
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).
    """
    _check_can_open(self._is_closed, self._file is not None)
    try:
        if self._external_file is not None:
            self._file = cast(Any, self._external_file)
            self._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
            self._file = await aiofiles.open(  # type: ignore
                self._filename, self._file_mode
            )
            self._owns_file = True

        # Initialize compression/decompression engine based on mode
        if self._writing_mode:
            self._engine = _engine.compressobj(
                self._compresslevel,
                -_engine.MAX_WBITS,
                fast=self._fast_compress,
            )
            header = _build_gzip_header(
                _derive_header_filename(
                    self._header_filename_override, self._filename
                ),
                self._header_mtime,
                self._compresslevel,
            )
            await self._write_all(header)
            self._crc = 0
            self._input_size = 0
        else:  # read mode
            self._engine = _engine.decompressobj(GZIP_WBITS)
            self._position = 0
            self._mtime = None
            self._header_probe_buffer.clear()
            self._compressed_cache.clear()
            self._replay_offset = None
            self._saw_compressed_data = False
            self._underlying_seekable = await self._probe_underlying_seekable()
            self._cache_rewindable_reads = not self._underlying_seekable

        return self
    except BaseException:
        # BaseException, not Exception: a task cancelled mid-open (e.g.
        # during the header write) must not leave _file set — the handle
        # would leak and every retry would hit "File is already open".
        await self._cleanup_failed_enter()
        raise

peek async 

peek(size: int = -1) -> bytes

Return up to size bytes without advancing the read position.

Source code in src/aiogzip/_binary.py 
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
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.")
    self._check_read_usable()
    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
    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
    end = self._buffer_offset + min(target, available)
    return bytes(self._buffer[self._buffer_offset : end])

raw 

raw() -> Any

Expose the underlying file object for advanced integrations.

Source code in src/aiogzip/_binary.py 
414
415
416
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:

Type Description
bytes

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

Source code in src/aiogzip/_binary.py 
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
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.")
    self._check_read_usable()

    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)

        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.
        while not self._eof:
            for piece in await self._decompress_next():
                chunks.append(piece)
                total_read += len(piece)

        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
        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 
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
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.")
    self._check_read_usable()

    if size is None:
        size = -1
    if size == 0:
        return b""

    available = len(self._buffer) - self._buffer_offset
    while available <= 0 and not self._eof:
        await self._fill_buffer()
        available = len(self._buffer) - self._buffer_offset

    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 
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
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.")
    self._check_read_usable()
    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.
    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 
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
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.")
    self._check_read_usable()
    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
    while available <= 0 and not self._eof:
        await self._fill_buffer()
        available = len(self._buffer) - self._buffer_offset
    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 
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
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.")
    self._check_read_usable()
    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""

    # Fast path: line fits in current buffer (common case)
    buf = self._buffer
    start = self._buffer_offset
    buf_len = len(buf)
    if start < buf_len:
        newline_index = buf.find(b"\n", start)
        if newline_index != -1:
            end = newline_index + 1
            if limit != -1:
                end = min(end, start + limit)
            result = bytes(memoryview(buf)[start:end])
            consumed = end - start
            self._buffer_offset = end
            self._position += consumed
            if end >= buf_len:
                del buf[:]
                self._buffer_offset = 0
            return result

    # Slow path: need more data or no newline in buffer
    chunks: List[bytes] = []
    total = 0
    while True:
        if self._buffer_offset >= len(self._buffer):
            if self._eof:
                break
            await self._fill_buffer()
            if self._buffer_offset >= len(self._buffer) and self._eof:
                break
            if self._buffer_offset >= len(self._buffer):
                continue

        start = self._buffer_offset
        end = len(self._buffer)
        newline_index = self._buffer.find(b"\n", start)
        if newline_index != -1:
            end = newline_index + 1
        if limit != -1:
            remaining = limit - total
            if remaining <= 0:
                break
            end = min(end, start + remaining)

        if end <= start:
            break

        chunk = bytes(memoryview(self._buffer)[start:end])
        chunks.append(chunk)
        consumed = end - start
        self._buffer_offset = end
        self._position += consumed
        total += consumed

        if self._buffer_offset >= len(self._buffer):
            del self._buffer[:]
            self._buffer_offset = 0

        if (newline_index != -1 and end == newline_index + 1) or (
            limit != -1 and total >= limit
        ):
            break

    return b"".join(chunks)

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 
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
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.")

    lines: List[bytes] = []
    total = 0
    while True:
        line = await self.readline()
        if not line:
            break
        lines.append(line)
        total += len(line)
        if hint > 0 and total >= hint:
            break
    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 
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
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 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
            while remaining > 0:
                chunk = (
                    zero_chunk
                    if remaining >= len(zero_chunk)
                    else zero_chunk[:remaining]
                )
                await self.write(chunk)
                remaining -= len(chunk)
        return self._position

    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:
        while not self._eof:
            await self._fill_buffer()
            buffered = len(self._buffer) - self._buffer_offset
            if buffered > 0:
                self._buffer_offset = len(self._buffer)
                self._position += buffered
                del self._buffer[:]
                self._buffer_offset = 0
        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()

    await self._consume_bytes(target - self._position)
    return self._position

tell async 

tell() -> int

Return the current uncompressed file position.

Source code in src/aiogzip/_binary.py 
347
348
349
350
351
async def tell(self) -> int:
    """Return the current uncompressed file position."""
    if self._is_closed:
        raise ValueError("I/O operation on closed file.")
    return self._position

truncate 

truncate(size: Optional[int] = None) -> int

Truncation is unsupported for gzip-compressed streams.

Source code in src/aiogzip/_binary.py 
484
485
486
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]

Bytes to write

 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 
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
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:
        raise OSError(
            "write stream is broken after a prior write failure; "
            "the gzip member is unusable"
        )

    # Declared explicitly so the two assignments share one type: the
    # bytes/bytearray fast path and the coerced memoryview path would
    # otherwise infer incompatible types under newer mypy (which treats
    # memoryview as generic, memoryview[int]).
    buffer: Union[bytes, bytearray, memoryview]
    if isinstance(data, (bytes, bytearray)):
        buffer = data
    else:
        buffer = self._coerce_byteslike(data)
    length = len(buffer)

    if self._strict_size and self._input_size + length > 0xFFFFFFFF:
        raise OSError(
            f"uncompressed member size would exceed the gzip ISIZE "
            f"field's 4 GiB limit ({self._input_size} + {length} > "
            f"{0xFFFFFFFF}); drop strict_size to allow ISIZE "
            f"truncation or split the payload into multiple members"
        )

    # Compress first. If compress() raises, the compressor's state is
    # intact (no output was emitted) and we can leave our accounting
    # untouched. If it succeeds but the downstream file write fails,
    # the compressor *has* advanced past these bytes, so the gzip
    # member is no longer recoverable — mark the stream broken.
    try:
        if length >= _ZLIB_OFFLOAD_THRESHOLD:
            # Pass bytes into the thread — bytearray/memoryview are
            # not guaranteed safe across thread boundaries while we
            # may still mutate the caller's bytearray.
            payload = bytes(buffer) if not isinstance(buffer, bytes) else buffer
            try:
                compressed = await _run_zlib_in_thread(
                    self._engine.compress, payload
                )
            except asyncio.CancelledError:
                # Cancelling this await does not stop the executor
                # thread: compress() may still run and advance the
                # shared compressor past bytes we never accounted
                # for, so the member is no longer recoverable.
                self._write_broken = True
                raise
        else:
            compressed = self._engine.compress(buffer)
    except _engine.ZLIB_ERRORS as e:
        raise OSError(f"Error compressing data: {e}") from e
    except Exception as e:
        raise OSError(f"Unexpected error during compression: {e}") from e

    if compressed:
        try:
            await self._write_all(compressed)
        except BaseException:
            # File write failed after compressor already consumed input;
            # further writes would produce a torn member.
            self._write_broken = True
            raise

    # Only credit the input once both stages succeed, and mask the CRC
    # to 32 bits so tell()/the trailer always see the same uint32 zlib
    # would have produced.
    self._crc = _engine.crc32(buffer, self._crc) & 0xFFFFFFFF
    self._input_size += length
    self._position = self._input_size

    return length

writelines async 

writelines(lines: Iterable[bytes]) -> None

Write bytes-like lines in bounded batches without adding separators.

Source code in src/aiogzip/_binary.py 
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
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.")

    pending = bytearray()
    iterator = iter(lines)
    while True:
        try:
            line = next(iterator)
        except StopIteration:
            break
        except BaseException:
            if pending:
                await self.write(pending)
            raise

        try:
            data = self._coerce_byteslike(line)
        except BaseException:
            if pending:
                await self.write(pending)
            raise

        length = len(data)
        if length >= self._chunk_size:
            if pending:
                await self.write(pending)
                pending.clear()
            await self.write(data)
        else:
            if pending and len(pending) + length > self._chunk_size:
                await self.write(pending)
                pending.clear()
            pending.extend(data)

    if pending:
        await self.write(pending)

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 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
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:
    # 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 = closefd if 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._is_closed: bool = False

    # 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)
    initial_decoder_state = self._decoder.getstate()
    self._buffer_origin_offset: int = 0
    self._buffer_origin_decoder_state: Tuple[Any, int] = initial_decoder_state
    self._buffer_origin_trailing_cr: bool = False
    self._buffer_origin_seen_newline_types: int = 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: bool = bool(strict_size)
    self._fast_compress: bool = bool(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"
        # Unsubscripted re.Pattern: re.Pattern[str] is not subscriptable at
        # runtime on Python 3.8, and an attribute annotation is evaluated.
        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 

closed: bool

Return True when this file has been closed.

encoding property 

encoding: str

Return the configured text encoding.

errors property 

errors: str

Return the configured text error handler.

mtime property 

mtime: Optional[int]

Return the gzip member mtime after the header has been read.

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 
332
333
334
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 
336
337
338
339
340
341
342
343
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()

__aiter__ 

__aiter__() -> AsyncGzipTextFile

Make AsyncGzipTextFile iterable for line-by-line reading.

Source code in src/aiogzip/_text.py 
1341
1342
1343
def __aiter__(self) -> "AsyncGzipTextFile":
    """Make AsyncGzipTextFile iterable for line-by-line reading."""
    return self

__anext__ async 

__anext__() -> str

Return the next line from the file.

Source code in src/aiogzip/_text.py 
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
async def __anext__(self) -> str:
    """Return the next line from the file."""
    if self._is_closed:
        raise StopAsyncIteration

    if self._line_term is not None:
        # Keep this inline block in sync with _readline_fast; a helper call
        # in this per-line hot path measurably lowers iteration throughput.
        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
        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
        has_more = await self._read_chunk_and_decode()
        if not has_more:
            remaining = len(self._text_buffer) - self._text_buffer_offset
            if remaining > 0:
                return self._consume_buffer(remaining)
            raise StopAsyncIteration
        search_from = max(0, buf_len - 1)

close async 

close() -> None

Closes the file.

Source code in src/aiogzip/_text.py 
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
async def close(self) -> None:
    """Closes the file."""
    if self._is_closed:
        return

    # Mark as closed immediately to prevent concurrent close attempts
    self._is_closed = True

    binary_file = self._binary_file
    if binary_file is None:
        return

    encoder_failed = False
    try:
        if self._writing_mode and self._encoder_used:
            encoder = self._encoder
            assert encoder is not None
            final_data = encoder.encode("", final=True)
            if final_data:
                await binary_file.write(final_data)
    except BaseException:
        encoder_failed = True
        raise
    finally:
        try:
            await binary_file.close()
        except BaseException:
            if not encoder_failed:
                raise

detach 

detach() -> Any

Detach is unsupported to mirror gzip.GzipFile behavior.

Source code in src/aiogzip/_text.py 
465
466
467
def detach(self) -> Any:
    """Detach is unsupported to mirror gzip.GzipFile behavior."""
    raise io.UnsupportedOperation("detach")

flush async 

flush() -> 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")

Source code in src/aiogzip/_text.py 
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
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:
        await self._binary_file.flush()

isatty 

isatty() -> bool

Return True if the underlying stream is interactive.

Source code in src/aiogzip/_text.py 
459
460
461
462
463
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

If hint is not positive.
Source code in src/aiogzip/_text.py 
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
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 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
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).
    """
    _check_can_open(self._is_closed, self._binary_file is not None)
    filename = os.fspath(self._filename) if self._filename is not None else None
    self._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 self._binary_file.open()
    except BaseException:
        # BaseException, not Exception: a cancelled open must not leave
        # _binary_file set, or the instance wedges on "File is already
        # open" at the next attempt.
        try:
            await self._binary_file.close()
        except Exception:
            pass
        self._binary_file = None
        raise
    return self

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:

Type Description
str

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

Source code in src/aiogzip/_text.py 
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
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 size is None:
        size = -1
    if size < 0:
        size = -1

    # Handle read(0) - should return empty string without draining buffer
    if size == 0:
        return ""

    # read() consumes the buffer by character count; drop any batched
    # pending lines so they cannot serve stale content afterwards. The
    # synced offset means the buffer remainder is read correctly regardless.
    if self._pending_lines:
        self._pending_lines = []
        self._pending_idx = 0

    if size == -1:
        # Fast path: drain buffer, then read all remaining binary data
        # in one shot to avoid per-chunk buffer append/consume overhead
        chunks = []
        buf = self._text_buffer
        off = self._text_buffer_offset
        if off < len(buf):
            chunks.append(buf[off:] if off else buf)
            self._text_buffer = ""
            self._text_buffer_offset = 0

        bf = self._binary_file
        decoder = self._decoder
        apply_nl = self._apply_newline_decoding

        raw_all = await bf.read(-1)
        if raw_all:
            decoded = decoder.decode(raw_all, final=False)
            if decoded:
                chunks.append(apply_nl(decoded))

        final = decoder.decode(b"", final=True)
        if final:
            chunks.append(apply_nl(final))
        self._finalize_pending_newline_state()

        return "".join(chunks)
    else:
        # Serve from already-buffered text first, then pull freshly decoded
        # chunks into a local list and join once. Appending each chunk to
        # the str buffer via _append_buffer is O(n^2) for large sized reads
        # (CPython can't do in-place += on a slotted attribute); local
        # accumulation keeps this O(n).
        result_parts: List[str] = []
        need = size

        buffered = self._buffered_text_len()
        if buffered:
            take = min(need, buffered)
            result_parts.append(self._consume_buffer(take))
            need -= take

        while need > 0:
            text, more = await self._decode_next_chunk()
            if text:
                if len(text) <= need:
                    result_parts.append(text)
                    need -= len(text)
                else:
                    # Keep the whole chunk in the buffer so its replay
                    # origin stays valid for tell()/seek(), then consume
                    # exactly `need` characters from it.
                    self._set_buffer(text)
                    result_parts.append(self._consume_buffer(need))
                    need = 0
                    break
            if not more:
                break

        return "".join(result_parts)

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 
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
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 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:
        return await self._readline_fast()

    # 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

    # Try to get a line from our buffer using newline-aware search
    search_from = 0
    while True:
        pos, length = self._find_line_terminator(search_from)

        if pos != -1:
            end = pos + length
            if limit != -1 and end > limit:
                return self._consume_buffer(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)

        has_more = await self._read_chunk_and_decode()
        if not has_more:
            if buf_len == 0:
                return ""
            remaining = len(self._text_buffer) - self._text_buffer_offset
            if limit != -1 and remaining > limit:
                return self._consume_buffer(limit)
            return self._consume_buffer(remaining)
        search_from = max(0, buf_len - 1)

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]

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

Source code in src/aiogzip/_text.py 
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
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._line_term is not None:
        return await self._readlines_fast(hint)

    lines: List[str] = []
    total_size = 0

    while True:
        line = await self.readline()
        if not line:
            break
        lines.append(line)
        total_size += len(line)
        if hint > 0 and total_size >= hint:
            break

    return lines

truncate 

truncate(size: Optional[int] = None) -> int

Truncation is unsupported for gzip-compressed streams.

Source code in src/aiogzip/_text.py 
469
470
471
def truncate(self, size: Optional[int] = None) -> int:
    """Truncation is unsupported for gzip-compressed streams."""
    raise io.UnsupportedOperation("truncate")

write async 

write(data: str) -> int

Encodes and writes text data to the file.

Parameters:

Name Type Description Default
data str

String to write

 required

Examples:

async with AsyncGzipTextFile("file.gz", "wt") as f: await f.write("Hello, World!") # String input

Source code in src/aiogzip/_text.py 
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
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 one encoder for the lifetime of the stream. Stateless str.encode()
    # would emit a fresh BOM or reset sequence on every write for encodings
    # such as UTF-16 and ISO-2022-JP.
    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)
        await self._binary_file.write(encoded_data)
    except BaseException:
        # Preserve retryability when the binary layer rejected the write
        # before consuming it (for example strict_size validation).
        encoder.setstate(encoder_state)
        self._encoder_used = encoder_was_used
        raise
    return len(data)

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 
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
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.")

    pending: List[str] = []
    pending_chars = 0
    iterator = iter(lines)
    while True:
        try:
            line = next(iterator)
        except StopIteration:
            break
        except BaseException:
            if pending:
                await self.write("".join(pending))
            raise

        if not isinstance(line, str):
            if pending:
                await self.write("".join(pending))
            await self.write(line)
            continue

        length = len(line)
        if length >= self._chunk_size:
            if pending:
                await self.write("".join(pending))
                pending = []
                pending_chars = 0
            await self.write(line)
        else:
            if pending and pending_chars + length > self._chunk_size:
                await self.write("".join(pending))
                pending = []
                pending_chars = 0
            pending.append(line)
            pending_chars += length

    if pending:
        await self.write("".join(pending))

EngineInfo dataclass 

EngineInfo(compression: str, decompression: str, crc32: str = 'stdlib-zlib')

Human-readable codec selections used by aiogzip by default.

GzipInfo dataclass 

GzipInfo(members: Tuple[GzipMemberInfo, ...], compressed_size: int, uncompressed_size: int)

Aggregate information for a completely validated gzip stream.

member_count property 

member_count: int

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: Optional[str], comment: Optional[str], extra: Optional[bytes], 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

Path to the file

 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:

Type Description
Union[AsyncGzipBinaryFile, AsyncGzipTextFile]

AsyncGzipBinaryFile for binary modes ('rb', 'wb', 'ab', 'xb')
Union[AsyncGzipBinaryFile, AsyncGzipTextFile]

AsyncGzipTextFile for text modes ('rt', 'wt', 'at', 'xt')
Source code in src/aiogzip/__init__.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
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 
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
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 
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
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 
132
133
134
135
136
137
138
139
140
141
142
143
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 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
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 
257
258
259
260
261
262
263
264
265
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 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
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 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
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 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
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)