Skip to content

Zstandard

cachetory.serializers.compressors.ZstdCompressor

Uses Zstandard compression.

Warning

This compressor requires zstd extra.

Source code in cachetory/serializers/compressors/zstd.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class ZstdCompressor(Serializer[bytes, bytes]):
    """
    Uses [Zstandard](https://facebook.github.io/zstd/) compression.

    Warning:
        This compressor requires [`zstd`](https://github.com/sergey-dryabzhinsky/python-zstd) extra.
    """

    __slots__ = ("_level", "_threads")

    @classmethod
    def from_url(cls, url: str) -> ZstdCompressor:
        """
        Construct serializer from the URL.

        # Supported schema's

        - `zstd://`
        - `zstandard://`

        # URL parameters

        | Parameter             |                                                                             |
        |-----------------------|-----------------------------------------------------------------------------|
        | `compression-level`   | [Compression level](https://github.com/sergey-dryabzhinsky/python-zstd#api) |
        | `compression-threads` | [Number of threads](https://github.com/sergey-dryabzhinsky/python-zstd#api) |
        """
        params = _UrlParams.model_validate(dict(parse_qsl(urlparse(url).query)))
        return cls(compression_level=params.compression_level, compression_threads=params.compression_threads)

    def __init__(
        self,
        *,
        compression_level: int = 3,
        compression_threads: int = 0,
    ) -> None:
        self._level = compression_level
        self._threads = compression_threads

    def serialize(self, value: bytes) -> bytes:
        return zstd.compress(value, self._level, self._threads)  # type: ignore[no-any-return]

    def deserialize(self, data: bytes) -> bytes:
        return zstd.decompress(data)  # type: ignore[no-any-return]

from_url classmethod 

from_url(url: str) -> ZstdCompressor

Construct serializer from the URL.

Supported schema's

  • zstd://
  • zstandard://

URL parameters

Parameter
compression-level Compression level
compression-threads Number of threads
Source code in cachetory/serializers/compressors/zstd.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@classmethod
def from_url(cls, url: str) -> ZstdCompressor:
    """
    Construct serializer from the URL.

    # Supported schema's

    - `zstd://`
    - `zstandard://`

    # URL parameters

    | Parameter             |                                                                             |
    |-----------------------|-----------------------------------------------------------------------------|
    | `compression-level`   | [Compression level](https://github.com/sergey-dryabzhinsky/python-zstd#api) |
    | `compression-threads` | [Number of threads](https://github.com/sergey-dryabzhinsky/python-zstd#api) |
    """
    params = _UrlParams.model_validate(dict(parse_qsl(urlparse(url).query)))
    return cls(compression_level=params.compression_level, compression_threads=params.compression_threads)