chore: allow non-dict headers types in from_http()

from_http() conversion function was requiring its headers argument to
be a typing.Dict, which makes it incompatible with headers types of http
libraries, which support features like multiple values per key.
typing.Mapping and even _typeshed.SupportsItems do not cover these
types. For example,
samples/http-image-cloudevents/image_sample_server.py was failing to
type check where it calls `from_http(request.headers, ...)`.

To support these kind of headers types in from_http(), we now define our
own SupportsDuplicateItems protocol, which is broader than
_typeshed.SupportsItems.

I've only applied this to from_http(), as typing.Mapping is OK for most
other methods that accept dict-like objects, and using this more lenient
interface everywhere would impose restrictions on our implementation,
even though it might be more flexible for users.

Signed-off-by: Hal Blackburn <[email protected]>

Author: h4l

cloudevents/conversion.py

@@ -91,7 +91,7 @@ def from_json(
 
 def from_http(
     event_type: typing.Type[AnyCloudEvent],
-    headers: typing.Mapping[str, str],
+    headers: types.SupportsDuplicateItems[str, str],
     data: typing.Optional[typing.Union[str, bytes]],
     data_unmarshaller: typing.Optional[types.UnmarshallerType] = None,
 ) -> AnyCloudEvent:

cloudevents/http/conversion.py

@@ -37,7 +37,7 @@ def from_json(
 
 
 def from_http(
-    headers: typing.Dict[str, str],
+    headers: types.SupportsDuplicateItems[str, str],
     data: typing.Optional[typing.Union[str, bytes]],
     data_unmarshaller: typing.Optional[types.UnmarshallerType] = None,
 ) -> CloudEvent:

cloudevents/pydantic/v1/conversion.py

@@ -21,7 +21,7 @@
 
 
 def from_http(
-    headers: typing.Dict[str, str],
+    headers: types.SupportsDuplicateItems[str, str],
     data: typing.Optional[typing.AnyStr],
     data_unmarshaller: typing.Optional[types.UnmarshallerType] = None,
 ) -> CloudEvent:

cloudevents/pydantic/v2/conversion.py

@@ -22,7 +22,7 @@
 
 
 def from_http(
-    headers: typing.Dict[str, str],
+    headers: types.SupportsDuplicateItems[str, str],
     data: typing.Optional[typing.AnyStr],
     data_unmarshaller: typing.Optional[types.UnmarshallerType] = None,
 ) -> CloudEvent:

cloudevents/sdk/types.py

@@ -14,9 +14,25 @@
 
 import typing
 
+_K_co = typing.TypeVar("_K_co", covariant=True)
+_V_co = typing.TypeVar("_V_co", covariant=True)
+
 # Use consistent types for marshal and unmarshal functions across
 # both JSON and Binary format.
 
 MarshallerType = typing.Callable[[typing.Any], typing.AnyStr]
 
 UnmarshallerType = typing.Callable[[typing.AnyStr], typing.Any]
+
+
+class SupportsDuplicateItems(typing.Protocol[_K_co, _V_co]):
+    """
+    Dict-like objects with an items() method that may produce duplicate keys.
+    """
+
+    # This is wider than _typeshed.SupportsItems, which expects items() to
+    # return type an AbstractSet. werkzeug's Headers class satisfies this type,
+    # but not _typeshed.SupportsItems.
+
+    def items(self) -> typing.Iterable[typing.Tuple[_K_co, _V_co]]:
+        pass