Skip to content

HTTP

Request markers

CustomHeader dataclass

Mark a parameter as a header value. Argument is passed «as is» during a service call.

Examples:

>>> class Service(Protocol):
>>>     def service(self, accept_language: Annotated[str, CustomHeader("Accept-Language")]):
>>>         ...
Source code in combadge/support/http/markers/request.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@dataclass(**SLOTS)
class CustomHeader(ParameterMarker[ContainsHeaders]):
    """
    Mark a parameter as a header value. Argument is passed «as is» during a service call.

    Examples:
        >>> class Service(Protocol):
        >>>     def service(self, accept_language: Annotated[str, CustomHeader("Accept-Language")]):
        >>>         ...
    """

    name: str

    @override
    def __call__(self, request: ContainsHeaders, value: Any) -> None:  # noqa: D102
        request.headers.append((self.name, value))

QueryParam dataclass

Mark parameter as a query parameter.

Examples:

>>> def call(query: Annotated[str, QueryParam("query")]) -> ...:
>>>     ...
Source code in combadge/support/http/markers/request.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@dataclass(**SLOTS)
class QueryParam(ParameterMarker[ContainsQueryParams]):
    """
    Mark parameter as a query parameter.

    Examples:
        >>> def call(query: Annotated[str, QueryParam("query")]) -> ...:
        >>>     ...
    """

    name: str

    @override
    def __call__(self, request: ContainsQueryParams, value: Any) -> None:  # noqa: D102
        request.query_params.append((self.name, value.value if isinstance(value, Enum) else value))

QueryArrayParam dataclass

Mark parameter as an array-like query parameter.

Supports any iterable value as a call argument.

Examples:

>>> def call(query: Annotated[list[str], QueryParam("query")]) -> ...:
>>>     ...

>>> def call(query: Annotated[Iterable[str], QueryParam("query")]) -> ...:
>>>     ...
Source code in combadge/support/http/markers/request.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@dataclass(**SLOTS)
class QueryArrayParam(ParameterMarker[ContainsQueryParams]):
    """
    Mark parameter as an array-like query parameter.

    Supports any iterable value as a call argument.

    Examples:
        >>> def call(query: Annotated[list[str], QueryParam("query")]) -> ...:
        >>>     ...

        >>> def call(query: Annotated[Iterable[str], QueryParam("query")]) -> ...:
        >>>     ...
    """

    name: str

    @override
    def __call__(self, request: ContainsQueryParams, value: Any) -> None:  # noqa: D102
        for sub_value in value:
            request.query_params.append((self.name, sub_value.value if isinstance(sub_value, Enum) else sub_value))

Payload dataclass

Mark parameter as a request payload. An argument gets converted to a dictionary and passed over to a backend.

Examples:

Simple usage:

>>> def call(body: Payload[BodyModel]) -> ...:
>>>     ...

Equivalent expanded usage:

>>> def call(body: Annotated[BodyModel, Payload()]) -> ...:
>>>     ...
Source code in combadge/support/http/markers/request.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
@dataclass(**SLOTS)
class Payload(ParameterMarker[ContainsPayload]):
    """
    Mark parameter as a request payload. An argument gets converted to a dictionary and passed over to a backend.

    Examples:
        Simple usage:

        >>> def call(body: Payload[BodyModel]) -> ...:
        >>>     ...

        Equivalent expanded usage:

        >>> def call(body: Annotated[BodyModel, Payload()]) -> ...:
        >>>     ...
    """

    exclude_unset: bool = False
    by_alias: bool = False

    @override
    def __call__(self, request: ContainsPayload, value: Any) -> None:  # noqa: D102
        value = get_type_adapter(type(value)).dump_python(
            value,
            by_alias=self.by_alias,
            exclude_unset=self.exclude_unset,
        )
        if request.payload is None:
            request.payload = value
        elif isinstance(request.payload, dict):
            request.payload.update(value)  # merge into the existing payload
        else:
            raise ValueError(f"attempting to merge {type(value)} into {type(request.payload)}")

    def __class_getitem__(cls, item: type[Any]) -> Any:
        return Annotated[item, cls()]

Field dataclass

Mark a parameter as a value of a separate payload field.

Examples:

>>> def call(param: Annotated[int, Field("param")]) -> ...:
>>>     ...
Notes
  • Enum values are passed by value
Source code in combadge/support/http/markers/request.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
@dataclass(**SLOTS)
class Field(ParameterMarker[ContainsPayload]):
    """
    Mark a parameter as a value of a separate payload field.

    Examples:
        >>> def call(param: Annotated[int, Field("param")]) -> ...:
        >>>     ...

    Notes:
        - Enum values are passed by value
    """

    name: str

    @override
    def __call__(self, request: ContainsPayload, value: Any) -> None:  # noqa: D102
        if request.payload is None:
            request.payload = {}
        request.payload[self.name] = value.value if isinstance(value, Enum) else value

FormData dataclass

Mark parameter as a request form data.

An argument gets converted to a dictionary and passed over to a backend.

Examples:

>>> def call(body: FormData[FormModel]) -> ...:
>>>     ...

>>> def call(body: Annotated[FormModel, FormData()]) -> ...:
>>>     ...
Source code in combadge/support/http/markers/request.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
@dataclass(**SLOTS)
class FormData(ParameterMarker[ContainsFormData]):
    """
    Mark parameter as a request form data.

    An argument gets converted to a dictionary and passed over to a backend.

    Examples:
        >>> def call(body: FormData[FormModel]) -> ...:
        >>>     ...

        >>> def call(body: Annotated[FormModel, FormData()]) -> ...:
        >>>     ...
    """

    @override
    def __call__(self, request: ContainsFormData, value: Any) -> None:  # noqa: D102
        value = get_type_adapter(type(value)).dump_python(value, by_alias=True)
        if not isinstance(value, dict):
            raise TypeError(f"form data requires a dictionary, got {type(value)}")
        for item_name, item_value in value.items():
            request.append_form_field(item_name, item_value)

    def __class_getitem__(cls, item: type[Any]) -> Any:
        return Annotated[item, FormData()]

