feat: Add type hints to stream module

falcon/stream.py

@@ -15,10 +15,19 @@
 """WSGI BoundedStream class."""
 
 import io
+import typing
 
 __all__ = ['BoundedStream']
 
 
+Result = typing.TypeVar(
+    'Result',
+    bound=typing.Union[bytes, typing.List[bytes]]
+)
+
+OptionalInt = typing.Optional[int]
+
+
 class BoundedStream(io.IOBase):
     """Wrap *wsgi.input* streams to make them more robust.
 
@@ -45,21 +54,23 @@ class BoundedStream(io.IOBase):
 
     """
 
-    def __init__(self, stream, stream_len):
+    def __init__(self, stream: typing.BinaryIO, stream_len: int) -> None:
         self.stream = stream
         self.stream_len = stream_len
 
         self._bytes_remaining = self.stream_len
 
-    def __iter__(self):
+    def __iter__(self) -> typing.Self:
         return self
 
-    def __next__(self):
+    def __next__(self) -> bytes:
         return next(self.stream)
 
     next = __next__
 
-    def _read(self, size, target):
+    def _read(
+        self, size: OptionalInt, target: typing.Callable[[int], Result]
+    ) -> Result:
         """Proxy reads to the underlying stream.
 
         Args:
@@ -85,19 +96,19 @@ def _read(self, size, target):
         self._bytes_remaining -= size
         return target(size)
 
-    def readable(self):
+    def readable(self) -> bool:
         """Return ``True`` always."""
         return True
 
-    def seekable(self):
+    def seekable(self) -> bool:
         """Return ``False`` always."""
         return False
 
-    def writable(self):
+    def writable(self) -> bool:
         """Return ``False`` always."""
         return False
 
-    def read(self, size=None):
+    def read(self, size: OptionalInt = None) -> bytes:
         """Read from the stream.
 
         Args:
@@ -111,7 +122,7 @@ def read(self, size=None):
 
         return self._read(size, self.stream.read)
 
-    def readline(self, limit=None):
+    def readline(self, limit: OptionalInt = None) -> bytes:
         """Read a line from the stream.
 
         Args:
@@ -125,7 +136,7 @@ def readline(self, limit=None):
 
         return self._read(limit, self.stream.readline)
 
-    def readlines(self, hint=None):
+    def readlines(self, hint: OptionalInt = None) -> typing.List[bytes]:
         """Read lines from the stream.
 
         Args:
@@ -139,12 +150,12 @@ def readlines(self, hint=None):
 
         return self._read(hint, self.stream.readlines)
 
-    def write(self, data):
+    def write(self, data: bytes) -> None:
         """Raise IOError always; writing is not supported."""
 
         raise IOError('Stream is not writeable')
 
-    def exhaust(self, chunk_size=64 * 1024):
+    def exhaust(self, chunk_size: int = 64 * 1024) -> None:
         """Exhaust the stream.
 
         This consumes all the data left until the limit is reached.
@@ -159,7 +170,7 @@ def exhaust(self, chunk_size=64 * 1024):
                 break
 
     @property
-    def eof(self):
+    def eof(self) -> bool:
         return self._bytes_remaining <= 0
 
     is_exhausted = eof

pyproject.toml

@@ -8,7 +8,6 @@
 
 [tool.mypy]
     exclude = "falcon/bench/|falcon/cmd/"
-
     [[tool.mypy.overrides]]
         module = [
             "cbor2",
@@ -34,6 +33,12 @@
         ]
         ignore_missing_imports = true
 
+    [[tool.mypy.overrides]]
+        module = [
+            "falcon.stream"
+        ]
+        disallow_untyped_defs = true
+
 [tool.towncrier]
     package = "falcon"
     package_dir = ""