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>
This commit is contained in:
Hal Blackburn 2025-03-25 05:18:00 +00:00
parent f81c902843
commit b70ef0ad6b
No known key found for this signature in database
5 changed files with 20 additions and 4 deletions

View File

@ -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:

View File

@ -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:

View File

@ -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:

View File

@ -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:

View File

@ -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