Add TypedDict callbacks (#98)

Author: Kludex

.gitignore

@@ -28,7 +28,7 @@ lib64
 pip-log.txt
 
 # Unit test / coverage reports
-.coverage.*
+.coverage*
 .tox
 nosetests.xml
 

multipart/multipart.py

@@ -9,11 +9,38 @@
 from enum import IntEnum
 from io import BytesIO
 from numbers import Number
-from typing import Dict, Tuple, Union
+from typing import TYPE_CHECKING
 
 from .decoders import Base64Decoder, QuotedPrintableDecoder
 from .exceptions import FileError, FormParserError, MultipartParseError, QuerystringParseError
 
+if TYPE_CHECKING:  # pragma: no cover
+    from typing import Callable, TypedDict
+
+    class QuerystringCallbacks(TypedDict, total=False):
+        on_field_start: Callable[[], None]
+        on_field_name: Callable[[bytes, int, int], None]
+        on_field_data: Callable[[bytes, int, int], None]
+        on_field_end: Callable[[], None]
+        on_end: Callable[[], None]
+
+    class OctetStreamCallbacks(TypedDict, total=False):
+        on_start: Callable[[], None]
+        on_data: Callable[[bytes, int, int], None]
+        on_end: Callable[[], None]
+
+    class MultipartCallbacks(TypedDict, total=False):
+        on_part_begin: Callable[[], None]
+        on_part_data: Callable[[bytes, int, int], None]
+        on_part_end: Callable[[], None]
+        on_headers_begin: Callable[[], None]
+        on_header_field: Callable[[bytes, int, int], None]
+        on_header_value: Callable[[bytes, int, int], None]
+        on_header_end: Callable[[], None]
+        on_headers_finished: Callable[[], None]
+        on_end: Callable[[], None]
+
+
 # Unique missing object.
 _missing = object()
 
@@ -86,7 +113,7 @@ def join_bytes(b):
     return bytes(list(b))
 
 
-def parse_options_header(value: Union[str, bytes]) -> Tuple[bytes, Dict[bytes, bytes]]:
+def parse_options_header(value: str | bytes) -> tuple[bytes, dict[bytes, bytes]]:
     """
     Parses a Content-Type header into a value in the following format:
         (content_type, {parameters})
@@ -148,15 +175,15 @@ class Field:
     :param name: the name of the form field
     """
 
-    def __init__(self, name):
+    def __init__(self, name: str):
         self._name = name
-        self._value = []
+        self._value: list[bytes] = []
 
         # We cache the joined version of _value for speed.
         self._cache = _missing
 
     @classmethod
-    def from_value(klass, name, value):
+    def from_value(cls, name: str, value: bytes | None) -> Field:
         """Create an instance of a :class:`Field`, and set the corresponding
         value - either None or an actual value.  This method will also
         finalize the Field itself.
@@ -166,22 +193,22 @@ def from_value(klass, name, value):
                       None
         """
 
-        f = klass(name)
+        f = cls(name)
         if value is None:
             f.set_none()
         else:
             f.write(value)
         f.finalize()
         return f
 
-    def write(self, data):
+    def write(self, data: bytes) -> int:
         """Write some data into the form field.
 
         :param data: a bytestring
         """
         return self.on_data(data)
 
-    def on_data(self, data):
+    def on_data(self, data: bytes) -> int:
         """This method is a callback that will be called whenever data is
         written to the Field.
 
@@ -191,24 +218,24 @@ def on_data(self, data):
         self._cache = _missing
         return len(data)
 
-    def on_end(self):
+    def on_end(self) -> None:
         """This method is called whenever the Field is finalized."""
         if self._cache is _missing:
             self._cache = b"".join(self._value)
 
-    def finalize(self):
+    def finalize(self) -> None:
         """Finalize the form field."""
         self.on_end()
 
-    def close(self):
+    def close(self) -> None:
         """Close the Field object.  This will free any underlying cache."""
         # Free our value array.
         if self._cache is _missing:
             self._cache = b"".join(self._value)
 
         del self._value
 
-    def set_none(self):
+    def set_none(self) -> None:
         """Some fields in a querystring can possibly have a value of None - for
         example, the string "foo&bar=&baz=asdf" will have a field with the
         name "foo" and value None, one with name "bar" and value "", and one
@@ -218,7 +245,7 @@ def set_none(self):
         self._cache = None
 
     @property
-    def field_name(self):
+    def field_name(self) -> str:
         """This property returns the name of the field."""
         return self._name
 
@@ -230,13 +257,13 @@ def value(self):
 
         return self._cache
 
-    def __eq__(self, other):
+    def __eq__(self, other: object) -> bool:
         if isinstance(other, Field):
             return self.field_name == other.field_name and self.value == other.value
         else:
             return NotImplemented
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         if len(self.value) > 97:
             # We get the repr, and then insert three dots before the final
             # quote.
@@ -553,7 +580,7 @@ class BaseParser:
     def __init__(self):
         self.logger = logging.getLogger(__name__)
 
-    def callback(self, name, data=None, start=None, end=None):
+    def callback(self, name: str, data=None, start=None, end=None):
         """This function calls a provided callback with some data.  If the
         callback is not set, will do nothing.
 
@@ -584,7 +611,7 @@ def callback(self, name, data=None, start=None, end=None):
             self.logger.debug("Calling %s with no data", name)
             func()
 
-    def set_callback(self, name, new_func):
+    def set_callback(self, name: str, new_func):
         """Update the function for a callback.  Removes from the callbacks dict
         if new_func is None.
 
@@ -637,7 +664,7 @@ class OctetStreamParser(BaseParser):
                      i.e. unbounded.
     """
 
-    def __init__(self, callbacks={}, max_size=float("inf")):
+    def __init__(self, callbacks: OctetStreamCallbacks = {}, max_size=float("inf")):
         super().__init__()
         self.callbacks = callbacks
         self._started = False
@@ -647,7 +674,7 @@ def __init__(self, callbacks={}, max_size=float("inf")):
         self.max_size = max_size
         self._current_size = 0
 
-    def write(self, data):
+    def write(self, data: bytes):
         """Write some data to the parser, which will perform size verification,
         and then pass the data to the underlying callback.
 
@@ -732,7 +759,9 @@ class QuerystringParser(BaseParser):
                      i.e. unbounded.
     """
 
-    def __init__(self, callbacks={}, strict_parsing=False, max_size=float("inf")):
+    state: QuerystringState
+
+    def __init__(self, callbacks: QuerystringCallbacks = {}, strict_parsing=False, max_size=float("inf")):
         super().__init__()
         self.state = QuerystringState.BEFORE_FIELD
         self._found_sep = False
@@ -748,7 +777,7 @@ def __init__(self, callbacks={}, strict_parsing=False, max_size=float("inf")):
         # Should parsing be strict?
         self.strict_parsing = strict_parsing
 
-    def write(self, data):
+    def write(self, data: bytes):
         """Write some data to the parser, which will perform size verification,
         parse into either a field name or value, and then pass the
         corresponding data to the underlying callback.  If an error is
@@ -780,7 +809,7 @@ def write(self, data):
 
         return l
 
-    def _internal_write(self, data, length):
+    def _internal_write(self, data: bytes, length: int):
         state = self.state
         strict_parsing = self.strict_parsing
         found_sep = self._found_sep
@@ -989,7 +1018,7 @@ class MultipartParser(BaseParser):
                      i.e. unbounded.
     """
 
-    def __init__(self, boundary, callbacks={}, max_size=float("inf")):
+    def __init__(self, boundary, callbacks: MultipartCallbacks = {}, max_size=float("inf")):
         # Initialize parser state.
         super().__init__()
         self.state = MultipartState.START

tests/test_multipart.py

@@ -333,9 +333,9 @@ def on_field_end():
             del name_buffer[:]
             del data_buffer[:]
 
-        callbacks = {"on_field_name": on_field_name, "on_field_data": on_field_data, "on_field_end": on_field_end}
-
-        self.p = QuerystringParser(callbacks)
+        self.p = QuerystringParser(
+            callbacks={"on_field_name": on_field_name, "on_field_data": on_field_data, "on_field_end": on_field_end}
+        )
 
     def test_simple_querystring(self):
         self.p.write(b"foo=bar")
@@ -464,18 +464,16 @@ def setUp(self):
         self.started = 0
         self.finished = 0
 
-        def on_start():
+        def on_start() -> None:
             self.started += 1
 
-        def on_data(data, start, end):
+        def on_data(data: bytes, start: int, end: int) -> None:
             self.d.append(data[start:end])
 
-        def on_end():
+        def on_end() -> None:
             self.finished += 1
 
-        callbacks = {"on_start": on_start, "on_data": on_data, "on_end": on_end}
-
-        self.p = OctetStreamParser(callbacks)
+        self.p = OctetStreamParser(callbacks={"on_start": on_start, "on_data": on_data, "on_end": on_end})
 
     def assert_data(self, data, finalize=True):
         self.assertEqual(b"".join(self.d), data)