API Reference

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

AsyncGzipBinaryFile — binary-mode reader/writer
AsyncGzipTextFile — text-mode reader/writer
AsyncGzipFile — factory returning the right class for a mode string (accepts r/w/a/x ops with a b or t suffix)
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, and aiogzip._text. Treat those modules as private and unstable.

`seek()` and `tell()` in text mode

AsyncGzipBinaryFile.tell() returns the current position as a plain non-negative count of decompressed bytes, and seek(offset) accepts any such offset.

AsyncGzipTextFile.tell() returns one of two things:

A plain non-negative offset (decompressed bytes) when the stream is at a clean boundary — no buffered text, the decoder holds no partial multibyte sequence, and there is no pending \r. This is the same value the underlying binary layer reports (await f.buffer.tell()).
An opaque cookie (a negative integer) otherwise. The cookie encodes the decoder state needed to resume mid-character, mid-line, or mid-\r\n, so round-tripping seek(await f.tell()) is exact.

seek() accepts both: a non-negative argument is treated as a plain offset (decompression is replayed forward to that byte — from the current position when the decoder is at a clean boundary at or behind the target, from the start of the stream otherwise), and a negative argument is decoded as a cookie.

Cookies are bound to the handle that minted them

Warning. A text tell() cookie is valid only on the same open handle. This differs from io.TextIOWrapper and gzip.open("rt"), whose tell() cookies encode only decoder state and remain usable after re-opening the same file. An aiogzip text cookie embeds a random per-instance nonce, which seek() validates; a cookie from a different handle (or from before the file was re-opened) is rejected with OSError rather than silently restoring the wrong decoder state. Do not persist cookies across processes or re-opens — persist a plain offset instead.

Resumable processing recipe

Because cookies are not portable, checkpoint progress as a plain offset taken at a line boundary, where the decoder is guaranteed clean (\n is single-byte, so it never splits a multibyte character). Drive the binary layer — which splits lines without the text layer's read-ahead — so await f.tell() is an exact decompressed byte offset, then resume in a new process by opening in "rt" and seeking to that offset.

A forward plain seek() is O(n) in the target offset: gzip has no random access, so aiogzip replays decompressed bytes up to the offset. Checkpoint at a coarse granularity if that cost matters. See Resumable text processing for the full worked example.

aiogzip

Async gzip file reader/writer public API.

AsyncGzipBinaryFile

AsyncGzipBinaryFile(filename: Union[str, bytes, Path, None], mode: str = 'rb', chunk_size: int = DEFAULT_CHUNK_SIZE, compresslevel: int = 6, mtime: Optional[Union[int, float]] = None, original_filename: Optional[Union[str, bytes]] = None, fileobj: Optional[Union[WithAsyncRead, WithAsyncWrite, WithAsyncReadWrite]] = None, closefd: Optional[bool] = None, max_decompressed_size: Optional[int] = None, max_rewind_cache_size: Optional[int] = _MAX_CHUNK_SIZE, strict_size: bool = False, fast_compress: bool = False)

An asynchronous gzip file reader/writer for binary data.

This class provides async gzip compression/decompression for binary data, making it a drop-in replacement for gzip.open() in binary mode.

Features: - Full compatibility with gzip.open() file format - Binary mode only (no text encoding/decoding) - Async context manager support - Configurable chunk size for performance tuning

Basic Usage

Write binary data

async with AsyncGzipBinaryFile("data.gz", "wb") as f: await f.write(b"Hello, World!")

Read binary data

async with AsyncGzipBinaryFile("data.gz", "rb") as f: data = await f.read() # Returns bytes

Imperative lifecycle when `async with` is impractical

f = AsyncGzipBinaryFile("data.gz", "rb") await f.open() try: data = await f.read() finally: await f.close()

Interoperability with gzip.open(): # Files created by AsyncGzipBinaryFile can be read by gzip.open() async with AsyncGzipBinaryFile("data.gz", "wb") as f: await f.write(b"data")

with gzip.open("data.gz", "rb") as f:
    data = f.read()  # Works perfectly!

Source code in src/aiogzip/_binary.py