FormField dataclass

Mark a parameter as a separate form field value.

Examples:

>>> def call(param: Annotated[int, FormField("param")]) -> ...:
>>>     ...
Notes
  • Multiple arguments with the same field name are allowed
  • FormData marker's fields get merged with FormField ones (if present)
  • Enum values are passed by value
Source code in combadge/support/http/markers/request.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
@dataclass(**SLOTS)
class FormField(ParameterMarker[ContainsFormData]):
    """
    Mark a parameter as a separate form field value.

    Examples:
        >>> def call(param: Annotated[int, FormField("param")]) -> ...:
        >>>     ...

    Notes:
        - Multiple arguments with the same field name are allowed
        - [`FormData`][combadge.support.http.markers.FormData] marker's fields get merged with `FormField` ones (if present)
        - Enum values are passed by value
    """  # noqa: E501

    name: str

    @override
    def __call__(self, request: ContainsFormData, value: Any) -> None:  # noqa: D102
        request.append_form_field(self.name, value.value if isinstance(value, Enum) else value)

path 

path(
    path_or_factory: str | Callable[..., str]
) -> Callable[[FunctionT], FunctionT]

Specify a URL path.

Examples:

>>> @path("/hello/world")
>>> def call() -> None: ...

>>> @path("/hello/{name}")
>>> def call(name: str) -> None: ...

>>> @path("/hello/{0}")
>>> def call(name: str) -> None: ...

>>> @path(lambda name, **_: f"/hello/{name}")
>>> def call(name: str) -> None: ...
Source code in combadge/support/http/markers/request.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def path(path_or_factory: str | Callable[..., str]) -> Callable[[FunctionT], FunctionT]:
    """
    Specify a URL path.

    Examples:
        >>> @path("/hello/world")
        >>> def call() -> None: ...

        >>> @path("/hello/{name}")
        >>> def call(name: str) -> None: ...

        >>> @path("/hello/{0}")
        >>> def call(name: str) -> None: ...

        >>> @path(lambda name, **_: f"/hello/{name}")
        >>> def call(name: str) -> None: ...
    """
    return Path[Any](path_or_factory).mark

http_method 

http_method(method: str) -> Callable[[FunctionT], FunctionT]

Specify an HTTP method.

Examples:

>>> @http_method("POST")
>>> def call() -> None: ...
Source code in combadge/support/http/markers/request.py 
 95
 96
 97
 98
 99
100
101
102
103
def http_method(method: str) -> Callable[[FunctionT], FunctionT]:
    """
    Specify an HTTP method.

    Examples:
        >>> @http_method("POST")
        >>> def call() -> None: ...
    """
    return HttpMethod[Any](method).mark

Response markers

StatusCode dataclass

Enrich the payload with response status code.

Examples:

>>> def call(...) -> Annotated[Model, Mixin(StatusCode())]:
>>>     ...
Source code in combadge/support/http/markers/response.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@dataclass(**SLOTS)
class StatusCode(ResponseMarker):
    """
    Enrich the payload with response status code.

    Examples:
        >>> def call(...) -> Annotated[Model, Mixin(StatusCode())]:
        >>>     ...
    """

    key: Any = "status_code"
    """Key under which the status code should mapped in the payload."""

    @override
    def __call__(self, response: SupportsStatusCode, payload: Any) -> dict[Any, Any]:  # noqa: D102
        return {self.key: HTTPStatus(response.status_code)}

key class-attribute instance-attribute 

key: Any = 'status_code'

Key under which the status code should mapped in the payload.

ReasonPhrase dataclass

Enrich the payload with HTTP reason message.

Source code in combadge/support/http/markers/response.py 
33
34
35
36
37
38
39
40
41
42
@dataclass(**SLOTS)
class ReasonPhrase(ResponseMarker):
    """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, payload: Any) -> dict[Any, Any]:  # noqa: D102
        return {self.key: response.reason_phrase}

key class-attribute instance-attribute 

key: Any = 'reason'

Key under which the reason message should mapped in the payload.

Text dataclass

Enrich the payload with HTTP response text.

Examples:

>>> class MyResponse(BaseModel):
>>>     my_text: str
>>>
>>> class MyService(Protocol):
>>>     @http_method("GET")
>>>     @path(...)
>>>     def get_text(self) -> Annotated[MyResponse, Text("my_text")]:
>>>         ...
Source code in combadge/support/http/markers/response.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@dataclass(**SLOTS)
class Text(ResponseMarker):
    """
    Enrich the payload with HTTP response text.

    Examples:
        >>> class MyResponse(BaseModel):
        >>>     my_text: str
        >>>
        >>> class MyService(Protocol):
        >>>     @http_method("GET")
        >>>     @path(...)
        >>>     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, payload: Any) -> dict[Any, Any]:  # noqa: D102
        return {self.key: response.text}

key class-attribute instance-attribute 

key: Any = 'text'

Key under which the text contents should assigned in the payload.

Header dataclass

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"),
>>>     ]:
>>>         ...
Source code in combadge/support/http/markers/response.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@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}

header instance-attribute 

header: str

HTTP header name, case-insensitive.

key instance-attribute 

key: Any

Key under which the header contents should assigned in the payload.