httk.core.datastream.compression

Classes

CompressionCodec

A decompression codec for a single container format.

Functions

register_compression(→ None)	Register (or replace) a codec under its name (case-insensitive).
known_compressions(→ list[str])	Return the registered codec names, in registration order.
codec_for_name(→ CompressionCodec | None)	Return the codec whose extension matches the trailing suffix of name, else None.
split_compression_suffix(→ tuple[str, ...)	Split a trailing compression extension off name.
sniff_codec(→ tuple[io.IOBase, CompressionCodec | None])	Detect a codec from the leading magic bytes of stream without consuming data.
open_compressed(→ io.IOBase)	Return a decompressed view of stream according to the compression hint.
validate_compression(→ None)	Raise ValueError unless compression is a known mode or registered codec name.
reject_text_native_compression(→ None)	Validate a compression hint for a text-native source (an open text stream or a string).

Module Contents

class httk.core.datastream.compression.CompressionCodec

A decompression codec for a single container format.

A codec is an orthogonal layer below the datastream backends: it turns a compressed binary stream into an uncompressed binary stream, independently of where the compressed bytes come from (a filename, an open file, raw bytes, or a remote response).

name: str

Canonical, lower-case codec name (e.g. "gzip"); also how an explicit hint selects it.

extensions: tuple[str, Ellipsis]

Recognized filename suffixes including the leading dot (e.g. (".gz",)).

magics: tuple[bytes, Ellipsis]

Leading magic-byte signatures; an empty tuple means the format cannot be sniffed.

open_stream: collections.abc.Callable[[io.IOBase], io.IOBase]

Wrap a compressed binary stream and return a readable, decompressed binary stream.

httk.core.datastream.compression.register_compression(codec: CompressionCodec) → None

Register (or replace) a codec under its name (case-insensitive).

httk.core.datastream.compression.known_compressions() → list[str]

Return the registered codec names, in registration order.

httk.core.datastream.compression.codec_for_name(name: str) → CompressionCodec | None

Return the codec whose extension matches the trailing suffix of name, else None.

httk.core.datastream.compression.split_compression_suffix(name: str) → tuple[str, CompressionCodec | None]

Split a trailing compression extension off name.

"data.json.gz" becomes ("data.json", <gzip codec>); a name with no recognized compression extension is returned unchanged with None.

httk.core.datastream.compression.sniff_codec(stream: io.IOBase) → tuple[io.IOBase, CompressionCodec | None]

Detect a codec from the leading magic bytes of stream without consuming data.

A seekable stream is read and rewound; an unseekable stream is peeked (directly when it supports peek, otherwise via a wrapping io.BufferedReader). The returned stream must be used in place of the input, since it may be the wrapper.

httk.core.datastream.compression.open_compressed(stream: io.IOBase, *, compression: str = 'auto', name: str | None = None) → io.IOBase

Return a decompressed view of stream according to the compression hint.

Values are "none" (passthrough), "extension" (decide from name only), "detect" (always sniff magic bytes), "auto" (extension if recognized, else sniff), or a registered codec name (force that codec; an unknown name raises ValueError). When no codec applies the stream is returned unchanged.

httk.core.datastream.compression.validate_compression(compression: str) → None

Raise ValueError unless compression is a known mode or registered codec name.

httk.core.datastream.compression.reject_text_native_compression(compression: str | None) → None

Validate a compression hint for a text-native source (an open text stream or a string).

Such sources carry no compressed bytes to decode, so only the no-op modes "auto", "extension", and "none" are accepted; a codec name or "detect" raises ValueError. None (no hint given) is accepted.