def __init__(
    self,
    filename: Union[str, bytes, Path, None],
    mode: str = "rb",
    chunk_size: int = DEFAULT_CHUNK_SIZE,
    compresslevel: int = 6,
    mtime: Optional[Union[int, float]] = None,
    original_filename: Optional[Union[str, bytes]] = None,
    fileobj: Optional[
        Union[WithAsyncRead, WithAsyncWrite, WithAsyncReadWrite]
    ] = None,
    closefd: Optional[bool] = None,
    max_decompressed_size: Optional[int] = None,
    max_rewind_cache_size: Optional[int] = _MAX_CHUNK_SIZE,
    strict_size: bool = False,
    fast_compress: bool = False,
) -> None:
    # Validate inputs using shared validation functions
    _validate_filename(filename, fileobj)
    _validate_chunk_size(chunk_size)
    if max_decompressed_size is not None and max_decompressed_size <= 0:
        raise ValueError("max_decompressed_size must be a positive integer")
    if max_rewind_cache_size is not None and max_rewind_cache_size <= 0:
        raise ValueError("max_rewind_cache_size must be a positive integer")

    # 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._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

async def __aenter__(self) -> "AsyncGzipBinaryFile":
    """Enter the async context manager and initialize resources."""
    return await self.open()

aexit `async`

__aexit__(exc_type: Optional[type], exc_val: Optional[BaseException], exc_tb: Optional[Any]) -> None

Exit the context manager, flushing and closing the file.

Source code in src/aiogzip/_binary.py

async def __aexit__(
    self,
    exc_type: Optional[type],
    exc_val: Optional[BaseException],
    exc_tb: Optional[Any],
) -> None:
    """Exit the context manager, flushing and closing the file."""
    await self.close()

aiter

__aiter__() -> AsyncGzipBinaryFile

Make AsyncGzipBinaryFile iterable over newline-delimited chunks.

Source code in src/aiogzip/_binary.py

def __aiter__(self) -> "AsyncGzipBinaryFile":
    """Make AsyncGzipBinaryFile iterable over newline-delimited chunks."""
    return self

anext `async`

__anext__() -> bytes

Return the next line from the binary stream.

Source code in src/aiogzip/_binary.py

async def __anext__(self) -> bytes:
    """Return the next line from the binary stream."""
    if self._is_closed:
        raise StopAsyncIteration
    line = await self.readline()
    if line == b"":
        raise StopAsyncIteration
    return line

close `async`

close() -> None

Flushes any remaining compressed data and closes the file.

Source code in src/aiogzip/_binary.py

