from __future__ import annotations
from typing import TYPE_CHECKING, ClassVar, Protocol
if TYPE_CHECKING:
from io import BytesIO
from litestar.config.compression import CompressionConfig
from litestar.enums import CompressionEncoding
[docs]
class CompressionFacade(Protocol):
"""A unified facade offering a uniform interface for different compression libraries."""
__slots__ = ()
encoding: ClassVar[str]
"""The encoding of the compression."""
[docs]
def __init__(
self, buffer: BytesIO, compression_encoding: CompressionEncoding | str, config: CompressionConfig
) -> None:
"""Initialize ``CompressionFacade``.
Args:
buffer: A bytes IO buffer to write the compressed data into.
compression_encoding: The compression encoding used.
config: The app compression config.
"""
...
[docs]
def write(self, body: bytes | bytearray, final: bool = False) -> None:
"""Write compressed bytes to the buffer.
Args:
body: The message body to process. Can be `bytes` or `bytearray`.
final: Indicates whether this is the last chunk of data. If True,
the compressor may flush any remaining internal buffers.
Returns:
None
"""
...
[docs]
def close(self) -> None:
"""Close the compression stream.
Returns:
None
"""
...
Litestar