Source code for eventsourcing.compressor

from __future__ import annotations

import zlib

from eventsourcing.persistence import Compressor



[docs]
class ZlibCompressor(Compressor):

[docs]
    def compress(self, data: bytes) -> bytes:
        """Compress bytes using zlib."""
        return zlib.compress(data)



[docs]
    def decompress(self, data: bytes) -> bytes:
        """Decompress bytes using zlib."""
        return zlib.decompress(data)