NEW: add Header HTTP marker ✨

combadge/core/binder.py

@@ -16,8 +16,7 @@
 if TYPE_CHECKING:
     from combadge.core.interfaces import MethodBinder, ProvidesBinder, ServiceMethod
 
-    def lru_cache(maxsize: int | None) -> Callable[[FunctionT], FunctionT]:
-        ...
+    def lru_cache(maxsize: int | None) -> Callable[[FunctionT], FunctionT]: ...
 
 else:
     from functools import lru_cache

combadge/support/http/abc/interfaces.py

@@ -1,8 +1,18 @@
 """Interfaces for HTTP-related request and response classes."""
 
+from typing import Mapping
+
 from typing_extensions import Protocol
 
 
+class SupportsHeaders(Protocol):
+    """Supports read-only case-insensitive mapping of headers."""
+
+    @property
+    def headers(self) -> Mapping[str, str]:  # noqa: D102
+        raise NotImplementedError
+
+
 class SupportsStatusCode(Protocol):
     """Supports a read-only status code attribute or property."""
 

combadge/support/http/markers/response.py

@@ -9,13 +9,13 @@
 
 from combadge._helpers.dataclasses import SLOTS
 from combadge.core.markers.response import ResponseMarker
-from combadge.support.http.abc import SupportsReasonPhrase, SupportsStatusCode, SupportsText
+from combadge.support.http.abc import SupportsHeaders, SupportsReasonPhrase, SupportsStatusCode, SupportsText
 
 
 @dataclass(**SLOTS)
 class StatusCode(ResponseMarker):
     """
-    Build a payload with response status code.
+    Enrich the payload with response status code.
 
     Examples:
         >>> def call(...) -> Annotated[Model, Mixin(StatusCode())]:
@@ -26,26 +26,26 @@ class StatusCode(ResponseMarker):
     """Key under which the status code should mapped in the payload."""
 
     @override
-    def __call__(self, response: SupportsStatusCode, input_: Any) -> Dict[Any, Any]:  # noqa: D102
+    def __call__(self, response: SupportsStatusCode, payload: Any) -> Dict[Any, Any]:  # noqa: D102
         return {self.key: HTTPStatus(response.status_code)}
 
 
 @dataclass(**SLOTS)
 class ReasonPhrase(ResponseMarker):
-    """Build a payload with HTTP reason message."""
+    """Enrich the payload with HTTP reason message."""
 
     key: Any = "reason"
     """Key under which the reason message should mapped in the payload."""
 
     @override
-    def __call__(self, response: SupportsReasonPhrase, input_: Any) -> Dict[Any, Any]:  # noqa: D102
+    def __call__(self, response: SupportsReasonPhrase, payload: Any) -> Dict[Any, Any]:  # noqa: D102
         return {self.key: response.reason_phrase}
 
 
 @dataclass(**SLOTS)
 class Text(ResponseMarker):
     """
-    Build a payload with HTTP response text.
+    Enrich the payload with HTTP response text.
 
     Examples:
         >>> class MyResponse(BaseModel):
@@ -54,16 +54,55 @@ class Text(ResponseMarker):
         >>> class MyService(Protocol):
         >>>     @http_method("GET")
         >>>     @path(...)