async def close(self) -> None:
    """Flushes any remaining compressed data and closes the file."""
    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._file.write(remaining_data)
            trailer = _build_gzip_trailer(self._crc, self._input_size)
            await self._file.write(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

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

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

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:
                await self._file.write(flushed_data)

            # 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

def isatty(self) -> bool:
    """Return True if the underlying stream is interactive."""
    if self._file is None:
        return False
    isatty_method = getattr(self._file, "isatty", None)
    if not callable(isatty_method):
        return False
    result = isatty_method()
    if hasattr(result, "__await__"):
        close_method = getattr(result, "close", None)
        if callable(close_method):
            close_method()
        return False
    return bool(result)

open `async`

open() -> AsyncGzipBinaryFile

Open the file for I/O and return self.

This performs the same initialization as entering the async context manager, for callers who prefer an explicit try/finally over async with::

f = AsyncGzipBinaryFile("data.gz", "rb")
await f.open()
try:
    data = await f.read()
finally:
    await f.close()

Raises:

Type	Description
`ValueError`	if the file is already open, or has already been closed (a closed instance cannot be reopened, matching io objects).

Source code in src/aiogzip/_binary.py

async def open(self) -> "AsyncGzipBinaryFile":
    """Open the file for I/O and return ``self``.

    This performs the same initialization as entering the async context
    manager, for callers who prefer an explicit try/finally over ``async
    with``::

        f = AsyncGzipBinaryFile("data.gz", "rb")
        await f.open()
        try:
            data = await f.read()
        finally:
            await f.close()

    Raises:
        ValueError: if the file is already open, or has already been closed
            (a closed instance cannot be reopened, matching io objects).
    """
    _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._file.write(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

async def peek(self, size: int = -1) -> bytes:
    """Return up to size bytes without advancing the read position."""
    if self._mode_op != "r":
        raise OSError("File not open for reading")
    if self._is_closed:
        raise ValueError("I/O operation on closed file.")
    if self._file is None:
        raise ValueError("File not opened. Call await open() or use async with.")
    if 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

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

async def read(self, size: int = -1) -> bytes:
    """
    Reads and decompresses binary data from the file.

    Args:
        size: Number of bytes to read (-1 for all remaining data)

    Returns:
        bytes

    Examples:
        async with AsyncGzipBinaryFile("file.gz", "rb") as f:
            data = await f.read()  # Returns bytes
            partial = await f.read(100)  # Returns first 100 bytes
    """
    if self._mode_op != "r":
        raise OSError("File not open for reading")
    if self._is_closed:
        raise ValueError("I/O operation on closed file.")
    if self._file is None:
        raise ValueError("File not opened. Call await open() or use async with.")

    if 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

async def read1(self, size: int = -1) -> bytes:
    """Read up to size bytes with at most one data-producing fill.

    A single compressed chunk can decode to nothing (e.g. while consuming
    the gzip header), so fills repeat until at least one byte is available;
    like stdlib gzip's ``read1()``, an empty result means EOF.
    """
    if self._mode_op != "r":
        raise OSError("File not open for reading")
    if self._is_closed:
        raise ValueError("I/O operation on closed file.")
    if self._file is None:
        raise ValueError("File not opened. Call await open() or use async with.")

    if 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

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.")
    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

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.")
    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

async def readline(self, limit: int = -1) -> bytes:
    """Read and return one line from the binary stream."""
    if self._mode_op != "r":
        raise OSError("File not open for reading")
    if self._is_closed:
        raise ValueError("I/O operation on closed file.")
    if self._file is None:
        raise ValueError("File not opened. Call await open() or use async with.")
    if 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

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

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

    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

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

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

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._file.write(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 a sequence of bytes-like lines to the binary stream.

Source code in src/aiogzip/_binary.py

async def writelines(self, lines: Iterable[bytes]) -> None:
    """Write a sequence of bytes-like lines to the binary stream."""
    if not self._writing_mode:
        raise OSError("File not open for writing")
    if self._is_closed:
        raise ValueError("I/O operation on closed file.")

    for line in lines:
        await self.write(line)

AsyncGzipTextFile

AsyncGzipTextFile(filename: Union[str, bytes, Path, None], mode: str = 'rt', chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE, encoding: Optional[str] = 'utf-8', errors: Optional[str] = 'strict', newline: Union[str, None] = None, compresslevel: int = 6, mtime: Optional[Union[int, float]] = None, original_filename: Optional[Union[str, bytes]] = None, fileobj: Optional[Union[WithAsyncRead, WithAsyncWrite, WithAsyncReadWrite]] = None, closefd: Optional[bool] = None, max_decompressed_size: Optional[int] = None, max_rewind_cache_size: Optional[int] = _MAX_CHUNK_SIZE, strict_size: bool = False, fast_compress: bool = False)

An asynchronous gzip file reader/writer for text data.

This class wraps AsyncGzipBinaryFile and provides text mode operations with proper UTF-8 handling for multi-byte characters.

Features: - Full compatibility with gzip.open() file format - Text mode with automatic encoding/decoding - Proper handling of multi-byte UTF-8 characters - Line-by-line iteration support - Async context manager support

Basic Usage

Write text data

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

Read text data

async with AsyncGzipTextFile("data.gz", "rt") as f: text = await f.read() # Returns string

Line-by-line iteration

async with AsyncGzipTextFile("data.gz", "rt") as f: async for line in f: print(line.strip())

Imperative lifecycle when `async with` is impractical

f = AsyncGzipTextFile("data.gz", "rt") await f.open() try: text = await f.read() finally: await f.close()

Source code in src/aiogzip/_text.py

def __init__(
    self,
    filename: Union[str, bytes, Path, None],
    mode: str = "rt",
    chunk_size: int = AsyncGzipBinaryFile.DEFAULT_CHUNK_SIZE,
    encoding: Optional[str] = "utf-8",
    errors: Optional[str] = "strict",
    newline: Union[str, None] = None,
    compresslevel: int = 6,
    mtime: Optional[Union[int, float]] = None,
    original_filename: Optional[Union[str, bytes]] = None,
    fileobj: Optional[
        Union[WithAsyncRead, WithAsyncWrite, WithAsyncReadWrite]
    ] = None,
    closefd: Optional[bool] = None,
    max_decompressed_size: Optional[int] = None,
    max_rewind_cache_size: Optional[int] = _MAX_CHUNK_SIZE,
    strict_size: bool = False,
    fast_compress: bool = False,
) -> None:
    # Validate inputs using shared validation functions
    _validate_filename(filename, fileobj)
    _validate_chunk_size(chunk_size)
    if max_decompressed_size is not None and max_decompressed_size <= 0:
        raise ValueError("max_decompressed_size must be a positive integer")
    if max_rewind_cache_size is not None and max_rewind_cache_size <= 0:
        raise ValueError("max_rewind_cache_size must be a positive integer")

    # 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._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
        )
    else:
        self._line_term = None
        self._line_split_re = 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

async def __aenter__(self) -> "AsyncGzipTextFile":
    """Enter the async context manager and initialize resources."""
    return await self.open()

aexit `async`

__aexit__(exc_type: Optional[type], exc_val: Optional[BaseException], exc_tb: Optional[Any]) -> None

Exit the context manager, flushing and closing the file.

Source code in src/aiogzip/_text.py

async def __aexit__(
    self,
    exc_type: Optional[type],
    exc_val: Optional[BaseException],
    exc_tb: Optional[Any],
) -> None:
    """Exit the context manager, flushing and closing the file."""
    await self.close()

aiter

__aiter__() -> AsyncGzipTextFile

Make AsyncGzipTextFile iterable for line-by-line reading.

Source code in src/aiogzip/_text.py

def __aiter__(self) -> "AsyncGzipTextFile":
    """Make AsyncGzipTextFile iterable for line-by-line reading."""
    return self

anext `async`

__anext__() -> str

Return the next line from the file.

Source code in src/aiogzip/_text.py

async def __anext__(self) -> str:
    """Return the next line from the file."""
    if self._is_closed:
        raise StopAsyncIteration

    if self._line_term is not None:
        # Inlined pending-line pop (canonical copy: _take_pending_line);
        # kept inline in this hot path for throughput. Keep copies in sync.
        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

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

    if self._binary_file is not None:
        await self._binary_file.close()

detach

detach() -> Any

Detach is unsupported to mirror gzip.GzipFile behavior.

Source code in src/aiogzip/_text.py

def detach(self) -> Any:
    """Detach is unsupported to mirror gzip.GzipFile behavior."""
    raise io.UnsupportedOperation("detach")

flush `async`

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

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

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

open `async`

open() -> AsyncGzipTextFile

Open the file for I/O and return self.

This performs the same initialization as entering the async context manager, for callers who prefer an explicit try/finally over async with::

f = AsyncGzipTextFile("data.txt.gz", "rt")
await f.open()
try:
    text = await f.read()
finally:
    await f.close()

Raises:

Type	Description
`ValueError`	if the file is already open, or has already been closed (a closed instance cannot be reopened, matching io objects).

Source code in src/aiogzip/_text.py

async def open(self) -> "AsyncGzipTextFile":
    """Open the file for I/O and return ``self``.

    This performs the same initialization as entering the async context
    manager, for callers who prefer an explicit try/finally over ``async
    with``::

        f = AsyncGzipTextFile("data.txt.gz", "rt")
        await f.open()
        try:
            text = await f.read()
        finally:
            await f.close()

    Raises:
        ValueError: if the file is already open, or has already been closed
            (a closed instance cannot be reopened, matching io objects).
    """
    _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

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

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

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

    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

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

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

    # Encode string to bytes
    encoded_data = text_to_encode.encode(self._encoding, errors=self._errors)
    await self._binary_file.write(encoded_data)
    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

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.")

    for line in lines:
        await self.write(line)

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

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)

API Reference

seek() and tell() in text mode

Cookies are bound to the handle that minted them

Resumable processing recipe

aiogzip

AsyncGzipBinaryFile

Write binary data

Read binary data

Imperative lifecycle when async with is impractical

closed property

mtime property

name property

__aenter__ async

__aexit__ async

__aiter__

__anext__ async

close async

detach

fileno

flush async

isatty

open async

peek async

raw

read async

read1 async

readinto async

readinto1 async

readline async

readlines async

seek async

tell async

truncate

write async

writelines async

AsyncGzipTextFile

Write text data

Read text data

Line-by-line iteration

Imperative lifecycle when async with is impractical

buffer property

closed property

encoding property

errors property

mtime property

name property

newlines property

__aenter__ async

__aexit__ async

__aiter__

__anext__ async

close async

detach

flush async

isatty

open async

read async

readline async

readlines async

With size hint

truncate

write async

writelines async

From a generator

WithAsyncRead

WithAsyncReadWrite

WithAsyncWrite

AsyncGzipFile

`seek()` and `tell()` in text mode

Imperative lifecycle when `async with` is impractical

closed `property`

mtime `property`

name `property`

aenter `async`

aexit `async`

aiter

anext `async`

close `async`

flush `async`

open `async`

peek `async`

read `async`

read1 `async`

readinto `async`

readinto1 `async`

readline `async`

readlines `async`

seek `async`

tell `async`

write `async`

writelines `async`

Imperative lifecycle when `async with` is impractical

buffer `property`

closed `property`

encoding `property`

errors `property`

mtime `property`

name `property`

newlines `property`

aenter `async`

aexit `async`

aiter

anext `async`

close `async`

flush `async`

open `async`

read `async`

readline `async`

readlines `async`

write `async`

writelines `async`