Use protocols for gzip and _compression (#5170)

Closes: #5099

Author: srittau

stdlib/_compression.pyi

@@ -1,15 +1,20 @@
 from _typeshed import WriteableBuffer
 from io import BufferedIOBase, RawIOBase
-from typing import Any, Callable, Tuple, Type, Union
+from typing import Any, Callable, Protocol, Tuple, Type, Union
 
 BUFFER_SIZE: Any
 
+class _Reader(Protocol):
+    def read(self, __n: int) -> bytes: ...
+    def seekable(self) -> bool: ...
+    def seek(self, __n: int) -> Any: ...
+
 class BaseStream(BufferedIOBase): ...
 
 class DecompressReader(RawIOBase):
     def __init__(
         self,
-        fp: RawIOBase,
+        fp: _Reader,
         decomp_factory: Callable[..., object],
         trailing_error: Union[Type[Exception], Tuple[Type[Exception], ...]] = ...,
         **decomp_args: Any,

stdlib/gzip.pyi

@@ -2,15 +2,46 @@ import _compression
 import sys
 import zlib
 from _typeshed import AnyPath, ReadableBuffer
-from typing import IO, Optional, TextIO, Union, overload
+from io import FileIO
+from typing import Any, Optional, Protocol, TextIO, Union, overload
 from typing_extensions import Literal
 
-_OpenBinaryMode = Literal["r", "rb", "a", "ab", "w", "wb", "x", "xb"]
+_ReadBinaryMode = Literal["r", "rb"]
+_WriteBinaryMode = Literal["a", "ab", "w", "wb", "x", "xb"]
 _OpenTextMode = Literal["rt", "at", "wt", "xt"]
+
+READ: Literal[1]
+WRITE: Literal[2]
+
+class _ReadableFileobj(Protocol):
+    def read(self, __n: int) -> bytes: ...
+    def seek(self, __n: int) -> Any: ...
+    # The following attributes and methods are optional:
+    # name: str
+    # mode: str
+    # def fileno() -> int: ...
+
+class _WritableFileobj(Protocol):
+    def write(self, __b: bytes) -> Any: ...
+    def flush(self) -> Any: ...
+    # The following attributes and methods are optional:
+    # name: str
+    # mode: str
+    # def fileno() -> int: ...
+
+@overload
+def open(
+    filename: Union[AnyPath, _ReadableFileobj],
+    mode: _ReadBinaryMode = ...,
+    compresslevel: int = ...,
+    encoding: None = ...,
+    errors: None = ...,
+    newline: None = ...,
+) -> GzipFile: ...
 @overload
 def open(
-    filename: Union[AnyPath, IO[bytes]],
-    mode: _OpenBinaryMode = ...,
+    filename: Union[AnyPath, _WritableFileobj],
+    mode: _WriteBinaryMode,
     compresslevel: int = ...,
     encoding: None = ...,
     errors: None = ...,
@@ -27,7 +58,7 @@ def open(
 ) -> TextIO: ...
 @overload
 def open(
-    filename: Union[AnyPath, IO[bytes]],
+    filename: Union[AnyPath, _ReadableFileobj, _WritableFileobj],
     mode: str,
     compresslevel: int = ...,
     encoding: Optional[str] = ...,
@@ -36,8 +67,8 @@ def open(
 ) -> Union[GzipFile, TextIO]: ...
 
 class _PaddedFile:
-    file: IO[bytes]
-    def __init__(self, f: IO[bytes], prepend: bytes = ...) -> None: ...
+    file: _ReadableFileobj
+    def __init__(self, f: _ReadableFileobj, prepend: bytes = ...) -> None: ...
     def read(self, size: int) -> bytes: ...
     def prepend(self, prepend: bytes = ...) -> None: ...
     def seek(self, off: int) -> int: ...
@@ -47,17 +78,54 @@ if sys.version_info >= (3, 8):
     class BadGzipFile(OSError): ...
 
 class GzipFile(_compression.BaseStream):
-    myfileobj: Optional[IO[bytes]]
-    mode: str
+    myfileobj: Optional[FileIO]
+    mode: Literal[1, 2]
     name: str
     compress: zlib._Compress
-    fileobj: IO[bytes]
+    fileobj: Union[_ReadableFileobj, _WritableFileobj]
+    @overload
+    def __init__(
+        self,
+        filename: Optional[AnyPath],
+        mode: _ReadBinaryMode,
+        compresslevel: int = ...,
+        fileobj: Optional[_ReadableFileobj] = ...,
+        mtime: Optional[float] = ...,
+    ) -> None: ...
+    @overload
+    def __init__(
+        self,
+        *,
+        mode: _ReadBinaryMode,
+        compresslevel: int = ...,
+        fileobj: Optional[_ReadableFileobj] = ...,
+        mtime: Optional[float] = ...,
+    ) -> None: ...
+    @overload
+    def __init__(
+        self,
+        filename: Optional[AnyPath],
+        mode: _WriteBinaryMode,
+        compresslevel: int = ...,
+        fileobj: Optional[_WritableFileobj] = ...,
+        mtime: Optional[float] = ...,
+    ) -> None: ...
+    @overload
+    def __init__(
+        self,
+        *,
+        mode: _WriteBinaryMode,
+        compresslevel: int = ...,
+        fileobj: Optional[_WritableFileobj] = ...,
+        mtime: Optional[float] = ...,
+    ) -> None: ...
+    @overload
     def __init__(
         self,
         filename: Optional[AnyPath] = ...,
         mode: Optional[str] = ...,
         compresslevel: int = ...,
-        fileobj: Optional[IO[bytes]] = ...,
+        fileobj: Union[_ReadableFileobj, _WritableFileobj, None] = ...,
         mtime: Optional[float] = ...,
     ) -> None: ...
     @property
@@ -82,7 +150,7 @@ class GzipFile(_compression.BaseStream):
     def readline(self, size: Optional[int] = ...) -> bytes: ...
 
 class _GzipReader(_compression.DecompressReader):
-    def __init__(self, fp: IO[bytes]) -> None: ...
+    def __init__(self, fp: _ReadableFileobj) -> None: ...
     def read(self, size: int = ...) -> bytes: ...
 
 if sys.version_info >= (3, 8):