[compression] Add Protocol _Decompressor (#14885)

This commit is contained in:
Semyon Moroz
2025-11-15 01:23:15 +04:00
committed by GitHub
parent 684c0a6007
commit 1dad9a5857
3 changed files with 84 additions and 4 deletions
+13 -2
View File
@@ -1,4 +1,4 @@
from _typeshed import Incomplete, WriteableBuffer
from _typeshed import ReadableBuffer, WriteableBuffer
from collections.abc import Callable
from io import DEFAULT_BUFFER_SIZE, BufferedIOBase, RawIOBase
from typing import Any, Protocol, type_check_only
@@ -11,13 +11,24 @@ class _Reader(Protocol):
def seekable(self) -> bool: ...
def seek(self, n: int, /) -> Any: ...
@type_check_only
class _Decompressor(Protocol):
def decompress(self, data: ReadableBuffer, /, max_length: int = ...) -> bytes: ...
@property
def unused_data(self) -> bytes: ...
@property
def eof(self) -> bool: ...
# `zlib._Decompress` does not have next property, but `DecompressReader` calls it:
# @property
# def needs_input(self) -> bool: ...
class BaseStream(BufferedIOBase): ...
class DecompressReader(RawIOBase):
def __init__(
self,
fp: _Reader,
decomp_factory: Callable[..., Incomplete], # Consider backporting changes to _compression
decomp_factory: Callable[..., _Decompressor], # Consider backporting changes to _compression
trailing_error: type[Exception] | tuple[type[Exception], ...] = (),
**decomp_args: Any, # These are passed to decomp_factory.
) -> None: ...