From b70ef0ad6b0f1c427ff3ca6913ec8e1c1e007c0b Mon Sep 17 00:00:00 2001
From: Hal Blackburn <hwtb2@cam.ac.uk>
Date: Tue, 25 Mar 2025 05:18:00 +0000
Subject: [PATCH] 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 <hwtb2@cam.ac.uk>
---
 cloudevents/conversion.py             |  2 +-
 cloudevents/http/conversion.py        |  2 +-
 cloudevents/pydantic/v1/conversion.py |  2 +-
 cloudevents/pydantic/v2/conversion.py |  2 +-
 cloudevents/sdk/types.py              | 16 ++++++++++++++++
 5 files changed, 20 insertions(+), 4 deletions(-)

diff --git a/cloudevents/conversion.py b/cloudevents/conversion.py
index c73e3ed..e307344 100644
--- a/cloudevents/conversion.py
+++ b/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:
diff --git a/cloudevents/http/conversion.py b/cloudevents/http/conversion.py
index a7da926..c0830ee 100644
--- a/cloudevents/http/conversion.py
+++ b/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:
diff --git a/cloudevents/pydantic/v1/conversion.py b/cloudevents/pydantic/v1/conversion.py
index dcf0b7d..ded202f 100644
--- a/cloudevents/pydantic/v1/conversion.py
+++ b/cloudevents/pydantic/v1/conversion.py
@@ -21,7 +21,7 @@ from cloudevents.sdk import types
 
 
 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:
diff --git a/cloudevents/pydantic/v2/conversion.py b/cloudevents/pydantic/v2/conversion.py
index 6510854..47ce282 100644
--- a/cloudevents/pydantic/v2/conversion.py
+++ b/cloudevents/pydantic/v2/conversion.py
@@ -22,7 +22,7 @@ from cloudevents.sdk import types
 
 
 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:
diff --git a/cloudevents/sdk/types.py b/cloudevents/sdk/types.py
index e6ab46e..6baef6b 100644
--- a/cloudevents/sdk/types.py
+++ b/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