-        >>>     def get_text(self) -> Annotated[MyResponse, Text("my_text"), Extract("my_text")]:
+        >>>     def get_text(self) -> Annotated[MyResponse, Text("my_text")]:
         >>>         ...
     """
 
     key: Any = "text"
     """Key under which the text contents should assigned in the payload."""
 
     @override
-    def __call__(self, response: SupportsText, input_: Any) -> Dict[Any, Any]:  # noqa: D102
+    def __call__(self, response: SupportsText, payload: Any) -> Dict[Any, Any]:  # noqa: D102
         return {self.key: response.text}
 
 
-__all__ = ("StatusCode", "ReasonPhrase", "Text")
+@dataclass(**SLOTS)
+class Header(ResponseMarker):
+    """
+    Enrich the payload with the specified HTTP header's value.
+
+    If the header be missing, the payload will not be enriched.
+
+    Examples:
+        >>> class MyResponse(BaseModel):
+        >>>     content_length: int
+        >>>     optional: str = "default"
+        >>>
+        >>> class MyService(Protocol):
+        >>>     @http_method("GET")
+        >>>     @path(...)
+        >>>     def get_something(self) -> Annotated[
+        >>>         MyResponse,
+        >>>         Header(header="content-length", key="content_length"),
+        >>>         Header(header="x-optional", key="optional"),
+        >>>     ]:
+        >>>         ...
+    """
+
+    header: str
+    """HTTP header name, case-insensitive."""
+
+    key: Any
+    """Key under which the header contents should assigned in the payload."""
+
+    @override
+    def __call__(self, response: SupportsHeaders, payload: Any) -> Dict[Any, Any]:  # noqa: D102
+        try:
+            value = response.headers[self.header]
+        except KeyError:
+            return {}
+        else:
+            return {self.key: value}
+
+
+__all__ = ("StatusCode", "ReasonPhrase", "Text", "Header")

tests/core/test_binder.py

@@ -65,17 +65,15 @@ def get_expected() -> Tuple[Any, ...]:
 
 
 def test_protocol_class_var() -> None:
-    class ServiceProtocol(Protocol):
-        ...
+    class ServiceProtocol(Protocol): ...
 
     service = bind(ServiceProtocol, Mock())  # type: ignore[type-abstract]
     assert isinstance(service, BaseBoundService)
     assert service.__combadge_protocol__ is ServiceProtocol
 
 
 def test_service_type() -> None:
-    class ServiceProtocol(SupportsService):
-        ...
+    class ServiceProtocol(SupportsService): ...
 
     service = ServiceProtocol.bind(Mock())
     assert_type(service, ServiceProtocol)

tests/integration/test_httpbin.py

@@ -8,7 +8,8 @@
 
 from combadge.core.errors import BackendError
 from combadge.core.interfaces import SupportsService
-from combadge.support.http.markers import CustomHeader, FormData, FormField, QueryParam, http_method, path
+from combadge.core.markers import Mixin
+from combadge.support.http.markers import CustomHeader, FormData, FormField, Header, QueryParam, http_method, path
 from combadge.support.httpx.backends.async_ import HttpxBackend as AsyncHttpxBackend
 from combadge.support.httpx.backends.sync import HttpxBackend as SyncHttpxBackend
 
@@ -30,8 +31,7 @@ def post_anything(
             data: FormData[Data],
             bar: Annotated[int, FormField("barqux")],
             qux: Annotated[int, FormField("barqux")],
-        ) -> Response:
-            ...
+        ) -> Response: ...
 
     service = SupportsHttpbin.bind(SyncHttpxBackend(Client(base_url="https://httpbin.org")))
     response = service.post_anything(data=Data(foo=42), bar=100500, qux=100501)
@@ -52,20 +52,22 @@ def get_anything(
             self,
             foo: Annotated[int, QueryParam("foobar")],
             bar: Annotated[int, QueryParam("foobar")],
-        ) -> Response:
-            ...
+        ) -> Response: ...
 
     service = SupportsHttpbin.bind(SyncHttpxBackend(Client(base_url="https://httpbin.org")))
     response = service.get_anything(foo=100500, bar=100501)
 
     assert response == Response(args={"foobar": ["100500", "100501"]})
 
 
+class _HeadersResponse(BaseModel):
+    headers: Dict[str, Any]
+    content_length: int
+    missing_header: int = 42
+
+
 @pytest.mark.vcr()
 def test_headers_sync() -> None:
-    class Response(BaseModel):
-        headers: Dict[str, Any]
-
     class SupportsHttpbin(SupportsService, Protocol):
         @http_method("GET")
         @path("/headers")
@@ -75,20 +77,22 @@ def get_headers(
             foo: Annotated[str, CustomHeader("x-foo")],
             bar: Annotated[str, CustomHeader("x-bar")] = "barval",
             baz: Annotated[Union[str, Callable[[], str]], CustomHeader("x-baz")] = lambda: "bazval",
-        ) -> Response:
-            ...
+        ) -> Annotated[_HeadersResponse, Mixin(Header("content-length", "content_length"))]: ...
 
     service = SupportsHttpbin.bind(SyncHttpxBackend(Client(base_url="https://httpbin.org")))
     response = service.get_headers(foo="fooval")
     assert response.headers["X-Foo"] == "fooval"
     assert response.headers["X-Bar"] == "barval"
     assert response.headers["X-Baz"] == "bazval"
+    assert response.content_length == 363
+    assert response.missing_header == 42
 
 
 @pytest.mark.vcr()
 async def test_headers_async() -> None:
     class Response(BaseModel):
         headers: Dict[str, Any]
+        content_length: int
 
     class SupportsHttpbin(SupportsService, Protocol):
         @http_method("GET")
@@ -99,14 +103,15 @@ async def get_headers(
             foo: Annotated[str, CustomHeader("x-foo")],
             bar: Annotated[str, CustomHeader("x-bar")] = "barval",
             baz: Annotated[Union[str, Callable[[], str]], CustomHeader("x-baz")] = lambda: "bazval",
-        ) -> Response:
-            ...
+        ) -> Annotated[_HeadersResponse, Mixin(Header("content-length", "content_length"))]: ...
 
     service = SupportsHttpbin.bind(AsyncHttpxBackend(AsyncClient(base_url="https://httpbin.org")))
     response = await service.get_headers(foo="fooval")
     assert response.headers["X-Foo"] == "fooval"
     assert response.headers["X-Bar"] == "barval"
     assert response.headers["X-Baz"] == "bazval"
+    assert response.content_length == 363
+    assert response.missing_header == 42
 
 
 @pytest.mark.vcr()
@@ -115,8 +120,7 @@ class SupportsHttpbin(SupportsService, Protocol):
         @http_method("GET")
         @path("/get")
         @abstractmethod
-        def get_non_dict(self) -> List[int]:
-            ...
+        def get_non_dict(self) -> List[int]: ...
 
     # Since httpbin.org is not capable of returning a non-dict JSON,
     # I manually patched the recorded VCR.py response.
@@ -132,8 +136,7 @@ class SupportsHttpbin(SupportsService, Protocol):
         @http_method("GET")
         @path("/status/500")
         @abstractmethod
-        def get_internal_server_error(self) -> None:
-            ...
+        def get_internal_server_error(self) -> None: ...
 
     service = SyncHttpxBackend(Client(base_url="https://httpbin.org"))[SupportsHttpbin]  # type: ignore[type-abstract]
     with pytest.raises(BackendError):

tests/support/http/test_markers.py

@@ -6,7 +6,7 @@
 from httpx import Response
 
 from combadge.support.http.abc import ContainsUrlPath
-from combadge.support.http.markers import Path, ReasonPhrase, StatusCode, Text
+from combadge.support.http.markers import Header, Path, ReasonPhrase, StatusCode, Text
 
 
 @pytest.mark.parametrize(
@@ -44,6 +44,14 @@ def test_text() -> None:
     assert Text("key")(Response(status_code=200, text="my text"), ...) == {"key": "my text"}
 
 
+def test_present_header() -> None:
+    assert Header("x-foo", "key")(Response(200, headers={"X-Foo": "42"}), ...) == {"key": "42"}
+
+
+def test_missing_header() -> None:
+    assert Header("x-foo", "key")(Response(200, headers={}), ...) == {}
+
+
 def _example(positional: str, *, keyword: str) -> None:
     pass
 

tests/support/zeep/backends/test_base.py

@@ -9,12 +9,10 @@
 from combadge.support.zeep.backends.base import BaseZeepBackend
 
 
-class _TestFault1(BaseSoapFault):
-    ...
+class _TestFault1(BaseSoapFault): ...
 
 
-class _TestFault2(BaseSoapFault):
-    ...
+class _TestFault2(BaseSoapFault): ...
 
 
 @pytest.mark.parametrize(