Skip to content

Compression API

Compression applies to the complete archive byte stream. none and gzip are always available; zstd requires the zstd extra.

glyphive.compression

Named whole-stream compression methods.

__all__ = ['CompressionMethod', 'GzipCompression', 'NoneCompression', 'ZstdCompression', 'available', 'default', 'get', 'names'] module-attribute

CompressionMethod

Bases: ABC

Base class for stateless, no-argument compression implementations.

name class-attribute

__init_subclass__(**kwargs)

Source code in src/glyphive/compression/_base.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init_subclass__(cls, **kwargs: _ty.Any) -> None:
    super().__init_subclass__(**kwargs)
    if any(
        getattr(getattr(cls, method, None), "__isabstractmethod__", False)
        for method in ("compress", "decompress")
    ):
        return
    name = getattr(cls, "name", None)
    if not isinstance(name, str) or not re.fullmatch(r"[a-z][a-z0-9_-]*", name):
        raise ValueError(
            f"compression {cls.__module__}.{cls.__qualname__} must define a "
            "valid lowercase ASCII name"
        )
    if name in CompressionMethod._registry:
        existing = CompressionMethod._registry[name]
        raise ValueError(
            f"duplicate compression name {name!r}: "
            f"{existing.__module__}.{existing.__qualname__} and "
            f"{cls.__module__}.{cls.__qualname__}"
        )
    CompressionMethod._registry[name] = cls

available() classmethod

Return method names whose optional backend is available.

Source code in src/glyphive/compression/_base.py 
44
45
46
47
48
49
50
51
@classmethod
def available(cls) -> _ty.List[str]:
    """Return method names whose optional backend is available."""
    return sorted(
        name
        for name, implementation in CompressionMethod._registry.items()
        if implementation.is_available()
    )

compress(data, level=None) abstractmethod

Compress one whole archive stream.

Source code in src/glyphive/compression/_base.py 
90
91
92
@abstractmethod
def compress(self, data: bytes, level: _ty.Optional[int] = None) -> bytes:
    """Compress one whole archive stream."""

compress_stream(source, sink, *, level=None, chunk_size=1024 * 1024)

Compress from source to sink.

External methods retain a compatibility fallback through their one-shot implementation. Built-ins override this method with bounded-memory I/O.

Source code in src/glyphive/compression/_base.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def compress_stream(
    self,
    source: _ty.BinaryIO,
    sink: _ty.BinaryIO,
    *,
    level: _ty.Optional[int] = None,
    chunk_size: int = 1024 * 1024,
) -> None:
    """Compress from ``source`` to ``sink``.

    External methods retain a compatibility fallback through their one-shot
    implementation. Built-ins override this method with bounded-memory I/O.
    """
    _validate_chunk_size(chunk_size)
    sink.write(self.compress(source.read(), level))

decompress(data) abstractmethod

Decompress one whole archive stream.

Source code in src/glyphive/compression/_base.py 
94
95
96
@abstractmethod
def decompress(self, data: bytes) -> bytes:
    """Decompress one whole archive stream."""

decompress_stream(source, sink, *, chunk_size=1024 * 1024)

Decompress from source to sink (compatibility fallback).

Source code in src/glyphive/compression/_base.py 
114
115
116
117
118
119
120
121
122
123
def decompress_stream(
    self,
    source: _ty.BinaryIO,
    sink: _ty.BinaryIO,
    *,
    chunk_size: int = 1024 * 1024,
) -> None:
    """Decompress from ``source`` to ``sink`` (compatibility fallback)."""
    _validate_chunk_size(chunk_size)
    sink.write(self.decompress(source.read()))

get(name) classmethod

Return a fresh no-argument method by wire name.

Source code in src/glyphive/compression/_base.py 
53
54
55
56
57
58
59
60
61
62
63
@classmethod
def get(cls, name: str) -> "CompressionMethod":
    """Return a fresh no-argument method by wire name."""
    try:
        implementation = CompressionMethod._registry[name]
    except (KeyError, TypeError):
        valid = ", ".join(CompressionMethod.names()) or "(none)"
        raise ValueError(
            f"unknown compression method {name!r}; available methods: {valid}"
        ) from None
    return implementation()

is_available() classmethod

Return whether this method can be selected without an extra.

Source code in src/glyphive/compression/_base.py 
65
66
67
68
@classmethod
def is_available(cls) -> bool:
    """Return whether this method can be selected without an extra."""
    return True

names() classmethod

Return all registered method names in stable order.

Source code in src/glyphive/compression/_base.py 
39
40
41
42
@classmethod
def names(cls) -> _ty.List[str]:
    """Return all registered method names in stable order."""
    return sorted(CompressionMethod._registry)

GzipCompression

Bases: CompressionMethod

name = 'gzip' class-attribute instance-attribute

compress(data, level=None)

Source code in src/glyphive/compression/gzip.py 
15
16
17
18
def compress(self, data: bytes, level: _ty.Optional[int] = None) -> bytes:
    source, sink = io.BytesIO(data), io.BytesIO()
    self.compress_stream(source, sink, level=level)
    return sink.getvalue()

compress_stream(source, sink, *, level=None, chunk_size=1024 * 1024)

Source code in src/glyphive/compression/gzip.py 
25
26
27
28
29
30
31
32
33
34
def compress_stream(self, source, sink, *, level=None, chunk_size=1024 * 1024):
    chunk_size = _validate_chunk_size(chunk_size)
    with _gzip.GzipFile(
        fileobj=sink,
        mode="wb",
        filename="",
        compresslevel=9 if level is None else level,
        mtime=0,
    ) as compressed:
        _copy_chunks(source, compressed, chunk_size=chunk_size)

decompress(data)

Source code in src/glyphive/compression/gzip.py 
20
21
22
23
def decompress(self, data: bytes) -> bytes:
    source, sink = io.BytesIO(data), io.BytesIO()
    self.decompress_stream(source, sink)
    return sink.getvalue()

decompress_stream(source, sink, *, chunk_size=1024 * 1024)

Source code in src/glyphive/compression/gzip.py 
36
37
38
39
def decompress_stream(self, source, sink, *, chunk_size=1024 * 1024):
    chunk_size = _validate_chunk_size(chunk_size)
    with _gzip.GzipFile(fileobj=source, mode="rb") as decompressed:
        _copy_chunks(decompressed, sink, chunk_size=chunk_size)

NoneCompression

Bases: CompressionMethod

name = 'none' class-attribute instance-attribute

compress(data, level=None)

Source code in src/glyphive/compression/none.py 
13
14
def compress(self, data: bytes, level: _ty.Optional[int] = None) -> bytes:
    return data

compress_stream(source, sink, *, level=None, chunk_size=1024 * 1024)

Source code in src/glyphive/compression/none.py 
19
20
21
def compress_stream(self, source, sink, *, level=None, chunk_size=1024 * 1024):
    del level
    _copy_chunks(source, sink, chunk_size=chunk_size)

decompress(data)

Source code in src/glyphive/compression/none.py 
16
17
def decompress(self, data: bytes) -> bytes:
    return data

decompress_stream(source, sink, *, chunk_size=1024 * 1024)

Source code in src/glyphive/compression/none.py 
23
24
def decompress_stream(self, source, sink, *, chunk_size=1024 * 1024):
    _copy_chunks(source, sink, chunk_size=chunk_size)

ZstdCompression

Bases: CompressionMethod

name = 'zstd' class-attribute instance-attribute

compress(data, level=None)

Source code in src/glyphive/compression/zstd.py 
44
45
46
47
def compress(self, data: bytes, level: _ty.Optional[int] = None) -> bytes:
    source, sink = io.BytesIO(data), io.BytesIO()
    self.compress_stream(source, sink, level=level)
    return sink.getvalue()

compress_stream(source, sink, *, level=None, chunk_size=1024 * 1024)

Source code in src/glyphive/compression/zstd.py 
54
55
56
57
58
59
60
61
62
63
def compress_stream(self, source, sink, *, level=None, chunk_size=1024 * 1024):
    chunk_size = _validate_chunk_size(chunk_size)
    compressor = _available_zstd_module().ZstdCompressor(
        level=3 if level is None else level,
        threads=0,
        write_checksum=True,
        write_content_size=False,
    )
    with compressor.stream_writer(sink, closefd=False) as writer:
        _copy_chunks(source, writer, chunk_size=chunk_size)

decompress(data)

Source code in src/glyphive/compression/zstd.py 
49
50
51
52
def decompress(self, data: bytes) -> bytes:
    source, sink = io.BytesIO(data), io.BytesIO()
    self.decompress_stream(source, sink)
    return sink.getvalue()

decompress_stream(source, sink, *, chunk_size=1024 * 1024)

Source code in src/glyphive/compression/zstd.py 
65
66
67
68
69
def decompress_stream(self, source, sink, *, chunk_size=1024 * 1024):
    chunk_size = _validate_chunk_size(chunk_size)
    decompressor = _available_zstd_module().ZstdDecompressor()
    with decompressor.stream_reader(source, closefd=False) as reader:
        _copy_chunks(reader, sink, chunk_size=chunk_size)

is_available() classmethod

Source code in src/glyphive/compression/zstd.py 
36
37
38
39
40
41
42
@classmethod
def is_available(cls) -> bool:
    try:
        _zstd_module()
    except RuntimeError:
        return False
    return True

available()

Source code in src/glyphive/compression/__init__.py 
32
33
def available() -> _ty.List[str]:
    return CompressionMethod.available()

default()

Prefer zstd when installed, otherwise use gzip.

Source code in src/glyphive/compression/__init__.py 
36
37
38
39
40
def default() -> str:
    """Prefer zstd when installed, otherwise use gzip."""
    if ZstdCompression.is_available():
        return "zstd"
    return "gzip"

get(name)

Source code in src/glyphive/compression/__init__.py 
24
25
def get(name: str) -> CompressionMethod:
    return CompressionMethod.get(name)

names()

Source code in src/glyphive/compression/__init__.py 
28
29
def names() -> _ty.List[str]:
    return CompressionMethod.names()