Skip to content

smithy-http

Field

Bases: Field

A name-value pair representing a single field in an HTTP Request or Response.

The kind will dictate metadata placement within an HTTP message.

All field names are case insensitive and case-variance must be treated as equivalent. Names may be normalized but should be preserved for accuracy during transmission.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class Field(interfaces.Field):
    """A name-value pair representing a single field in an HTTP Request or Response.

    The kind will dictate metadata placement within an HTTP message.

    All field names are case insensitive and case-variance must be treated as
    equivalent. Names may be normalized but should be preserved for accuracy during
    transmission.
    """

    def __init__(
        self,
        *,
        name: str,
        values: Iterable[str] | None = None,
        kind: FieldPosition = FieldPosition.HEADER,
    ):
        self.name = name
        self.values: list[str] = list(values) if values is not None else []
        self.kind = kind

    def add(self, value: str) -> None:
        """Append a value to a field."""
        self.values.append(value)

    def set(self, values: list[str]) -> None:
        """Overwrite existing field values."""
        self.values = values

    def remove(self, value: str) -> None:
        """Remove all matching entries from list."""
        try:
            while True:
                self.values.remove(value)
        except ValueError:
            return

    def as_string(self, delimiter: str = ",") -> str:
        """Get delimited string of all values. A comma is used by default.

        If the ``Field`` has zero values, the empty string is returned. If the ``Field``
        has exactly one value, the value is returned unmodified.

        For ``Field``s with more than one value, the values are joined by a comma and a
        space. For such multi-valued ``Field``s, any values that already contain
        commas or double quotes will be surrounded by double quotes. Within any values
        that get quoted, pre-existing double quotes and backslashes are escaped with a
        backslash.
        """
        value_count = len(self.values)
        if value_count == 0:
            return ""
        if value_count == 1:
            return self.values[0]
        return ", ".join(quote_and_escape_field_value(val) for val in self.values)

    def as_tuples(self) -> list[tuple[str, str]]:
        """Get list of ``name``, ``value`` tuples where each tuple represents one
        value."""
        return [(self.name, val) for val in self.values]

    def __eq__(self, other: object) -> bool:
        """Name, values, and kind must match.

        Values order must match.
        """
        if not isinstance(other, Field):
            return False
        return (
            self.name == other.name
            and self.kind is other.kind
            and self.values == other.values
        )

    def __repr__(self) -> str:
        return f"Field(name={self.name!r}, value={self.values!r}, kind={self.kind!r})"

__eq__(other)

Name, values, and kind must match.

Values order must match.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
73
74
75
76
77
78
79
80
81
82
83
84
def __eq__(self, other: object) -> bool:
    """Name, values, and kind must match.

    Values order must match.
    """
    if not isinstance(other, Field):
        return False
    return (
        self.name == other.name
        and self.kind is other.kind
        and self.values == other.values
    )

add(value)

Append a value to a field.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
33
34
35
def add(self, value: str) -> None:
    """Append a value to a field."""
    self.values.append(value)

as_string(delimiter=',')

Get delimited string of all values. A comma is used by default.

If the Field has zero values, the empty string is returned. If the Field has exactly one value, the value is returned unmodified.

For Fields with more than one value, the values are joined by a comma and a space. For such multi-valued Fields, any values that already contain commas or double quotes will be surrounded by double quotes. Within any values that get quoted, pre-existing double quotes and backslashes are escaped with a backslash.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def as_string(self, delimiter: str = ",") -> str:
    """Get delimited string of all values. A comma is used by default.

    If the ``Field`` has zero values, the empty string is returned. If the ``Field``
    has exactly one value, the value is returned unmodified.

    For ``Field``s with more than one value, the values are joined by a comma and a
    space. For such multi-valued ``Field``s, any values that already contain
    commas or double quotes will be surrounded by double quotes. Within any values
    that get quoted, pre-existing double quotes and backslashes are escaped with a
    backslash.
    """
    value_count = len(self.values)
    if value_count == 0:
        return ""
    if value_count == 1:
        return self.values[0]
    return ", ".join(quote_and_escape_field_value(val) for val in self.values)

as_tuples()

Get list of name, value tuples where each tuple represents one value.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
68
69
70
71
def as_tuples(self) -> list[tuple[str, str]]:
    """Get list of ``name``, ``value`` tuples where each tuple represents one
    value."""
    return [(self.name, val) for val in self.values]

remove(value)

Remove all matching entries from list.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
41
42
43
44
45
46
47
def remove(self, value: str) -> None:
    """Remove all matching entries from list."""
    try:
        while True:
            self.values.remove(value)
    except ValueError:
        return

set(values)

Overwrite existing field values.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
37
38
39
def set(self, values: list[str]) -> None:
    """Overwrite existing field values."""
    self.values = values

Fields

Bases: Fields

Source code in packages/smithy-http/src/smithy_http/__init__.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class Fields(interfaces.Fields):
    def __init__(
        self,
        initial: Iterable[interfaces.Field] | None = None,
        *,
        encoding: str = "utf-8",
    ):
        """Collection of header and trailer entries mapped by name.

        :param initial: Initial list of ``Field`` objects. ``Field``s can also be added
        and later removed.
        :param encoding: The string encoding to be used when converting the ``Field``
        name and value from ``str`` to ``bytes`` for transmission.
        """
        init_fields = list(initial) if initial is not None else []
        init_field_names = [self._normalize_field_name(fld.name) for fld in init_fields]
        fname_counter = Counter(init_field_names)
        repeated_names_exist = (
            len(init_fields) > 0 and fname_counter.most_common(1)[0][1] > 1
        )
        if repeated_names_exist:
            non_unique_names = [name for name, num in fname_counter.items() if num > 1]
            raise ValueError(
                "Field names of the initial list of fields must be unique. The "
                "following normalized field names appear more than once: "
                f"{', '.join(non_unique_names)}."
            )
        init_tuples = zip(init_field_names, init_fields)
        self.entries: dict[str, interfaces.Field] = OrderedDict(init_tuples)
        self.encoding: str = encoding

    def set_field(self, field: interfaces.Field) -> None:
        """Alias for __setitem__ to utilize the field.name for the entry key."""
        self.__setitem__(field.name, field)

    def __setitem__(self, name: str, field: interfaces.Field) -> None:
        """Set or override entry for a Field name."""
        normalized_name = self._normalize_field_name(name)
        normalized_field_name = self._normalize_field_name(field.name)
        if normalized_name != normalized_field_name:
            raise ValueError(
                f"Supplied key {name} does not match Field.name "
                f"provided: {normalized_field_name}"
            )
        self.entries[normalized_name] = field

    def get(
        self, key: str, default: interfaces.Field | None = None
    ) -> interfaces.Field | None:
        return self[key] if key in self else default

    def __getitem__(self, name: str) -> interfaces.Field:
        """Retrieve Field entry."""
        normalized_name = self._normalize_field_name(name)
        return self.entries[normalized_name]

    def __delitem__(self, name: str) -> None:
        """Delete entry from collection."""
        normalized_name = self._normalize_field_name(name)
        del self.entries[normalized_name]

    def get_by_type(self, kind: FieldPosition) -> list[interfaces.Field]:
        """Helper function for retrieving specific types of fields.

        Used to grab all headers or all trailers.
        """
        return [entry for entry in self.entries.values() if entry.kind is kind]

    def extend(self, other: interfaces.Fields) -> None:
        """Merges ``entries`` of ``other`` into the current ``entries``.

        For every `Field` in the ``entries`` of ``other``: If the normalized name
        already exists in the current ``entries``, the values from ``other`` are
        appended. Otherwise, the ``Field`` is added to the list of ``entries``.
        """
        for other_field in other:
            try:
                cur_field = self.__getitem__(other_field.name)
                for other_value in other_field.values:
                    cur_field.add(other_value)
            except KeyError:
                self.__setitem__(other_field.name, other_field)

    def _normalize_field_name(self, name: str) -> str:
        """Normalize field names.

        For use as key in ``entries``.
        """
        return name.lower()

    def __eq__(self, other: object) -> bool:
        """Encoding must match.

        Entries must match in values and order.
        """
        if not isinstance(other, Fields):
            return False
        return self.encoding == other.encoding and self.entries == other.entries

    def __iter__(self) -> Iterator[interfaces.Field]:
        yield from self.entries.values()

    def __len__(self) -> int:
        return len(self.entries)

    def __repr__(self) -> str:
        return f"Fields({self.entries})"

    def __contains__(self, key: str) -> bool:
        return self._normalize_field_name(key) in self.entries

__delitem__(name)

Delete entry from collection.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
146
147
148
149
def __delitem__(self, name: str) -> None:
    """Delete entry from collection."""
    normalized_name = self._normalize_field_name(name)
    del self.entries[normalized_name]

__eq__(other)

Encoding must match.

Entries must match in values and order.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
180
181
182
183
184
185
186
187
def __eq__(self, other: object) -> bool:
    """Encoding must match.

    Entries must match in values and order.
    """
    if not isinstance(other, Fields):
        return False
    return self.encoding == other.encoding and self.entries == other.entries

__getitem__(name)

Retrieve Field entry.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
141
142
143
144
def __getitem__(self, name: str) -> interfaces.Field:
    """Retrieve Field entry."""
    normalized_name = self._normalize_field_name(name)
    return self.entries[normalized_name]

__init__(initial=None, *, encoding='utf-8')

Collection of header and trailer entries mapped by name.

:param initial: Initial list of Field objects. Fields can also be added and later removed. :param encoding: The string encoding to be used when converting the Field name and value from str to bytes for transmission.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def __init__(
    self,
    initial: Iterable[interfaces.Field] | None = None,
    *,
    encoding: str = "utf-8",
):
    """Collection of header and trailer entries mapped by name.

    :param initial: Initial list of ``Field`` objects. ``Field``s can also be added
    and later removed.
    :param encoding: The string encoding to be used when converting the ``Field``
    name and value from ``str`` to ``bytes`` for transmission.
    """
    init_fields = list(initial) if initial is not None else []
    init_field_names = [self._normalize_field_name(fld.name) for fld in init_fields]
    fname_counter = Counter(init_field_names)
    repeated_names_exist = (
        len(init_fields) > 0 and fname_counter.most_common(1)[0][1] > 1
    )
    if repeated_names_exist:
        non_unique_names = [name for name, num in fname_counter.items() if num > 1]
        raise ValueError(
            "Field names of the initial list of fields must be unique. The "
            "following normalized field names appear more than once: "
            f"{', '.join(non_unique_names)}."
        )
    init_tuples = zip(init_field_names, init_fields)
    self.entries: dict[str, interfaces.Field] = OrderedDict(init_tuples)
    self.encoding: str = encoding

__setitem__(name, field)

Set or override entry for a Field name.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
125
126
127
128
129
130
131
132
133
134
def __setitem__(self, name: str, field: interfaces.Field) -> None:
    """Set or override entry for a Field name."""
    normalized_name = self._normalize_field_name(name)
    normalized_field_name = self._normalize_field_name(field.name)
    if normalized_name != normalized_field_name:
        raise ValueError(
            f"Supplied key {name} does not match Field.name "
            f"provided: {normalized_field_name}"
        )
    self.entries[normalized_name] = field

extend(other)

Merges entries of other into the current entries.

For every Field in the entries of other: If the normalized name already exists in the current entries, the values from other are appended. Otherwise, the Field is added to the list of entries.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def extend(self, other: interfaces.Fields) -> None:
    """Merges ``entries`` of ``other`` into the current ``entries``.

    For every `Field` in the ``entries`` of ``other``: If the normalized name
    already exists in the current ``entries``, the values from ``other`` are
    appended. Otherwise, the ``Field`` is added to the list of ``entries``.
    """
    for other_field in other:
        try:
            cur_field = self.__getitem__(other_field.name)
            for other_value in other_field.values:
                cur_field.add(other_value)
        except KeyError:
            self.__setitem__(other_field.name, other_field)

get_by_type(kind)

Helper function for retrieving specific types of fields.

Used to grab all headers or all trailers.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
151
152
153
154
155
156
def get_by_type(self, kind: FieldPosition) -> list[interfaces.Field]:
    """Helper function for retrieving specific types of fields.

    Used to grab all headers or all trailers.
    """
    return [entry for entry in self.entries.values() if entry.kind is kind]

set_field(field)

Alias for setitem to utilize the field.name for the entry key.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
121
122
123
def set_field(self, field: interfaces.Field) -> None:
    """Alias for __setitem__ to utilize the field.name for the entry key."""
    self.__setitem__(field.name, field)

quote_and_escape_field_value(value)

Escapes and quotes a single :class:Field value if necessary.

See :func:Field.as_string for quoting and escaping logic.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
202
203
204
205
206
207
208
209
210
211
212
def quote_and_escape_field_value(value: str) -> str:
    """Escapes and quotes a single :class:`Field` value if necessary.

    See :func:`Field.as_string` for quoting and escaping logic.
    """
    chars_to_quote = (",", '"')
    if any(char in chars_to_quote for char in value):
        escaped = value.replace("\\", "\\\\").replace('"', '\\"')
        return f'"{escaped}"'
    else:
        return value

tuples_to_fields(tuples, *, kind=None)

Convert name, value tuples to Fields object. Each tuple represents one Field value.

:param kind: The Field kind to define for all tuples.

Source code in packages/smithy-http/src/smithy_http/__init__.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def tuples_to_fields(
    tuples: Iterable[tuple[str, str]], *, kind: FieldPosition | None = None
) -> Fields:
    """Convert ``name``, ``value`` tuples to ``Fields`` object. Each tuple represents
    one Field value.

    :param kind: The Field kind to define for all tuples.
    """
    fields = Fields()
    for name, value in tuples:
        try:
            fields[name].add(value)
        except KeyError:
            fields[name] = Field(
                name=name, values=[value], kind=kind or FieldPosition.HEADER
            )

    return fields

aio

HTTPRequest dataclass

Bases: HTTPRequest

HTTP primitives for an Exchange to construct a version agnostic HTTP message.

Source code in packages/smithy-http/src/smithy_http/aio/__init__.py 
13
14
15
16
17
18
19
20
@dataclass(kw_only=True)
class HTTPRequest(http_aio_interfaces.HTTPRequest):
    """HTTP primitives for an Exchange to construct a version agnostic HTTP message."""

    destination: core_interfaces.URI
    body: core_aio_interfaces.StreamingBlob = field(repr=False, default=b"")
    method: str
    fields: http_interfaces.Fields

HTTPResponse dataclass

Basic implementation of :py:class:.interfaces.HTTPResponse.

Implementations of :py:class:.interfaces.HTTPClient may return instances of this class or of custom response implementations.

Source code in packages/smithy-http/src/smithy_http/aio/__init__.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@dataclass(kw_only=True)
class HTTPResponse:
    """Basic implementation of :py:class:`.interfaces.HTTPResponse`.

    Implementations of :py:class:`.interfaces.HTTPClient` may return instances of this
    class or of custom response implementations.
    """

    body: core_aio_interfaces.StreamingBlob = field(repr=False, default=b"")
    """The response payload as iterable of chunks of bytes."""

    status: int
    """The 3 digit response status code (1xx, 2xx, 3xx, 4xx, 5xx)."""

    fields: http_interfaces.Fields
    """HTTP header and trailer fields."""

    reason: str | None = None
    """Optional string provided by the server explaining the status."""

    async def consume_body_async(self) -> bytes:
        """Iterate over response body and return as bytes."""
        return await read_streaming_blob_async(body=self.body)

    def consume_body(self) -> bytes:
        """Iterate over request body and return as bytes."""
        return read_streaming_blob(body=self.body)

body = field(repr=False, default=b'') class-attribute instance-attribute

The response payload as iterable of chunks of bytes.

fields instance-attribute

HTTP header and trailer fields.

reason = None class-attribute instance-attribute

Optional string provided by the server explaining the status.

status instance-attribute

The 3 digit response status code (1xx, 2xx, 3xx, 4xx, 5xx).

consume_body()

Iterate over request body and return as bytes.

Source code in packages/smithy-http/src/smithy_http/aio/__init__.py 
51
52
53
def consume_body(self) -> bytes:
    """Iterate over request body and return as bytes."""
    return read_streaming_blob(body=self.body)

consume_body_async() async

Iterate over response body and return as bytes.

Source code in packages/smithy-http/src/smithy_http/aio/__init__.py 
47
48
49
async def consume_body_async(self) -> bytes:
    """Iterate over response body and return as bytes."""
    return await read_streaming_blob_async(body=self.body)

aiohttp

AIOHTTPClient

Bases: HTTPClient

Implementation of :py:class:.interfaces.HTTPClient using aiohttp.

Source code in packages/smithy-http/src/smithy_http/aio/aiohttp.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class AIOHTTPClient(HTTPClient):
    """Implementation of :py:class:`.interfaces.HTTPClient` using aiohttp."""

    def __init__(
        self,
        *,
        client_config: AIOHTTPClientConfig | None = None,
        _session: "aiohttp.ClientSession | None" = None,
    ) -> None:
        """
        :param client_config: Configuration that applies to all requests made with this
        client.
        """
        _assert_aiohttp()
        self._config = client_config or AIOHTTPClientConfig()
        self._session = _session or aiohttp.ClientSession()

    async def send(
        self,
        request: HTTPRequest,
        *,
        request_config: HTTPRequestConfiguration | None = None,
    ) -> HTTPResponseInterface:
        """Send HTTP request using aiohttp client.

        :param request: The request including destination URI, fields, payload.
        :param request_config: Configuration specific to this request.
        """
        request_config = request_config or HTTPRequestConfiguration()

        headers_list = list(
            chain.from_iterable(fld.as_tuples() for fld in request.fields)
        )

        body: StreamingBlob = request.body
        if not isinstance(body, AsyncBytesReader):
            body = AsyncBytesReader(body)

        # The typing on `params` is incorrect, it'll happily accept a mapping whose
        # values are lists (or tuples) and produce expected values.
        # See: https://github.com/aio-libs/aiohttp/issues/8563
        async with self._session.request(
            method=request.method,
            url=self._serialize_uri_without_query(request.destination),
            params=parse_qs(request.destination.query),  # type: ignore
            headers=headers_list,
            data=body,
        ) as resp:
            return await self._marshal_response(resp)

    def _serialize_uri_without_query(self, uri: URI) -> yarl.URL:
        """Serialize all parts of the URI up to and including the path."""
        return yarl.URL.build(
            scheme=uri.scheme or "",
            host=uri.host,
            port=uri.port,
            user=uri.username,
            password=uri.password,
            path=uri.path or "",
            encoded=True,
        )

    async def _marshal_response(
        self, aiohttp_resp: "aiohttp.ClientResponse"
    ) -> HTTPResponseInterface:
        """Convert a ``aiohttp.ClientResponse`` to a ``smithy_http.aio.HTTPResponse``"""
        headers = Fields()
        for header_name, header_val in aiohttp_resp.headers.items():
            try:
                headers[header_name].add(header_val)
            except KeyError:
                headers[header_name] = Field(
                    name=header_name,
                    values=[header_val],
                    kind=FieldPosition.HEADER,
                )

        return HTTPResponse(
            status=aiohttp_resp.status,
            fields=headers,
            body=async_list([await aiohttp_resp.read()]),
            reason=aiohttp_resp.reason,
        )

    def __deepcopy__(self, memo: Any) -> "AIOHTTPClient":
        return AIOHTTPClient(
            client_config=deepcopy(self._config),
            _session=copy(self._session),
        )
__init__(*, client_config=None, _session=None)

:param client_config: Configuration that applies to all requests made with this client.

Source code in packages/smithy-http/src/smithy_http/aio/aiohttp.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    *,
    client_config: AIOHTTPClientConfig | None = None,
    _session: "aiohttp.ClientSession | None" = None,
) -> None:
    """
    :param client_config: Configuration that applies to all requests made with this
    client.
    """
    _assert_aiohttp()
    self._config = client_config or AIOHTTPClientConfig()
    self._session = _session or aiohttp.ClientSession()
send(request, *, request_config=None) async

Send HTTP request using aiohttp client.

:param request: The request including destination URI, fields, payload. :param request_config: Configuration specific to this request.

Source code in packages/smithy-http/src/smithy_http/aio/aiohttp.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
async def send(
    self,
    request: HTTPRequest,
    *,
    request_config: HTTPRequestConfiguration | None = None,
) -> HTTPResponseInterface:
    """Send HTTP request using aiohttp client.

    :param request: The request including destination URI, fields, payload.
    :param request_config: Configuration specific to this request.
    """
    request_config = request_config or HTTPRequestConfiguration()

    headers_list = list(
        chain.from_iterable(fld.as_tuples() for fld in request.fields)
    )

    body: StreamingBlob = request.body
    if not isinstance(body, AsyncBytesReader):
        body = AsyncBytesReader(body)

    # The typing on `params` is incorrect, it'll happily accept a mapping whose
    # values are lists (or tuples) and produce expected values.
    # See: https://github.com/aio-libs/aiohttp/issues/8563
    async with self._session.request(
        method=request.method,
        url=self._serialize_uri_without_query(request.destination),
        params=parse_qs(request.destination.query),  # type: ignore
        headers=headers_list,
        data=body,
    ) as resp:
        return await self._marshal_response(resp)

auth

apikey

API_KEY_RESOLVER_CONFIG = PropertyKey(key='config', value_type=APIKeyResolverConfig) module-attribute

A context property bearing an API key config.

APIKeyAuthScheme

Bases: AuthScheme[HTTPRequest, APIKeyIdentity, APIKeyIdentityProperties, Any]

An auth scheme containing necessary data and tools for api key auth.

Source code in packages/smithy-http/src/smithy_http/aio/auth/apikey.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class APIKeyAuthScheme(
    AuthScheme[HTTPRequest, APIKeyIdentity, APIKeyIdentityProperties, Any]
):
    """An auth scheme containing necessary data and tools for api key auth."""

    scheme_id = HTTPAPIKeyAuthTrait.id
    _signer: APIKeySigner

    def __init__(
        self, *, name: str, location: APIKeyLocation, scheme: str | None = None
    ) -> None:
        self._signer = APIKeySigner(name=name, location=location, scheme=scheme)

    def identity_properties(
        self, *, context: _TypedProperties
    ) -> APIKeyIdentityProperties:
        config = context.get(API_KEY_RESOLVER_CONFIG)
        if config is not None and config.api_key is not None:
            return {"api_key": config.api_key}
        return {}

    def identity_resolver(
        self, *, context: _TypedProperties
    ) -> IdentityResolver[APIKeyIdentity, APIKeyIdentityProperties]:
        config = context.get(API_KEY_RESOLVER_CONFIG)
        if config is None or config.api_key_identity_resolver is None:
            raise SmithyIdentityError(
                "Attempted to use API key auth, but api_key_identity_resolver was not "
                "set on the config."
            )
        return config.api_key_identity_resolver

    def signer_properties(self, *, context: _TypedProperties) -> Any:
        return {}

    def signer(self) -> Signer[HTTPRequest, APIKeyIdentity, Any]:
        return self._signer

    @classmethod
    def from_trait(cls, trait: HTTPAPIKeyAuthTrait, /) -> Self:
        return cls(name=trait.name, location=trait.location, scheme=trait.scheme)
APIKeyResolverConfig

Bases: Protocol

A config bearing API key properties.

Source code in packages/smithy-http/src/smithy_http/aio/auth/apikey.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class APIKeyResolverConfig(Protocol):
    """A config bearing API key properties."""

    api_key: str | None
    """An explicit API key.

    If not set, it MAY be retrieved from elsewhere by the resolver.
    """

    api_key_identity_resolver: (
        IdentityResolver[APIKeyIdentity, APIKeyIdentityProperties] | None
    )
    """An API key identity resolver.

    The default implementation only checks the explicitly configured key.
    """
api_key instance-attribute

An explicit API key.

If not set, it MAY be retrieved from elsewhere by the resolver.

api_key_identity_resolver instance-attribute

An API key identity resolver.

The default implementation only checks the explicitly configured key.

APIKeySigner

Bases: Signer[HTTPRequest, APIKeyIdentity, Any]

A signer that signs http requests with an api key.

Source code in packages/smithy-http/src/smithy_http/aio/auth/apikey.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class APIKeySigner(Signer[HTTPRequest, APIKeyIdentity, Any]):
    """A signer that signs http requests with an api key."""

    def __init__(
        self, *, name: str, location: APIKeyLocation, scheme: str | None = None
    ) -> None:
        self._name = name
        self._location = location
        self._scheme = scheme

    async def sign(
        self,
        *,
        request: HTTPRequest,
        identity: APIKeyIdentity,
        properties: Any,
    ) -> HTTPRequest:
        match self._location:
            case APIKeyLocation.QUERY:
                query = request.destination.query or ""
                if query:
                    query += "&"
                query += f"{self._name}={identity.api_key}"
                request.destination = URI(
                    scheme=request.destination.scheme,
                    username=request.destination.username,
                    password=request.destination.password,
                    host=request.destination.host,
                    port=request.destination.port,
                    path=request.destination.password,
                    query=query,
                    fragment=request.destination.fragment,
                )
            case APIKeyLocation.HEADER:
                value = identity.api_key
                if self._scheme is not None:
                    value = f"{self._scheme} {value}"
                request.fields.set_field(Field(name=self._name, values=[value]))

        return request

crt

AWSCRTHTTPClient

Bases: HTTPClient

Source code in packages/smithy-http/src/smithy_http/aio/crt.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
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
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
class AWSCRTHTTPClient(http_aio_interfaces.HTTPClient):
    _HTTP_PORT = 80
    _HTTPS_PORT = 443

    def __init__(
        self,
        eventloop: _AWSCRTEventLoop | None = None,
        client_config: AWSCRTHTTPClientConfig | None = None,
    ) -> None:
        """
        :param client_config: Configuration that applies to all requests made with this
        client.
        """
        _assert_crt()
        self._config = client_config or AWSCRTHTTPClientConfig()
        if eventloop is None:
            eventloop = _AWSCRTEventLoop()
        self._eventloop = eventloop
        self._client_bootstrap = self._eventloop.bootstrap
        self._tls_ctx = crt_io.ClientTlsContext(crt_io.TlsContextOptions())
        self._socket_options = crt_io.SocketOptions()
        self._connections: ConnectionPoolDict = {}
        self._async_reads: set[asyncio.Task[Any]] = set()

    async def send(
        self,
        request: http_aio_interfaces.HTTPRequest,
        *,
        request_config: http_aio_interfaces.HTTPRequestConfiguration | None = None,
    ) -> AWSCRTHTTPResponse:
        """Send HTTP request using awscrt client.

        :param request: The request including destination URI, fields, payload.
        :param request_config: Configuration specific to this request.
        """
        crt_request, crt_body = await self._marshal_request(request)
        connection = await self._get_connection(request.destination)
        response_body = CRTResponseBody()
        response_factory = CRTResponseFactory(response_body)
        crt_stream = connection.request(
            crt_request,
            response_factory.on_response,
            response_body.on_body,
        )
        response_factory.set_done_callback(crt_stream)
        response_body.set_stream(crt_stream)
        crt_stream.completion_future.add_done_callback(
            partial(self._close_input_body, body=crt_body)
        )

        response = await response_factory.await_response()
        if response.status != 200 and response.status >= 300:
            await close(crt_body)

        return response

    def _close_input_body(
        self, future: ConcurrentFuture[int], *, body: "BufferableByteStream | BytesIO"
    ) -> None:
        if future.exception(timeout=0):
            body.close()

    async def _create_connection(
        self, url: core_interfaces.URI
    ) -> "crt_http.HttpClientConnection":
        """Builds and validates connection to ``url``"""
        connect_future = self._build_new_connection(url)
        connection = await asyncio.wrap_future(connect_future)
        self._validate_connection(connection)
        return connection

    async def _get_connection(
        self, url: core_interfaces.URI
    ) -> "crt_http.HttpClientConnection":
        # TODO: Use CRT connection pooling instead of this basic kind
        connection_key = (url.scheme, url.host, url.port)
        connection = self._connections.get(connection_key)

        if connection and connection.is_open():
            return connection

        connection = await self._create_connection(url)
        self._connections[connection_key] = connection
        return connection

    def _build_new_connection(
        self, url: core_interfaces.URI
    ) -> ConcurrentFuture["crt_http.HttpClientConnection"]:
        if url.scheme == "http":
            port = self._HTTP_PORT
            tls_connection_options = None
        elif url.scheme == "https":
            port = self._HTTPS_PORT
            tls_connection_options = self._tls_ctx.new_connection_options()
            tls_connection_options.set_server_name(url.host)
            # TODO: Support TLS configuration, including alpn
            tls_connection_options.set_alpn_list(["h2", "http/1.1"])
        else:
            raise SmithyHTTPError(
                f"AWSCRTHTTPClient does not support URL scheme {url.scheme}"
            )
        if url.port is not None:
            port = url.port

        connect_future = cast(
            ConcurrentFuture[crt_http.HttpClientConnection],
            crt_http.HttpClientConnection.new(
                bootstrap=self._client_bootstrap,
                host_name=url.host,
                port=port,
                socket_options=self._socket_options,
                tls_connection_options=tls_connection_options,
            ),
        )
        return connect_future

    def _validate_connection(self, connection: "crt_http.HttpClientConnection") -> None:
        """Validates an existing connection against the client config.

        Checks performed:
        * If ``force_http_2`` is enabled: Is the connection HTTP/2?
        """
        force_http_2 = self._config.force_http_2
        if force_http_2 and connection.version is not crt_http.HttpVersion.Http2:
            connection.close()
            negotiated = crt_http.HttpVersion(connection.version).name
            raise SmithyHTTPError(f"HTTP/2 could not be negotiated: {negotiated}")

    def _render_path(self, url: core_interfaces.URI) -> str:
        path = url.path if url.path is not None else "/"
        query = f"?{url.query}" if url.query is not None else ""
        return f"{path}{query}"

    async def _marshal_request(
        self, request: http_aio_interfaces.HTTPRequest
    ) -> tuple["crt_http.HttpRequest", "BufferableByteStream | BytesIO"]:
        """Create :py:class:`awscrt.http.HttpRequest` from
        :py:class:`smithy_http.aio.HTTPRequest`"""
        headers_list = []
        if "host" not in request.fields:
            request.fields.set_field(
                Field(name="host", values=[request.destination.host])
            )

        if "accept" not in request.fields:
            request.fields.set_field(Field(name="accept", values=["*/*"]))

        for fld in request.fields.entries.values():
            # TODO: Use literal values for "header"/"trailer".
            if fld.kind.value != FieldPosition.HEADER.value:
                continue
            for val in fld.values:
                headers_list.append((fld.name, val))

        path = self._render_path(request.destination)
        headers = crt_http.HttpHeaders(headers_list)

        body = request.body
        if isinstance(body, bytes | bytearray):
            # If the body is already directly in memory, wrap in a BytesIO to hand
            # off to CRT.
            crt_body = BytesIO(body)
        else:
            # If the body is async, or potentially very large, start up a task to read
            # it into the intermediate object that CRT needs. By using
            # asyncio.create_task we'll start the coroutine without having to
            # explicitly await it.
            crt_body = BufferableByteStream()

            if not isinstance(body, AsyncIterable):
                body = AsyncBytesReader(body)

            # Start the read task in the background.
            read_task = asyncio.create_task(self._consume_body_async(body, crt_body))

            # Keep track of the read task so that it doesn't get garbage colllected,
            # and stop tracking it once it's done.
            self._async_reads.add(read_task)
            read_task.add_done_callback(self._async_reads.discard)

        crt_request = crt_http.HttpRequest(
            method=request.method,
            path=path,
            headers=headers,
            body_stream=crt_body,
        )
        return crt_request, crt_body

    async def _consume_body_async(
        self, source: AsyncIterable[bytes], dest: "BufferableByteStream"
    ) -> None:
        try:
            async for chunk in source:
                dest.write(chunk)
        except Exception:
            dest.close()
            raise
        finally:
            await close(source)
        dest.end_stream()

    def __deepcopy__(self, memo: Any) -> "AWSCRTHTTPClient":
        return AWSCRTHTTPClient(
            eventloop=self._eventloop,
            client_config=deepcopy(self._config),
        )
__init__(eventloop=None, client_config=None)

:param client_config: Configuration that applies to all requests made with this client.

Source code in packages/smithy-http/src/smithy_http/aio/crt.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def __init__(
    self,
    eventloop: _AWSCRTEventLoop | None = None,
    client_config: AWSCRTHTTPClientConfig | None = None,
) -> None:
    """
    :param client_config: Configuration that applies to all requests made with this
    client.
    """
    _assert_crt()
    self._config = client_config or AWSCRTHTTPClientConfig()
    if eventloop is None:
        eventloop = _AWSCRTEventLoop()
    self._eventloop = eventloop
    self._client_bootstrap = self._eventloop.bootstrap
    self._tls_ctx = crt_io.ClientTlsContext(crt_io.TlsContextOptions())
    self._socket_options = crt_io.SocketOptions()
    self._connections: ConnectionPoolDict = {}
    self._async_reads: set[asyncio.Task[Any]] = set()
send(request, *, request_config=None) async

Send HTTP request using awscrt client.

:param request: The request including destination URI, fields, payload. :param request_config: Configuration specific to this request.

Source code in packages/smithy-http/src/smithy_http/aio/crt.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
async def send(
    self,
    request: http_aio_interfaces.HTTPRequest,
    *,
    request_config: http_aio_interfaces.HTTPRequestConfiguration | None = None,
) -> AWSCRTHTTPResponse:
    """Send HTTP request using awscrt client.

    :param request: The request including destination URI, fields, payload.
    :param request_config: Configuration specific to this request.
    """
    crt_request, crt_body = await self._marshal_request(request)
    connection = await self._get_connection(request.destination)
    response_body = CRTResponseBody()
    response_factory = CRTResponseFactory(response_body)
    crt_stream = connection.request(
        crt_request,
        response_factory.on_response,
        response_body.on_body,
    )
    response_factory.set_done_callback(crt_stream)
    response_body.set_stream(crt_stream)
    crt_stream.completion_future.add_done_callback(
        partial(self._close_input_body, body=crt_body)
    )

    response = await response_factory.await_response()
    if response.status != 200 and response.status >= 300:
        await close(crt_body)

    return response

AWSCRTHTTPResponse

Bases: HTTPResponse

Source code in packages/smithy-http/src/smithy_http/aio/crt.py 
 65
 66
 67
 68
 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
class AWSCRTHTTPResponse(http_aio_interfaces.HTTPResponse):
    def __init__(self, *, status: int, fields: Fields, body: "CRTResponseBody") -> None:
        _assert_crt()
        self._status = status
        self._fields = fields
        self._body = body

    @property
    def status(self) -> int:
        return self._status

    @property
    def fields(self) -> Fields:
        return self._fields

    @property
    def body(self) -> AsyncIterable[bytes]:
        return self.chunks()

    @property
    def reason(self) -> str | None:
        """Optional string provided by the server explaining the status."""
        # TODO: See how CRT exposes reason.
        return None

    async def chunks(self) -> AsyncGenerator[bytes, None]:
        while True:
            chunk = await self._body.next()
            if chunk:
                yield chunk
            else:
                break

    def __repr__(self) -> str:
        return (
            f"AWSCRTHTTPResponse("
            f"status={self.status}, "
            f"fields={self.fields!r}, body=...)"
        )
reason property

Optional string provided by the server explaining the status.

BufferableByteStream

Bases: BufferedIOBase

A non-blocking bytes buffer.

Source code in packages/smithy-http/src/smithy_http/aio/crt.py 
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
class BufferableByteStream(BufferedIOBase):
    """A non-blocking bytes buffer."""

    def __init__(self) -> None:
        # We're always manipulating the front and back of the buffer, so a deque
        # will be much more efficient than a list.
        self._chunks: deque[bytes] = deque()
        self._closed = False
        self._done = False

    def read(self, size: int | None = -1) -> bytes:
        if self._closed:
            return b""

        if len(self._chunks) == 0:
            if self._done:
                self.close()
                return b""
            else:
                # When the CRT recieves this, it'll try again
                raise BlockingIOError("read")

        # We could compile all the chunks here instead of just returning
        # the one, BUT the CRT will keep calling read until empty bytes
        # are returned. So it's actually better to just return one chunk
        # since combining them would have some potentially bad memory
        # usage issues.
        result = self._chunks.popleft()
        if size is not None and size > 0:
            remainder = result[size:]
            result = result[:size]
            if remainder:
                self._chunks.appendleft(remainder)

        if self._done and len(self._chunks) == 0:
            self.close()

        return result

    def read1(self, size: int = -1) -> bytes:
        return self.read(size)

    def readinto(self, buffer: "WriteableBuffer") -> int:
        if not isinstance(buffer, memoryview):
            buffer = memoryview(buffer).cast("B")

        data = self.read(len(buffer))  # type: ignore
        n = len(data)
        buffer[:n] = data
        return n

    def write(self, buffer: "ReadableBuffer") -> int:
        if not isinstance(buffer, bytes):
            raise ValueError(
                f"Unexpected value written to BufferableByteStream. "
                f"Only bytes are support but {type(buffer)} was provided."
            )

        if self._closed:
            raise OSError("Stream is completed and doesn't support further writes.")

        if buffer:
            self._chunks.append(buffer)
        return len(buffer)

    @property
    def closed(self) -> bool:
        return self._closed

    def close(self) -> None:
        self._closed = True
        self._done = True

        # Clear out the remaining chunks so that they don't sit around in memory.
        self._chunks.clear()

    def end_stream(self) -> None:
        """End the stream, letting any remaining chunks be read before it is closed."""
        if len(self._chunks) == 0:
            self.close()
        else:
            self._done = True
end_stream()

End the stream, letting any remaining chunks be read before it is closed.

Source code in packages/smithy-http/src/smithy_http/aio/crt.py 
489
490
491
492
493
494
def end_stream(self) -> None:
    """End the stream, letting any remaining chunks be read before it is closed."""
    if len(self._chunks) == 0:
        self.close()
    else:
        self._done = True

identity

apikey

APIKeyIdentity dataclass

Bases: Identity

The identity for auth that uses an api key.

Source code in packages/smithy-http/src/smithy_http/aio/identity/apikey.py 
12
13
14
15
16
17
18
19
@dataclass(kw_only=True)
class APIKeyIdentity(Identity):
    """The identity for auth that uses an api key."""

    api_key: str
    """The API Key to add to requests."""

    expiration: datetime | None = None
api_key instance-attribute

The API Key to add to requests.

APIKeyIdentityResolver

Bases: IdentityResolver[APIKeyIdentity, APIKeyIdentityProperties]

Loads the API key identity from the configuration.

Source code in packages/smithy-http/src/smithy_http/aio/identity/apikey.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
class APIKeyIdentityResolver(
    IdentityResolver[APIKeyIdentity, APIKeyIdentityProperties]
):
    """Loads the API key identity from the configuration."""

    async def get_identity(
        self, *, properties: APIKeyIdentityProperties
    ) -> APIKeyIdentity:
        if (api_key := properties.get("api_key")) is not None:
            return APIKeyIdentity(api_key=api_key)
        raise SmithyIdentityError(
            "Attempted to use API key auth, but api_key was not set on the config."
        )

interfaces

HTTPClient

Bases: ClientTransport[HTTPRequest, HTTPResponse], Protocol

An asynchronous HTTP client interface.

Source code in packages/smithy-http/src/smithy_http/aio/interfaces/__init__.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class HTTPClient(ClientTransport[HTTPRequest, HTTPResponse], Protocol):
    """An asynchronous HTTP client interface."""

    def __init__(self, *, client_config: HTTPClientConfiguration | None) -> None:
        """
        :param client_config: Configuration that applies to all requests made with this
        client.
        """
        ...

    async def send(
        self,
        request: HTTPRequest,
        *,
        request_config: HTTPRequestConfiguration | None = None,
    ) -> HTTPResponse:
        """Send HTTP request over the wire and return the response.

        :param request: The request including destination URI, fields, payload.
        :param request_config: Configuration specific to this request.
        """
        ...
__init__(*, client_config)

:param client_config: Configuration that applies to all requests made with this client.

Source code in packages/smithy-http/src/smithy_http/aio/interfaces/__init__.py 
69
70
71
72
73
74
def __init__(self, *, client_config: HTTPClientConfiguration | None) -> None:
    """
    :param client_config: Configuration that applies to all requests made with this
    client.
    """
    ...
send(request, *, request_config=None) async

Send HTTP request over the wire and return the response.

:param request: The request including destination URI, fields, payload. :param request_config: Configuration specific to this request.

Source code in packages/smithy-http/src/smithy_http/aio/interfaces/__init__.py 
76
77
78
79
80
81
82
83
84
85
86
87
async def send(
    self,
    request: HTTPRequest,
    *,
    request_config: HTTPRequestConfiguration | None = None,
) -> HTTPResponse:
    """Send HTTP request over the wire and return the response.

    :param request: The request including destination URI, fields, payload.
    :param request_config: Configuration specific to this request.
    """
    ...

HTTPErrorIdentifier

A class that uses HTTP response metadata to identify errors.

The body of the response SHOULD NOT be touched by this. The payload codec will be used instead to check for an ID in the body.

Source code in packages/smithy-http/src/smithy_http/aio/interfaces/__init__.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class HTTPErrorIdentifier:
    """A class that uses HTTP response metadata to identify errors.

    The body of the response SHOULD NOT be touched by this. The payload codec will be
    used instead to check for an ID in the body.
    """

    def identify(
        self,
        *,
        operation: APIOperation[Any, Any],
        response: HTTPResponse,
    ) -> ShapeID | None:
        """Idenitfy the ShapeID of an error from an HTTP response."""
identify(*, operation, response)

Idenitfy the ShapeID of an error from an HTTP response.

Source code in packages/smithy-http/src/smithy_http/aio/interfaces/__init__.py 
 97
 98
 99
100
101
102
103
def identify(
    self,
    *,
    operation: APIOperation[Any, Any],
    response: HTTPResponse,
) -> ShapeID | None:
    """Idenitfy the ShapeID of an error from an HTTP response."""

HTTPRequest

Bases: Request, Protocol

HTTP primitive for an Exchange to construct a version agnostic HTTP message.

:param destination: The URI where the request should be sent to. :param method: The HTTP method of the request, for example "GET". :param fields: Fields object containing HTTP headers and trailers. :param body: A streamable collection of bytes.

Source code in packages/smithy-http/src/smithy_http/aio/interfaces/__init__.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class HTTPRequest(Request, Protocol):
    """HTTP primitive for an Exchange to construct a version agnostic HTTP message.

    :param destination: The URI where the request should be sent to.
    :param method: The HTTP method of the request, for example "GET".
    :param fields: ``Fields`` object containing HTTP headers and trailers.
    :param body: A streamable collection of bytes.
    """

    method: str
    fields: Fields

    async def consume_body_async(self) -> bytes:
        """Iterate over request body and return as bytes."""
        return await read_streaming_blob_async(self.body)

    def consume_body(self) -> bytes:
        """Iterate over request body and return as bytes."""
        return read_streaming_blob(self.body)
consume_body()

Iterate over request body and return as bytes.

Source code in packages/smithy-http/src/smithy_http/aio/interfaces/__init__.py 
33
34
35
def consume_body(self) -> bytes:
    """Iterate over request body and return as bytes."""
    return read_streaming_blob(self.body)
consume_body_async() async

Iterate over request body and return as bytes.

Source code in packages/smithy-http/src/smithy_http/aio/interfaces/__init__.py 
29
30
31
async def consume_body_async(self) -> bytes:
    """Iterate over request body and return as bytes."""
    return await read_streaming_blob_async(self.body)

HTTPResponse

Bases: Response, Protocol

HTTP primitives returned from an Exchange, used to construct a client response.

Source code in packages/smithy-http/src/smithy_http/aio/interfaces/__init__.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class HTTPResponse(Response, Protocol):
    """HTTP primitives returned from an Exchange, used to construct a client
    response."""

    @property
    def status(self) -> int:
        """The 3 digit response status code (1xx, 2xx, 3xx, 4xx, 5xx)."""
        ...

    @property
    def fields(self) -> Fields:
        """``Fields`` object containing HTTP headers and trailers."""
        ...

    @property
    def reason(self) -> str | None:
        """Optional string provided by the server explaining the status."""
        ...

    async def consume_body_async(self) -> bytes:
        """Iterate over request body and return as bytes."""
        return await read_streaming_blob_async(self.body)

    def consume_body(self) -> bytes:
        """Iterate over request body and return as bytes."""
        return read_streaming_blob(self.body)
fields property

Fields object containing HTTP headers and trailers.

reason property

Optional string provided by the server explaining the status.

status property

The 3 digit response status code (1xx, 2xx, 3xx, 4xx, 5xx).

consume_body()

Iterate over request body and return as bytes.

Source code in packages/smithy-http/src/smithy_http/aio/interfaces/__init__.py 
61
62
63
def consume_body(self) -> bytes:
    """Iterate over request body and return as bytes."""
    return read_streaming_blob(self.body)
consume_body_async() async

Iterate over request body and return as bytes.

Source code in packages/smithy-http/src/smithy_http/aio/interfaces/__init__.py 
57
58
59
async def consume_body_async(self) -> bytes:
    """Iterate over request body and return as bytes."""
    return await read_streaming_blob_async(self.body)

protocols

HttpBindingClientProtocol

Bases: HttpClientProtocol

An HTTP-based protocol that uses HTTP binding traits.

Source code in packages/smithy-http/src/smithy_http/aio/protocols.py 
 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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
class HttpBindingClientProtocol(HttpClientProtocol):
    """An HTTP-based protocol that uses HTTP binding traits."""

    @property
    def payload_codec(self) -> Codec:
        """The codec used for the serde of input and output payloads."""
        raise NotImplementedError()

    @property
    def content_type(self) -> str:
        """The media type of the http payload."""
        raise NotImplementedError()

    @property
    def error_identifier(self) -> HTTPErrorIdentifier:
        """The class used to identify the shape IDs of errors based on fields or other
        response information."""
        raise NotImplementedError()

    def serialize_request[
        OperationInput: "SerializeableShape",
        OperationOutput: "DeserializeableShape",
    ](
        self,
        *,
        operation: APIOperation[OperationInput, OperationOutput],
        input: OperationInput,
        endpoint: URI,
        context: TypedProperties,
    ) -> HTTPRequest:
        # TODO(optimization): request binding cache like done in SJ
        serializer = HTTPRequestSerializer(
            payload_codec=self.payload_codec,
            http_trait=operation.schema.expect_trait(HTTPTrait),
            endpoint_trait=operation.schema.get_trait(EndpointTrait),
        )

        input.serialize(serializer=serializer)
        request = serializer.result

        if request is None:
            raise ExpectationNotMetError(
                "Expected request to be serialized, but was None"
            )

        return request

    async def deserialize_response[
        OperationInput: "SerializeableShape",
        OperationOutput: "DeserializeableShape",
    ](
        self,
        *,
        operation: APIOperation[OperationInput, OperationOutput],
        request: HTTPRequest,
        response: HTTPResponse,
        error_registry: TypeRegistry,
        context: TypedProperties,
    ) -> OperationOutput:
        if not self._is_success(operation, context, response):
            raise await self._create_error(
                operation=operation,
                request=request,
                response=response,
                response_body=await self._buffer_async_body(response.body),
                error_registry=error_registry,
                context=context,
            )

        # if body is not streaming and is async, we have to buffer it
        body: SyncStreamingBlob | None = None
        if not operation.output_stream_member and not is_streaming_blob(body):
            body = await self._buffer_async_body(response.body)

        # TODO(optimization): response binding cache like done in SJ
        deserializer = HTTPResponseDeserializer(
            payload_codec=self.payload_codec,
            http_trait=operation.schema.expect_trait(HTTPTrait),
            response=response,
            body=body,
        )

        return operation.output.deserialize(deserializer)

    async def _buffer_async_body(self, stream: AsyncStreamingBlob) -> SyncStreamingBlob:
        match stream:
            case AsyncByteStream():
                if not iscoroutinefunction(stream.read):
                    return stream  # type: ignore
                return await stream.read()
            case AsyncIterable():
                full = b""
                async for chunk in stream:
                    full += chunk
                return full
            case _:
                return stream

    def _is_success(
        self,
        operation: APIOperation[Any, Any],
        context: TypedProperties,
        response: HTTPResponse,
    ) -> bool:
        return 200 <= response.status < 300

    async def _create_error(
        self,
        operation: APIOperation[Any, Any],
        request: HTTPRequest,
        response: HTTPResponse,
        response_body: SyncStreamingBlob,
        error_registry: TypeRegistry,
        context: TypedProperties,
    ) -> CallError:
        error_id = self.error_identifier.identify(
            operation=operation, response=response
        )

        if error_id is None and self._matches_content_type(response):
            if isinstance(response_body, bytearray):
                response_body = bytes(response_body)
            deserializer = self.payload_codec.create_deserializer(source=response_body)
            document = deserializer.read_document(schema=DOCUMENT)

            if document.discriminator in error_registry:
                error_id = document.discriminator
                if isinstance(response_body, SeekableBytesReader):
                    response_body.seek(0)

        if error_id is not None and error_id in error_registry:
            error_shape = error_registry.get(error_id)

            # make sure the error shape is derived from modeled exception
            if not issubclass(error_shape, ModeledError):
                raise ExpectationNotMetError(
                    f"Modeled errors must be derived from 'ModeledError', "
                    f"but got {error_shape}"
                )

            deserializer = HTTPResponseDeserializer(
                payload_codec=self.payload_codec,
                http_trait=operation.schema.expect_trait(HTTPTrait),
                response=response,
                body=response_body,
            )
            return error_shape.deserialize(deserializer)

        is_throttle = response.status == 429
        message = (
            f"Unknown error for operation {operation.schema.id} "
            f"- status: {response.status}"
        )
        if error_id is not None:
            message += f" - id: {error_id}"
        if response.reason is not None:
            message += f" - reason: {response.status}"
        return CallError(
            message=message,
            fault="client" if response.status < 500 else "server",
            is_throttling_error=is_throttle,
            is_retry_safe=is_throttle or None,
        )

    def _matches_content_type(self, response: HTTPResponse) -> bool:
        if "content-type" not in response.fields:
            return False
        return response.fields["content-type"].as_string() == self.content_type
content_type property

The media type of the http payload.

error_identifier property

The class used to identify the shape IDs of errors based on fields or other response information.

payload_codec property

The codec used for the serde of input and output payloads.

HttpClientProtocol

Bases: ClientProtocol[HTTPRequest, HTTPResponse]

An HTTP-based protocol.

Source code in packages/smithy-http/src/smithy_http/aio/protocols.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class HttpClientProtocol(ClientProtocol[HTTPRequest, HTTPResponse]):
    """An HTTP-based protocol."""

    def set_service_endpoint(
        self,
        *,
        request: HTTPRequest,
        endpoint: Endpoint,
    ) -> HTTPRequest:
        uri = endpoint.uri
        previous = request.destination

        path = previous.path or uri.path
        if uri.path is not None and previous.path is not None:
            path = os.path.join(uri.path, previous.path.lstrip("/"))

        if path is not None and not path.startswith("/"):
            path = "/" + path

        query = previous.query or uri.query
        if uri.query and previous.query:
            query = f"{uri.query}&{previous.query}"

        request.destination = _URI(
            scheme=uri.scheme,
            username=uri.username or previous.username,
            password=uri.password or previous.password,
            host=uri.host,
            port=uri.port or previous.port,
            path=path,
            query=query,
            fragment=uri.fragment or previous.fragment,
        )

        return request

restjson

parse_rest_json_error_info(http_response, check_body=True) async

Parses generic RestJson error info from an HTTP response.

:param http_response: The HTTP response to parse. :param check_body: Whether to check the body for the code / message. :returns: The parsed error information.

Source code in packages/smithy-http/src/smithy_http/aio/restjson.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
async def parse_rest_json_error_info(
    http_response: HTTPResponse, check_body: bool = True
) -> RestJsonErrorInfo:
    """Parses generic RestJson error info from an HTTP response.

    :param http_response: The HTTP response to parse.
    :param check_body: Whether to check the body for the code / message.
    :returns: The parsed error information.
    """
    code: str | None = None
    message: str | None = None
    json_body: dict[str, DocumentValue] | None = None

    for field in http_response.fields:
        if field.name.lower() == _REST_JSON_CODE_HEADER:
            code = field.values[0]
            break

    if check_body:
        if body := await http_response.consume_body_async():
            try:
                json_body = json.loads(body)
            except json.JSONDecodeError:
                # In some cases the body might end up being HTML depending on the
                # configuration of the server. In those cases we can simply ignore
                # the body because we can't find the information we need there.
                pass

        if json_body:
            for key, value in json_body.items():
                key_lower = key.lower()
                if not code and key_lower in _REST_JSON_CODE_KEYS:
                    code = expect_type(str, value)
                if not message and key_lower in _REST_JSON_MESSAGE_KEYS:
                    message = expect_type(str, value)

    # Normalize the error code. Some services may try to send a fully-qualified shape
    # ID or a URI, but we don't want to include those.
    if code:
        if "#" in code:
            code = code.split("#")[1]
        code = code.split(":")[0]

    return RestJsonErrorInfo(code or "Unknown", message or "Unknown", json_body)

bindings

Binding

Bases: Enum

HTTP binding locations.

Source code in packages/smithy-http/src/smithy_http/bindings.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class Binding(Enum):
    """HTTP binding locations."""

    HEADER = 0
    """Indicates the member is bound to a header."""

    QUERY = 1
    """Indicates the member is bound to a query parameter."""

    PAYLOAD = 2
    """Indicates the member is bound to the entire HTTP payload."""

    BODY = 3
    """Indicates the member is a property in the HTTP payload structure."""

    LABEL = 4
    """Indicates the member is bound to a path segment in the URI."""

    STATUS = 5
    """Indicates the member is bound to the response status code."""

    PREFIX_HEADERS = 6
    """Indicates the member is bound to multiple headers with a shared prefix."""

    QUERY_PARAMS = 7
    """Indicates the member is bound to the query string as multiple key-value pairs."""

    HOST = 8
    """Indicates the member is bound to a prefix to the host AND to the body."""

BODY = 3 class-attribute instance-attribute

Indicates the member is a property in the HTTP payload structure.

HEADER = 0 class-attribute instance-attribute

Indicates the member is bound to a header.

HOST = 8 class-attribute instance-attribute

Indicates the member is bound to a prefix to the host AND to the body.

LABEL = 4 class-attribute instance-attribute

Indicates the member is bound to a path segment in the URI.

PAYLOAD = 2 class-attribute instance-attribute

Indicates the member is bound to the entire HTTP payload.

PREFIX_HEADERS = 6 class-attribute instance-attribute

Indicates the member is bound to multiple headers with a shared prefix.

QUERY = 1 class-attribute instance-attribute

Indicates the member is bound to a query parameter.

QUERY_PARAMS = 7 class-attribute instance-attribute

Indicates the member is bound to the query string as multiple key-value pairs.

STATUS = 5 class-attribute instance-attribute

Indicates the member is bound to the response status code.

RequestBindingMatcher dataclass

Bases: _BindingMatcher

Matches structure members to HTTP request binding locations.

Source code in packages/smithy-http/src/smithy_http/bindings.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
@dataclass(init=False)
class RequestBindingMatcher(_BindingMatcher):
    """Matches structure members to HTTP request binding locations."""

    def __init__(self, struct: Schema) -> None:
        """Initialize a RequestBindingMatcher.

        :param struct: The structure to examine for HTTP bindings.
        """
        super().__init__(struct, -1)

    def _do_match(self, member: Schema) -> Binding:
        if HTTPLabelTrait.id in member.traits:
            return Binding.LABEL
        if HTTPQueryTrait.id in member.traits:
            return Binding.QUERY
        if HTTPQueryParamsTrait.id in member.traits:
            return Binding.QUERY_PARAMS
        if HTTPHeaderTrait.id in member.traits:
            return Binding.HEADER
        if HTTPPrefixHeadersTrait.id in member.traits:
            return Binding.PREFIX_HEADERS
        if HTTPPayloadTrait.id in member.traits:
            return Binding.PAYLOAD
        if HostLabelTrait.id in member.traits:
            return Binding.HOST
        return Binding.BODY

__init__(struct)

Initialize a RequestBindingMatcher.

:param struct: The structure to examine for HTTP bindings.

Source code in packages/smithy-http/src/smithy_http/bindings.py 
120
121
122
123
124
125
def __init__(self, struct: Schema) -> None:
    """Initialize a RequestBindingMatcher.

    :param struct: The structure to examine for HTTP bindings.
    """
    super().__init__(struct, -1)

ResponseBindingMatcher dataclass

Bases: _BindingMatcher

Matches structure members to HTTP response binding locations.

Source code in packages/smithy-http/src/smithy_http/bindings.py 
145
146
147
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
@dataclass(init=False)
class ResponseBindingMatcher(_BindingMatcher):
    """Matches structure members to HTTP response binding locations."""

    def __init__(self, struct: Schema) -> None:
        """Initialize a ResponseBindingMatcher.

        :param struct: The structure to examine for HTTP bindings.
        """
        super().__init__(struct, self._compute_response(struct))

    def _compute_response(self, struct: Schema) -> int:
        if (http_error := struct.get_trait(HTTPErrorTrait)) is not None:
            return http_error.code
        if (error := struct.get_trait(ErrorTrait)) is not None:
            return 400 if error.fault is ErrorFault.CLIENT else 500
        return -1

    def _do_match(self, member: Schema) -> Binding:
        if HTTPResponseCodeTrait.id in member.traits:
            return Binding.STATUS
        if HTTPHeaderTrait.id in member.traits:
            return Binding.HEADER
        if HTTPPrefixHeadersTrait.id in member.traits:
            return Binding.PREFIX_HEADERS
        if HTTPPayloadTrait.id in member.traits:
            return Binding.PAYLOAD
        return Binding.BODY

__init__(struct)

Initialize a ResponseBindingMatcher.

:param struct: The structure to examine for HTTP bindings.

Source code in packages/smithy-http/src/smithy_http/bindings.py 
149
150
151
152
153
154
def __init__(self, struct: Schema) -> None:
    """Initialize a ResponseBindingMatcher.

    :param struct: The structure to examine for HTTP bindings.
    """
    super().__init__(struct, self._compute_response(struct))

deserializers

HTTPHeaderDeserializer

Bases: SpecificShapeDeserializer

Binds HTTP header values to a deserializable shape.

For headers with list values, see :py:class:HTTPHeaderListDeserializer.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
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
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
class HTTPHeaderDeserializer(SpecificShapeDeserializer):
    """Binds HTTP header values to a deserializable shape.

    For headers with list values, see :py:class:`HTTPHeaderListDeserializer`.
    """

    def __init__(self, value: str) -> None:
        """Initialize an HTTPHeaderDeserializer.

        :param value: The string value of the header.
        """
        self._value = value

    def is_null(self) -> bool:
        return False

    def read_boolean(self, schema: Schema) -> bool:
        return strict_parse_bool(self._value)

    def read_byte(self, schema: Schema) -> int:
        return self.read_integer(schema)

    def read_short(self, schema: Schema) -> int:
        return self.read_integer(schema)

    def read_integer(self, schema: Schema) -> int:
        return int(self._value)

    def read_long(self, schema: Schema) -> int:
        return self.read_integer(schema)

    def read_big_integer(self, schema: Schema) -> int:
        return self.read_integer(schema)

    def read_float(self, schema: Schema) -> float:
        return strict_parse_float(self._value)

    def read_double(self, schema: Schema) -> float:
        return self.read_float(schema)

    def read_big_decimal(self, schema: Schema) -> Decimal:
        return Decimal(self._value).canonical()

    def read_string(self, schema: Schema) -> str:
        if MediaTypeTrait in schema:
            return b64decode(self._value).decode("utf-8")
        return self._value

    def read_timestamp(self, schema: Schema) -> datetime.datetime:
        format = TimestampFormat.HTTP_DATE
        if (trait := schema.get_trait(TimestampFormatTrait)) is not None:
            format = trait.format
        return ensure_utc(format.deserialize(self._value))

__init__(value)

Initialize an HTTPHeaderDeserializer.

:param value: The string value of the header.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
162
163
164
165
166
167
def __init__(self, value: str) -> None:
    """Initialize an HTTPHeaderDeserializer.

    :param value: The string value of the header.
    """
    self._value = value

HTTPHeaderListDeserializer

Bases: SpecificShapeDeserializer

Binds HTTP header lists to a deserializable shape.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
class HTTPHeaderListDeserializer(SpecificShapeDeserializer):
    """Binds HTTP header lists to a deserializable shape."""

    def __init__(self, field: Field) -> None:
        """Initialize an HTTPHeaderListDeserializer.

        :param field: The field to deserialize.
        """
        self._field = field

    def read_list(
        self, schema: Schema, consumer: Callable[["ShapeDeserializer"], None]
    ) -> None:
        values = self._field.values
        if len(values) == 1:
            is_http_date_list = False
            value_schema = schema.members["member"]
            if value_schema.shape_type is ShapeType.TIMESTAMP:
                trait = value_schema.get_trait(TimestampFormatTrait)
                is_http_date_list = (
                    trait is None or trait.format is TimestampFormat.HTTP_DATE
                )
            values = split_header(values[0], is_http_date_list)
        for value in values:
            consumer(HTTPHeaderDeserializer(value))

__init__(field)

Initialize an HTTPHeaderListDeserializer.

:param field: The field to deserialize.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
214
215
216
217
218
219
def __init__(self, field: Field) -> None:
    """Initialize an HTTPHeaderListDeserializer.

    :param field: The field to deserialize.
    """
    self._field = field

HTTPHeaderMapDeserializer

Bases: SpecificShapeDeserializer

Binds HTTP header maps to a deserializable shape.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
class HTTPHeaderMapDeserializer(SpecificShapeDeserializer):
    """Binds HTTP header maps to a deserializable shape."""

    def __init__(self, fields: Fields, prefix: str = "") -> None:
        """Initialize an HTTPHeaderMapDeserializer.

        :param fields: The collection of headers to search for map values.
        :param prefix: An optional prefix to limit which headers are pulled in to the
            map. By default, all headers are pulled in, including headers that are bound
            to other properties on the shape.
        """
        self._prefix = prefix.lower()
        self._fields = fields

    def read_map(
        self,
        schema: Schema,
        consumer: Callable[[str, "ShapeDeserializer"], None],
    ) -> None:
        trim = len(self._prefix)
        for field in self._fields:
            if field.name.lower().startswith(self._prefix):
                consumer(field.name[trim:], HTTPHeaderDeserializer(field.as_string()))

__init__(fields, prefix='')

Initialize an HTTPHeaderMapDeserializer.

:param fields: The collection of headers to search for map values. :param prefix: An optional prefix to limit which headers are pulled in to the map. By default, all headers are pulled in, including headers that are bound to other properties on the shape.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
241
242
243
244
245
246
247
248
249
250
def __init__(self, fields: Fields, prefix: str = "") -> None:
    """Initialize an HTTPHeaderMapDeserializer.

    :param fields: The collection of headers to search for map values.
    :param prefix: An optional prefix to limit which headers are pulled in to the
        map. By default, all headers are pulled in, including headers that are bound
        to other properties on the shape.
    """
    self._prefix = prefix.lower()
    self._fields = fields

HTTPResponseCodeDeserializer

Bases: SpecificShapeDeserializer

Binds HTTP response codes to a deserializeable shape.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
class HTTPResponseCodeDeserializer(SpecificShapeDeserializer):
    """Binds HTTP response codes to a deserializeable shape."""

    def __init__(self, response_code: int) -> None:
        """Initialize an HTTPResponseCodeDeserializer.

        :param response_code: The response code to bind.
        """
        self._response_code = response_code

    def read_byte(self, schema: Schema) -> int:
        return self._response_code

    def read_short(self, schema: Schema) -> int:
        return self._response_code

    def read_integer(self, schema: Schema) -> int:
        return self._response_code

__init__(response_code)

Initialize an HTTPResponseCodeDeserializer.

:param response_code: The response code to bind.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
266
267
268
269
270
271
def __init__(self, response_code: int) -> None:
    """Initialize an HTTPResponseCodeDeserializer.

    :param response_code: The response code to bind.
    """
    self._response_code = response_code

HTTPResponseDeserializer

Bases: SpecificShapeDeserializer

Binds :py:class:HTTPResponse properties to a DeserializableShape.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class HTTPResponseDeserializer(SpecificShapeDeserializer):
    """Binds :py:class:`HTTPResponse` properties to a DeserializableShape."""

    # Note: caller will have to read the body if it's async and not streaming
    def __init__(
        self,
        *,
        payload_codec: Codec,
        response: HTTPResponse,
        http_trait: HTTPTrait | None = None,
        body: "SyncStreamingBlob | None" = None,
    ) -> None:
        """Initialize an HTTPResponseDeserializer.

        :param payload_codec: The Codec to use to deserialize the payload, if present.
        :param response: The HTTP response to read from.
        :param http_trait: The HTTP trait of the operation being handled.
        :param body: The HTTP response body in a synchronously readable form. This is
            necessary for async response bodies when there is no streaming member.
        """
        self._payload_codec = payload_codec
        self._response = response
        self._http_trait = http_trait
        self._body = body

    def read_struct(
        self, schema: Schema, consumer: Callable[[Schema, ShapeDeserializer], None]
    ) -> None:
        binding_matcher = ResponseBindingMatcher(schema)

        for member in schema.members.values():
            match binding_matcher.match(member):
                case Binding.HEADER:
                    trait = member.expect_trait(HTTPHeaderTrait)
                    header = self._response.fields.entries.get(trait.key.lower())
                    if header is not None:
                        if member.shape_type is ShapeType.LIST:
                            consumer(member, HTTPHeaderListDeserializer(header))
                        else:
                            consumer(member, HTTPHeaderDeserializer(header.as_string()))
                case Binding.PREFIX_HEADERS:
                    trait = member.expect_trait(HTTPPrefixHeadersTrait)
                    consumer(
                        member,
                        HTTPHeaderMapDeserializer(self._response.fields, trait.prefix),
                    )
                case Binding.STATUS:
                    consumer(
                        member, HTTPResponseCodeDeserializer(self._response.status)
                    )
                case Binding.PAYLOAD:
                    if binding_matcher.event_stream_member is None:
                        assert binding_matcher.payload_member is not None  # noqa: S101
                        if self._should_read_payload(binding_matcher.payload_member):
                            deserializer = self._create_payload_deserializer(
                                binding_matcher.payload_member
                            )
                            consumer(binding_matcher.payload_member, deserializer)
                case _:
                    pass

        if binding_matcher.has_body and not self._has_empty_body(
            self._response, self._body
        ):
            deserializer = self._create_body_deserializer()
            deserializer.read_struct(schema, consumer)

    def _should_read_payload(self, schema: Schema) -> bool:
        if schema.shape_type not in (
            ShapeType.LIST,
            ShapeType.MAP,
            ShapeType.UNION,
            ShapeType.STRUCTURE,
        ):
            return True
        return not self._has_empty_body(self._response, self._body)

    def _has_empty_body(
        self, response: HTTPResponse, body: "SyncStreamingBlob | None"
    ) -> bool:
        if "content-length" in response.fields:
            return int(response.fields["content-length"].as_string()) == 0
        if isinstance(body, bytes | bytearray):
            return len(body) == 0
        if (seek := getattr(self._body, "seek", None)) is not None:
            content_length = seek(0, 2)
            if content_length == 0:
                return True
            seek(0, 0)
        return False

    def _create_payload_deserializer(self, payload_member: Schema) -> ShapeDeserializer:
        if payload_member.shape_type in (
            ShapeType.BLOB,
            ShapeType.STRING,
            ShapeType.ENUM,
        ):
            body = self._body if self._body is not None else self._response.body
            return RawPayloadDeserializer(body)
        return self._create_body_deserializer()

    def _create_body_deserializer(self):
        body = self._body if self._body is not None else self._response.body
        if not is_streaming_blob(body):
            raise UnsupportedStreamError(
                "Unable to read async stream. This stream must be buffered prior "
                "to creating the deserializer."
            )

        if isinstance(body, bytearray):
            body = bytes(body)

        return self._payload_codec.create_deserializer(body)

__init__(*, payload_codec, response, http_trait=None, body=None)

Initialize an HTTPResponseDeserializer.

:param payload_codec: The Codec to use to deserialize the payload, if present. :param response: The HTTP response to read from. :param http_trait: The HTTP trait of the operation being handled. :param body: The HTTP response body in a synchronously readable form. This is necessary for async response bodies when there is no streaming member.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    *,
    payload_codec: Codec,
    response: HTTPResponse,
    http_trait: HTTPTrait | None = None,
    body: "SyncStreamingBlob | None" = None,
) -> None:
    """Initialize an HTTPResponseDeserializer.

    :param payload_codec: The Codec to use to deserialize the payload, if present.
    :param response: The HTTP response to read from.
    :param http_trait: The HTTP trait of the operation being handled.
    :param body: The HTTP response body in a synchronously readable form. This is
        necessary for async response bodies when there is no streaming member.
    """
    self._payload_codec = payload_codec
    self._response = response
    self._http_trait = http_trait
    self._body = body

RawPayloadDeserializer

Bases: SpecificShapeDeserializer

Binds an HTTP payload to a deserializeable shape.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
class RawPayloadDeserializer(SpecificShapeDeserializer):
    """Binds an HTTP payload to a deserializeable shape."""

    def __init__(self, payload: "AsyncStreamingBlob") -> None:
        """Initialize a RawPayloadDeserializer.

        :param payload: The payload to bind. If the member that is bound to the payload
            is a string or blob, it MUST NOT be an async stream. Async streams MUST be
            buffered into a synchronous stream ahead of time.
        """
        self._payload = payload

    def read_string(self, schema: Schema) -> str:
        return self._consume_payload().decode("utf-8")

    def read_blob(self, schema: Schema) -> bytes:
        return self._consume_payload()

    def read_data_stream(self, schema: Schema) -> "AsyncStreamingBlob":
        if self._is_async_reader(self._payload):
            return self._payload
        return AsyncBytesReader(self._payload)

    def _is_async_reader(self, obj: Any) -> TypeGuard[AsyncByteStream]:
        return isinstance(obj, AsyncByteStream) and iscoroutinefunction(
            getattr(obj, "read")
        )

    def _consume_payload(self) -> bytes:
        if isinstance(self._payload, bytes):
            return self._payload
        if isinstance(self._payload, bytearray):
            return bytes(self._payload)
        if is_bytes_reader(self._payload):
            return self._payload.read()
        raise UnsupportedStreamError(
            "Unable to read async stream. This stream must be buffered prior "
            "to creating the deserializer."
        )

__init__(payload)

Initialize a RawPayloadDeserializer.

:param payload: The payload to bind. If the member that is bound to the payload is a string or blob, it MUST NOT be an async stream. Async streams MUST be buffered into a synchronous stream ahead of time.

Source code in packages/smithy-http/src/smithy_http/deserializers.py 
286
287
288
289
290
291
292
293
def __init__(self, payload: "AsyncStreamingBlob") -> None:
    """Initialize a RawPayloadDeserializer.

    :param payload: The payload to bind. If the member that is bound to the payload
        is a string or blob, it MUST NOT be an async stream. Async streams MUST be
        buffered into a synchronous stream ahead of time.
    """
    self._payload = payload

endpoints

HEADERS = PropertyKey(key='headers', value_type=(interfaces.Fields)) module-attribute

An Endpoint property indicating the given fields MUST be added to the request.

HTTPEndpoint dataclass

Bases: Endpoint

A resolved endpoint with optional HTTP headers.

Source code in packages/smithy-http/src/smithy_http/endpoints.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@dataclass(init=False, kw_only=True)
class HTTPEndpoint(Endpoint):
    """A resolved endpoint with optional HTTP headers."""

    def __init__(
        self,
        *,
        uri: URI,
        properties: _TypedProperties | None = None,
        headers: interfaces.Fields | None = None,
    ) -> None:
        self.uri = uri
        self.properties = properties if properties is not None else TypedProperties()
        headers = headers if headers is not None else Fields()
        self.properties[HEADERS] = headers

exceptions

SmithyHTTPError

Bases: SmithyError

Base exception type for all exceptions raised in HTTP clients.

Source code in packages/smithy-http/src/smithy_http/exceptions.py 
6
7
class SmithyHTTPError(SmithyError):
    """Base exception type for all exceptions raised in HTTP clients."""

interceptors

user_agent

UserAgentInterceptor

Bases: Interceptor[Any, Any, HTTPRequest, None]

Adds interceptors that initialize UserAgent in the context and add the user-agent header.

Source code in packages/smithy-http/src/smithy_http/interceptors/user_agent.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class UserAgentInterceptor(Interceptor[Any, Any, HTTPRequest, None]):
    """Adds interceptors that initialize UserAgent in the context and add the user-agent
    header."""

    def read_before_execution(self, context: InputContext[Any]) -> None:
        context.properties[USER_AGENT] = _UserAgentBuilder.from_environment().build()

    def modify_before_signing(
        self, context: RequestContext[Any, HTTPRequest]
    ) -> HTTPRequest:
        user_agent = context.properties[USER_AGENT]
        request = context.transport_request
        request.fields.set_field(Field(name="User-Agent", values=[str(user_agent)]))
        return context.transport_request

interfaces

Field

Bases: Protocol

A name-value pair representing a single field in a request or response.

The kind will dictate metadata placement within an the message, for example as header or trailer field in a HTTP request as defined in RFC 9110 Section 5.

All field names are case insensitive and case-variance must be treated as equivalent. Names may be normalized but should be preserved for accuracy during transmission.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class Field(Protocol):
    """A name-value pair representing a single field in a request or response.

    The kind will dictate metadata placement within an the message, for example as
    header or trailer field in a HTTP request as defined in RFC 9110 Section 5.

    All field names are case insensitive and case-variance must be treated as
    equivalent. Names may be normalized but should be preserved for accuracy during
    transmission.
    """

    name: str
    values: list[str]
    kind: FieldPosition = FieldPosition.HEADER

    def add(self, value: str) -> None:
        """Append a value to a field."""
        ...

    def set(self, values: list[str]) -> None:
        """Overwrite existing field values."""
        ...

    def remove(self, value: str) -> None:
        """Remove all matching entries from list."""
        ...

    def as_string(self) -> str:
        """Serialize the ``Field``'s values into a single line string."""
        ...

    def as_tuples(self) -> list[tuple[str, str]]:
        """Get list of ``name``, ``value`` tuples where each tuple represents one
        value."""
        ...

add(value)

Append a value to a field.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
45
46
47
def add(self, value: str) -> None:
    """Append a value to a field."""
    ...

as_string()

Serialize the Field's values into a single line string.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
57
58
59
def as_string(self) -> str:
    """Serialize the ``Field``'s values into a single line string."""
    ...

as_tuples()

Get list of name, value tuples where each tuple represents one value.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
61
62
63
64
def as_tuples(self) -> list[tuple[str, str]]:
    """Get list of ``name``, ``value`` tuples where each tuple represents one
    value."""
    ...

remove(value)

Remove all matching entries from list.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
53
54
55
def remove(self, value: str) -> None:
    """Remove all matching entries from list."""
    ...

set(values)

Overwrite existing field values.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
49
50
51
def set(self, values: list[str]) -> None:
    """Overwrite existing field values."""
    ...

FieldPosition

Bases: Enum

The type of a field.

Defines its placement in a request or response.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class FieldPosition(Enum):
    """The type of a field.

    Defines its placement in a request or response.
    """

    HEADER = 0
    """Header field.

    In HTTP this is a header as defined in RFC 9110 Section 6.3. Implementations of
    other protocols may use this FieldPosition for similar types of metadata.
    """

    TRAILER = 1
    """Trailer field.

    In HTTP this is a trailer as defined in RFC 9110 Section 6.5. Implementations of
    other protocols may use this FieldPosition for similar types of metadata.
    """

HEADER = 0 class-attribute instance-attribute

Header field.

In HTTP this is a header as defined in RFC 9110 Section 6.3. Implementations of other protocols may use this FieldPosition for similar types of metadata.

TRAILER = 1 class-attribute instance-attribute

Trailer field.

In HTTP this is a trailer as defined in RFC 9110 Section 6.5. Implementations of other protocols may use this FieldPosition for similar types of metadata.

Fields

Bases: Protocol

Protocol agnostic mapping of key-value pair request metadata, such as HTTP fields.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
 67
 68
 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
106
107
108
109
110
111
112
113
114
115
116
117
class Fields(Protocol):
    """Protocol agnostic mapping of key-value pair request metadata, such as HTTP
    fields."""

    # Entries are keyed off the name of a provided Field
    entries: dict[str, Field]
    encoding: str = "utf-8"

    def set_field(self, field: Field) -> None:
        """Alias for __setitem__ to utilize the field.name for the entry key."""
        ...

    def __setitem__(self, name: str, field: Field) -> None:
        """Set entry for a Field name."""
        ...

    def __getitem__(self, name: str) -> Field:
        """Retrieve Field entry."""
        ...

    def __delitem__(self, name: str) -> None:
        """Delete entry from collection."""
        ...

    def __iter__(self) -> Iterator[Field]:
        """Allow iteration over entries."""
        ...

    def __len__(self) -> int:
        """Get total number of Field entries."""
        ...

    def __contains__(self, key: str) -> bool:
        """Allows in/not in checks on Field entries."""
        ...

    def get_by_type(self, kind: FieldPosition) -> list[Field]:
        """Helper function for retrieving specific types of fields.

        Used to grab all headers or all trailers.
        """
        ...

    def extend(self, other: "Fields") -> None:
        """Merges ``entries`` of ``other`` into the current ``entries``.

        For every `Field` in the ``entries`` of ``other``: If the normalized name
        already exists in the current ``entries``, the values from ``other`` are
        appended. Otherwise, the ``Field`` is added to the list of ``entries``.
        """
        ...

__contains__(key)

Allows in/not in checks on Field entries.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
 99
100
101
def __contains__(self, key: str) -> bool:
    """Allows in/not in checks on Field entries."""
    ...

__delitem__(name)

Delete entry from collection.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
87
88
89
def __delitem__(self, name: str) -> None:
    """Delete entry from collection."""
    ...

__getitem__(name)

Retrieve Field entry.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
83
84
85
def __getitem__(self, name: str) -> Field:
    """Retrieve Field entry."""
    ...

__iter__()

Allow iteration over entries.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
91
92
93
def __iter__(self) -> Iterator[Field]:
    """Allow iteration over entries."""
    ...

__len__()

Get total number of Field entries.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
95
96
97
def __len__(self) -> int:
    """Get total number of Field entries."""
    ...

__setitem__(name, field)

Set entry for a Field name.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
79
80
81
def __setitem__(self, name: str, field: Field) -> None:
    """Set entry for a Field name."""
    ...

extend(other)

Merges entries of other into the current entries.

For every Field in the entries of other: If the normalized name already exists in the current entries, the values from other are appended. Otherwise, the Field is added to the list of entries.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
110
111
112
113
114
115
116
117
def extend(self, other: "Fields") -> None:
    """Merges ``entries`` of ``other`` into the current ``entries``.

    For every `Field` in the ``entries`` of ``other``: If the normalized name
    already exists in the current ``entries``, the values from ``other`` are
    appended. Otherwise, the ``Field`` is added to the list of ``entries``.
    """
    ...

get_by_type(kind)

Helper function for retrieving specific types of fields.

Used to grab all headers or all trailers.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
103
104
105
106
107
108
def get_by_type(self, kind: FieldPosition) -> list[Field]:
    """Helper function for retrieving specific types of fields.

    Used to grab all headers or all trailers.
    """
    ...

set_field(field)

Alias for setitem to utilize the field.name for the entry key.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
75
76
77
def set_field(self, field: Field) -> None:
    """Alias for __setitem__ to utilize the field.name for the entry key."""
    ...

HTTPClientConfiguration dataclass

Client-level HTTP configuration.

:param force_http_2: Whether to require HTTP/2.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
123
124
125
126
127
128
129
130
@dataclass(kw_only=True)
class HTTPClientConfiguration:
    """Client-level HTTP configuration.

    :param force_http_2: Whether to require HTTP/2.
    """

    force_http_2: bool = False

HTTPRequestConfiguration dataclass

Request-level HTTP configuration.

:param read_timeout: How long, in seconds, the client will attempt to read the first byte over an established, open connection before timing out.

Source code in packages/smithy-http/src/smithy_http/interfaces/__init__.py 
133
134
135
136
137
138
139
140
141
@dataclass(kw_only=True)
class HTTPRequestConfiguration:
    """Request-level HTTP configuration.

    :param read_timeout: How long, in seconds, the client will attempt to read the first
        byte over an established, open connection before timing out.
    """

    read_timeout: float | None = None

restjson

RestJsonErrorInfo

Bases: NamedTuple

Generic error information from a RestJson protocol error.

Source code in packages/smithy-http/src/smithy_http/restjson.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class RestJsonErrorInfo(NamedTuple):
    """Generic error information from a RestJson protocol error."""

    code: str
    """The error code."""

    message: str
    """The generic error message.

    A modeled error may have the error bound somewhere else. This is based off of
    checking the most common locations and is intended for use with excpetions that
    either didn't model the message or which are unknown.
    """

    json_body: dict[str, DocumentValue] | None = None
    """The HTTP response body parsed as JSON."""

code instance-attribute

The error code.

json_body = None class-attribute instance-attribute

The HTTP response body parsed as JSON.

message instance-attribute

The generic error message.

A modeled error may have the error bound somewhere else. This is based off of checking the most common locations and is intended for use with excpetions that either didn't model the message or which are unknown.

serializers

CapturingSerializer

Bases: SpecificShapeSerializer

Directly passes along a string through a serializer.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
548
549
550
551
552
553
554
555
556
557
558
559
560
561
class CapturingSerializer(SpecificShapeSerializer):
    """Directly passes along a string through a serializer."""

    result: str | None
    """The captured string.

    This will only be set after the serializer has been used.
    """

    def __init__(self) -> None:
        self.result = None

    def write_string(self, schema: Schema, value: str) -> None:
        self.result = value

result = None instance-attribute

The captured string.

This will only be set after the serializer has been used.

HTTPHeaderMapSerializer

Bases: MapSerializer

Binds a mapping property of a serializeable shape to multiple HTTP headers.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
class HTTPHeaderMapSerializer(MapSerializer):
    """Binds a mapping property of a serializeable shape to multiple HTTP headers."""

    def __init__(self, prefix: str, headers: list[tuple[str, str]]) -> None:
        """Initialize an HTTPHeaderMapSerializer.

        :param prefix: The prefix to prepend to each of the map keys.
        :param headers: The list of header tuples to append to.
        """
        self._prefix = prefix
        self._headers = headers
        self._delegate = CapturingSerializer()

    def entry(self, key: str, value_writer: Callable[[ShapeSerializer], None]):
        value_writer(self._delegate)
        assert self._delegate.result is not None  # noqa: S101
        self._headers.append((self._prefix + key, self._delegate.result))

__init__(prefix, headers)

Initialize an HTTPHeaderMapSerializer.

:param prefix: The prefix to prepend to each of the map keys. :param headers: The list of header tuples to append to.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
532
533
534
535
536
537
538
539
540
def __init__(self, prefix: str, headers: list[tuple[str, str]]) -> None:
    """Initialize an HTTPHeaderMapSerializer.

    :param prefix: The prefix to prepend to each of the map keys.
    :param headers: The list of header tuples to append to.
    """
    self._prefix = prefix
    self._headers = headers
    self._delegate = CapturingSerializer()

HTTPHeaderSerializer

Bases: SpecificShapeSerializer

Binds properties of a serializable shape to HTTP headers.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
class HTTPHeaderSerializer(SpecificShapeSerializer):
    """Binds properties of a serializable shape to HTTP headers."""

    headers: list[tuple[str, str]]
    """A list of serialized headers.

    This should only be accessed after serialization.
    """

    def __init__(
        self, key: str | None = None, headers: list[tuple[str, str]] | None = None
    ) -> None:
        """Initialize an HTTPHeaderSerializer.

        :param key: An optional key to specifically write. If not set, the
            :py:class:`HTTPHeaderTrait` will be checked instead. Required when
            collecting list entries.
        :param headers: An optional list of header tuples to append to. If not
            set, one will be created. Values appended will not be escaped.
        """
        self.headers: list[tuple[str, str]] = headers if headers is not None else []
        self._key = key

    @contextmanager
    def begin_list(self, schema: Schema, size: int) -> Iterator[ShapeSerializer]:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        delegate = HTTPHeaderSerializer(key=key, headers=self.headers)
        yield delegate

    @contextmanager
    def begin_map(self, schema: Schema, size: int) -> Iterator[MapSerializer]:
        prefix = schema.expect_trait(HTTPPrefixHeadersTrait).prefix
        yield HTTPHeaderMapSerializer(prefix, self.headers)

    def write_boolean(self, schema: Schema, value: bool) -> None:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        self.headers.append((key, "true" if value else "false"))

    def write_byte(self, schema: Schema, value: int) -> None:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        self.headers.append((key, str(value)))

    def write_short(self, schema: Schema, value: int) -> None:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        self.headers.append((key, str(value)))

    def write_integer(self, schema: Schema, value: int) -> None:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        self.headers.append((key, str(value)))

    def write_long(self, schema: Schema, value: int) -> None:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        self.headers.append((key, str(value)))

    def write_big_integer(self, schema: Schema, value: int) -> None:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        self.headers.append((key, str(value)))

    def write_float(self, schema: Schema, value: float) -> None:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        self.headers.append((key, serialize_float(value)))

    def write_double(self, schema: Schema, value: float) -> None:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        self.headers.append((key, serialize_float(value)))

    def write_big_decimal(self, schema: Schema, value: Decimal) -> None:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        self.headers.append((key, serialize_float(value)))

    def write_string(self, schema: Schema, value: str) -> None:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        if MediaTypeTrait in schema:
            value = b64encode(value.encode("utf-8")).decode("utf-8")
        self.headers.append((key, value))

    def write_timestamp(self, schema: Schema, value: datetime) -> None:
        key = self._key or schema.expect_trait(HTTPHeaderTrait).key
        format = TimestampFormat.HTTP_DATE
        if (trait := schema.get_trait(TimestampFormatTrait)) is not None:
            format = trait.format
        self.headers.append((key, str(format.serialize(value))))

headers = headers if headers is not None else [] instance-attribute

A list of serialized headers.

This should only be accessed after serialization.

__init__(key=None, headers=None)

Initialize an HTTPHeaderSerializer.

:param key: An optional key to specifically write. If not set, the :py:class:HTTPHeaderTrait will be checked instead. Required when collecting list entries. :param headers: An optional list of header tuples to append to. If not set, one will be created. Values appended will not be escaped.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
454
455
456
457
458
459
460
461
462
463
464
465
466
def __init__(
    self, key: str | None = None, headers: list[tuple[str, str]] | None = None
) -> None:
    """Initialize an HTTPHeaderSerializer.

    :param key: An optional key to specifically write. If not set, the
        :py:class:`HTTPHeaderTrait` will be checked instead. Required when
        collecting list entries.
    :param headers: An optional list of header tuples to append to. If not
        set, one will be created. Values appended will not be escaped.
    """
    self.headers: list[tuple[str, str]] = headers if headers is not None else []
    self._key = key

HTTPPathSerializer

Bases: SpecificShapeSerializer

Binds properties of a serializable shape to the HTTP URI path.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
class HTTPPathSerializer(SpecificShapeSerializer):
    """Binds properties of a serializable shape to the HTTP URI path."""

    def __init__(self, path_pattern: PathPattern) -> None:
        """Initialize an HTTPPathSerializer.

        :param path_pattern: The pattern to bind properties to. This is also used to
            detect greedy labels, which have different escaping requirements.
        """
        self._path_pattern = path_pattern
        self._path_params: dict[str, str] = {}

    @property
    def path(self) -> str:
        """Get the formatted path.

        This must not be accessed before serialization is complete, otherwise an
        exception will be raised.
        """
        return self._path_pattern.format(**self._path_params)

    def write_boolean(self, schema: Schema, value: bool) -> None:
        self._path_params[schema.expect_member_name()] = "true" if value else "false"

    def write_byte(self, schema: Schema, value: int) -> None:
        self.write_integer(schema, value)

    def write_short(self, schema: Schema, value: int) -> None:
        self.write_integer(schema, value)

    def write_integer(self, schema: Schema, value: int) -> None:
        self._path_params[schema.expect_member_name()] = str(value)

    def write_long(self, schema: Schema, value: int) -> None:
        self.write_integer(schema, value)

    def write_big_integer(self, schema: Schema, value: int) -> None:
        self.write_integer(schema, value)

    def write_float(self, schema: Schema, value: float) -> None:
        self._path_params[schema.expect_member_name()] = serialize_float(value)

    def write_double(self, schema: Schema, value: float) -> None:
        self.write_float(schema, value)

    def write_big_decimal(self, schema: Schema, value: Decimal) -> None:
        self._path_params[schema.expect_member_name()] = serialize_float(value)

    def write_string(self, schema: Schema, value: str) -> None:
        key = schema.expect_member_name()
        if key in self._path_pattern.greedy_labels:
            value = urlquote(value)
        else:
            value = urlquote(value, safe="")
        self._path_params[schema.expect_member_name()] = value

    def write_timestamp(self, schema: Schema, value: datetime) -> None:
        format = TimestampFormat.DATE_TIME
        if (trait := schema.get_trait(TimestampFormatTrait)) is not None:
            format = trait.format
        self._path_params[schema.expect_member_name()] = urlquote(
            str(format.serialize(value))
        )

path property

Get the formatted path.

This must not be accessed before serialization is complete, otherwise an exception will be raised.

__init__(path_pattern)

Initialize an HTTPPathSerializer.

:param path_pattern: The pattern to bind properties to. This is also used to detect greedy labels, which have different escaping requirements.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
636
637
638
639
640
641
642
643
def __init__(self, path_pattern: PathPattern) -> None:
    """Initialize an HTTPPathSerializer.

    :param path_pattern: The pattern to bind properties to. This is also used to
        detect greedy labels, which have different escaping requirements.
    """
    self._path_pattern = path_pattern
    self._path_params: dict[str, str] = {}

HTTPQueryMapSerializer

Bases: MapSerializer

Binds properties of a serializable shape to a map of HTTP query params.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
698
699
700
701
702
703
704
705
706
707
708
709
class HTTPQueryMapSerializer(MapSerializer):
    """Binds properties of a serializable shape to a map of HTTP query params."""

    def __init__(self, query_params: list[tuple[str, str]]) -> None:
        """Initialize an HTTPQueryMapSerializer.

        :param query_params: The list of query param tuples to append to.
        """
        self._query_params = query_params

    def entry(self, key: str, value_writer: Callable[[ShapeSerializer], None]):
        value_writer(HTTPQueryMapValueSerializer(key, self._query_params))

__init__(query_params)

Initialize an HTTPQueryMapSerializer.

:param query_params: The list of query param tuples to append to.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
701
702
703
704
705
706
def __init__(self, query_params: list[tuple[str, str]]) -> None:
    """Initialize an HTTPQueryMapSerializer.

    :param query_params: The list of query param tuples to append to.
    """
    self._query_params = query_params

HTTPQueryMapValueSerializer

Bases: SpecificShapeSerializer

Source code in packages/smithy-http/src/smithy_http/serializers.py 
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
class HTTPQueryMapValueSerializer(SpecificShapeSerializer):
    def __init__(self, key: str, query_params: list[tuple[str, str]]) -> None:
        """Initialize an HTTPQueryMapValueSerializer.

        :param key: The key of the query parameter.
        :param query_params: The list of query param tuples to append to.
        """
        self._key = key
        self._query_params = query_params

    def write_string(self, schema: Schema, value: str) -> None:
        # Note: values are escaped when query params are joined
        self._query_params.append((self._key, value))

    @contextmanager
    def begin_list(self, schema: Schema, size: int) -> Iterator[ShapeSerializer]:
        yield self

__init__(key, query_params)

Initialize an HTTPQueryMapValueSerializer.

:param key: The key of the query parameter. :param query_params: The list of query param tuples to append to.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
713
714
715
716
717
718
719
720
def __init__(self, key: str, query_params: list[tuple[str, str]]) -> None:
    """Initialize an HTTPQueryMapValueSerializer.

    :param key: The key of the query parameter.
    :param query_params: The list of query param tuples to append to.
    """
    self._key = key
    self._query_params = query_params

HTTPQuerySerializer

Bases: SpecificShapeSerializer

Binds properties of a serializable shape to HTTP URI query params.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
class HTTPQuerySerializer(SpecificShapeSerializer):
    """Binds properties of a serializable shape to HTTP URI query params."""

    def __init__(
        self, key: str | None = None, params: list[tuple[str, str]] | None = None
    ) -> None:
        """Initialize an HTTPQuerySerializer.

        :param key: An optional key to specifically write. If not set, the
            :py:class:`HTTPQueryTrait` will be checked instead. Required when
            collecting list or map entries.
        :param headers: An optional list of header tuples to append to. If not
            set, one will be created.
        """
        self.query_params: list[tuple[str, str]] = params if params is not None else []
        self._key = key

    @contextmanager
    def begin_list(self, schema: Schema, size: int) -> Iterator[ShapeSerializer]:
        key = self._key or schema.expect_trait(HTTPQueryTrait).key
        yield HTTPQuerySerializer(key=key, params=self.query_params)

    @contextmanager
    def begin_map(self, schema: Schema, size: int) -> Iterator[MapSerializer]:
        yield HTTPQueryMapSerializer(self.query_params)

    def write_boolean(self, schema: Schema, value: bool) -> None:
        key = self._key or schema.expect_trait(HTTPQueryTrait).key
        self.query_params.append((key, "true" if value else "false"))

    def write_byte(self, schema: Schema, value: int) -> None:
        self.write_integer(schema, value)

    def write_short(self, schema: Schema, value: int) -> None:
        self.write_integer(schema, value)

    def write_integer(self, schema: Schema, value: int) -> None:
        key = self._key or schema.expect_trait(HTTPQueryTrait).key
        self.query_params.append((key, str(value)))

    def write_long(self, schema: Schema, value: int) -> None:
        self.write_integer(schema, value)

    def write_big_integer(self, schema: Schema, value: int) -> None:
        self.write_integer(schema, value)

    def write_float(self, schema: Schema, value: float) -> None:
        key = self._key or schema.expect_trait(HTTPQueryTrait).key
        self.query_params.append((key, serialize_float(value)))

    def write_double(self, schema: Schema, value: float) -> None:
        self.write_float(schema, value)

    def write_big_decimal(self, schema: Schema, value: Decimal) -> None:
        key = self._key or schema.expect_trait(HTTPQueryTrait).key
        self.query_params.append((key, serialize_float(value)))

    def write_string(self, schema: Schema, value: str) -> None:
        key = self._key or schema.expect_trait(HTTPQueryTrait).key
        self.query_params.append((key, value))

    def write_timestamp(self, schema: Schema, value: datetime) -> None:
        key = self._key or schema.expect_trait(HTTPQueryTrait).key
        format = TimestampFormat.DATE_TIME
        if (trait := schema.get_trait(TimestampFormatTrait)) is not None:
            format = trait.format
        self.query_params.append((key, str(format.serialize(value))))

__init__(key=None, params=None)

Initialize an HTTPQuerySerializer.

:param key: An optional key to specifically write. If not set, the :py:class:HTTPQueryTrait will be checked instead. Required when collecting list or map entries. :param headers: An optional list of header tuples to append to. If not set, one will be created.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
567
568
569
570
571
572
573
574
575
576
577
578
579
def __init__(
    self, key: str | None = None, params: list[tuple[str, str]] | None = None
) -> None:
    """Initialize an HTTPQuerySerializer.

    :param key: An optional key to specifically write. If not set, the
        :py:class:`HTTPQueryTrait` will be checked instead. Required when
        collecting list or map entries.
    :param headers: An optional list of header tuples to append to. If not
        set, one will be created.
    """
    self.query_params: list[tuple[str, str]] = params if params is not None else []
    self._key = key

HTTPRequestBindingSerializer

Bases: InterceptingSerializer

Delegates HTTP request bindings to binding-location-specific serializers.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
class HTTPRequestBindingSerializer(InterceptingSerializer):
    """Delegates HTTP request bindings to binding-location-specific serializers."""

    def __init__(
        self,
        payload_serializer: ShapeSerializer,
        path_pattern: PathPattern,
        host_prefix_pattern: str,
        binding_matcher: RequestBindingMatcher,
    ) -> None:
        """Initialize an HTTPRequestBindingSerializer.

        :param payload_serializer: The :py:class:`ShapeSerializer` to use to serialize
            the payload, if necessary.
        :param path_pattern: The pattern used to construct the path.
        :host_prefix_pattern: The pattern used to construct the host prefix.
        """
        self._payload_serializer = payload_serializer
        self.header_serializer = HTTPHeaderSerializer()
        self.query_serializer = HTTPQuerySerializer()
        self.path_serializer = HTTPPathSerializer(path_pattern)
        self.host_prefix_serializer = HostPrefixSerializer(
            payload_serializer, host_prefix_pattern
        )
        self._binding_matcher = binding_matcher

    def before(self, schema: Schema) -> ShapeSerializer:
        match self._binding_matcher.match(schema):
            case Binding.HEADER | Binding.PREFIX_HEADERS:
                return self.header_serializer
            case Binding.QUERY | Binding.QUERY_PARAMS:
                return self.query_serializer
            case Binding.LABEL:
                return self.path_serializer
            case Binding.HOST:
                return self.host_prefix_serializer
            case _:
                return self._payload_serializer

    def after(self, schema: Schema) -> None:
        pass

__init__(payload_serializer, path_pattern, host_prefix_pattern, binding_matcher)

Initialize an HTTPRequestBindingSerializer.

:param payload_serializer: The :py:class:ShapeSerializer to use to serialize the payload, if necessary. :param path_pattern: The pattern used to construct the path. :host_prefix_pattern: The pattern used to construct the host prefix.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def __init__(
    self,
    payload_serializer: ShapeSerializer,
    path_pattern: PathPattern,
    host_prefix_pattern: str,
    binding_matcher: RequestBindingMatcher,
) -> None:
    """Initialize an HTTPRequestBindingSerializer.

    :param payload_serializer: The :py:class:`ShapeSerializer` to use to serialize
        the payload, if necessary.
    :param path_pattern: The pattern used to construct the path.
    :host_prefix_pattern: The pattern used to construct the host prefix.
    """
    self._payload_serializer = payload_serializer
    self.header_serializer = HTTPHeaderSerializer()
    self.query_serializer = HTTPQuerySerializer()
    self.path_serializer = HTTPPathSerializer(path_pattern)
    self.host_prefix_serializer = HostPrefixSerializer(
        payload_serializer, host_prefix_pattern
    )
    self._binding_matcher = binding_matcher

HTTPRequestSerializer

Bases: SpecificShapeSerializer

Binds a serializable shape to an HTTP request.

The resultant HTTP request is not immediately sendable. In particular, the host of the destination URI is incomplete and MUST be suffixed before sending.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
class HTTPRequestSerializer(SpecificShapeSerializer):
    """Binds a serializable shape to an HTTP request.

    The resultant HTTP request is not immediately sendable. In particular, the host of
    the destination URI is incomplete and MUST be suffixed before sending.
    """

    def __init__(
        self,
        payload_codec: Codec,
        http_trait: HTTPTrait,
        endpoint_trait: EndpointTrait | None = None,
        omit_empty_payload: bool = True,
    ) -> None:
        """Initialize an HTTPRequestSerializer.

        :param payload_codec: The codec to use to serialize the HTTP payload, if one is
            present.
        :param http_trait: The HTTP trait of the operation being handled.
        :param endpoint_trait: The optional endpoint trait of the operation being
            handled.
        :param omit_empty_payload: Whether an empty payload should be omitted.
        """
        self._http_trait = http_trait
        self._endpoint_trait = endpoint_trait
        self._payload_codec = payload_codec
        self._omit_empty_payload = omit_empty_payload
        self.result: HTTPRequest | None = None

    @contextmanager
    def begin_struct(self, schema: Schema) -> Iterator[ShapeSerializer]:
        payload: AsyncBytesReader | AsyncBytesProvider
        binding_serializer: HTTPRequestBindingSerializer

        host_prefix = ""
        if self._endpoint_trait is not None:
            host_prefix = self._endpoint_trait.host_prefix

        content_type = self._payload_codec.media_type
        content_length: int | None = None
        content_length_required = False

        binding_matcher = RequestBindingMatcher(schema)
        if binding_matcher.event_stream_member is not None:
            payload = AsyncBytesProvider()
            content_type = "application/vnd.amazon.eventstream"
            binding_serializer = HTTPRequestBindingSerializer(
                SpecificShapeSerializer(),
                self._http_trait.path,
                host_prefix,
                binding_matcher,
            )
            yield binding_serializer
        elif (payload_member := binding_matcher.payload_member) is not None:
            content_length_required = RequiresLengthTrait in payload_member
            if payload_member.shape_type in (
                ShapeType.BLOB,
                ShapeType.STRING,
                ShapeType.ENUM,
            ):
                if (media_type := payload_member.get_trait(MediaTypeTrait)) is not None:
                    content_type = media_type.value
                elif payload_member.shape_type is ShapeType.BLOB:
                    content_type = "application/octet-stream"
                else:
                    content_type = "text/plain"

                payload_serializer = RawPayloadSerializer()
                binding_serializer = HTTPRequestBindingSerializer(
                    payload_serializer,
                    self._http_trait.path,
                    host_prefix,
                    binding_matcher,
                )
                yield binding_serializer
                if isinstance(payload_serializer.payload, Sized):
                    content_length = len(payload_serializer.payload)
                payload = AsyncBytesReader(payload_serializer.payload or b"")
            else:
                if (media_type := payload_member.get_trait(MediaTypeTrait)) is not None:
                    content_type = media_type.value
                sync_payload = BytesIO()
                payload_serializer = self._payload_codec.create_serializer(sync_payload)
                binding_serializer = HTTPRequestBindingSerializer(
                    payload_serializer,
                    self._http_trait.path,
                    host_prefix,
                    binding_matcher,
                )
                yield binding_serializer
                content_length = sync_payload.tell()
                sync_payload.seek(0)
                payload = AsyncBytesReader(sync_payload)
        else:
            sync_payload = BytesIO()
            payload_serializer = self._payload_codec.create_serializer(sync_payload)
            if binding_matcher.should_write_body(self._omit_empty_payload):
                with payload_serializer.begin_struct(schema) as body_serializer:
                    binding_serializer = HTTPRequestBindingSerializer(
                        body_serializer,
                        self._http_trait.path,
                        host_prefix,
                        binding_matcher,
                    )
                    yield binding_serializer
                content_length = sync_payload.tell()
            else:
                content_type = None
                content_length = None
                binding_serializer = HTTPRequestBindingSerializer(
                    payload_serializer,
                    self._http_trait.path,
                    host_prefix,
                    binding_matcher,
                )
                yield binding_serializer
            sync_payload.seek(0)
            payload = AsyncBytesReader(sync_payload)

        headers = binding_serializer.header_serializer.headers
        if content_type is not None:
            headers.append(("content-type", content_type))

        if content_length is not None:
            headers.append(("content-length", str(content_length)))

        fields = tuples_to_fields(headers)
        if content_length_required and "content-length" not in fields:
            content_length = _compute_content_length(payload)
            if content_length is None:
                raise SerializationError(
                    "This operation requires the the content length of the input "
                    "stream, but it was not provided and was unable to be computed."
                )
            fields.set_field(Field(name="content-length", values=[str(content_length)]))

        self.result = _HTTPRequest(
            method=self._http_trait.method,
            destination=URI(
                host=binding_serializer.host_prefix_serializer.host_prefix,
                path=binding_serializer.path_serializer.path,
                query=join_query_params(
                    params=binding_serializer.query_serializer.query_params,
                    prefix=self._http_trait.query or "",
                ),
            ),
            fields=fields,
            body=payload,
        )

__init__(payload_codec, http_trait, endpoint_trait=None, omit_empty_payload=True)

Initialize an HTTPRequestSerializer.

:param payload_codec: The codec to use to serialize the HTTP payload, if one is present. :param http_trait: The HTTP trait of the operation being handled. :param endpoint_trait: The optional endpoint trait of the operation being handled. :param omit_empty_payload: Whether an empty payload should be omitted.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def __init__(
    self,
    payload_codec: Codec,
    http_trait: HTTPTrait,
    endpoint_trait: EndpointTrait | None = None,
    omit_empty_payload: bool = True,
) -> None:
    """Initialize an HTTPRequestSerializer.

    :param payload_codec: The codec to use to serialize the HTTP payload, if one is
        present.
    :param http_trait: The HTTP trait of the operation being handled.
    :param endpoint_trait: The optional endpoint trait of the operation being
        handled.
    :param omit_empty_payload: Whether an empty payload should be omitted.
    """
    self._http_trait = http_trait
    self._endpoint_trait = endpoint_trait
    self._payload_codec = payload_codec
    self._omit_empty_payload = omit_empty_payload
    self.result: HTTPRequest | None = None

HTTPResponseBindingSerializer

Bases: InterceptingSerializer

Delegates HTTP response bindings to binding-location-specific serializers.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
class HTTPResponseBindingSerializer(InterceptingSerializer):
    """Delegates HTTP response bindings to binding-location-specific serializers."""

    def __init__(
        self,
        payload_serializer: ShapeSerializer,
        binding_matcher: ResponseBindingMatcher,
    ) -> None:
        """Initialize an HTTPResponseBindingSerializer.

        :param payload_serializer: The :py:class:`ShapeSerializer` to use to serialize
            the payload, if necessary.
        """
        self._payload_serializer = payload_serializer
        self.header_serializer = HTTPHeaderSerializer()
        self.response_code_serializer = HTTPResponseCodeSerializer()
        self._binding_matcher = binding_matcher

    def before(self, schema: Schema) -> ShapeSerializer:
        match self._binding_matcher.match(schema):
            case Binding.HEADER | Binding.PREFIX_HEADERS:
                return self.header_serializer
            case Binding.STATUS:
                return self.response_code_serializer
            case _:
                return self._payload_serializer

    def after(self, schema: Schema) -> None:
        pass

__init__(payload_serializer, binding_matcher)

Initialize an HTTPResponseBindingSerializer.

:param payload_serializer: The :py:class:ShapeSerializer to use to serialize the payload, if necessary.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
394
395
396
397
398
399
400
401
402
403
404
405
406
407
def __init__(
    self,
    payload_serializer: ShapeSerializer,
    binding_matcher: ResponseBindingMatcher,
) -> None:
    """Initialize an HTTPResponseBindingSerializer.

    :param payload_serializer: The :py:class:`ShapeSerializer` to use to serialize
        the payload, if necessary.
    """
    self._payload_serializer = payload_serializer
    self.header_serializer = HTTPHeaderSerializer()
    self.response_code_serializer = HTTPResponseCodeSerializer()
    self._binding_matcher = binding_matcher

HTTPResponseCodeSerializer

Bases: SpecificShapeSerializer

Binds properties of a serializable shape to the HTTP response code.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
class HTTPResponseCodeSerializer(SpecificShapeSerializer):
    """Binds properties of a serializable shape to the HTTP response code."""

    response_code: int | None
    """The bound response code, or None if one hasn't been bound."""

    def __init__(self) -> None:
        """Initialize an HTTPResponseCodeSerializer."""
        self.response_code: int | None = None

    def write_byte(self, schema: Schema, value: int) -> None:
        self.response_code = value

    def write_short(self, schema: Schema, value: int) -> None:
        self.response_code = value

    def write_integer(self, schema: Schema, value: int) -> None:
        self.response_code = value

response_code = None instance-attribute

The bound response code, or None if one hasn't been bound.

__init__()

Initialize an HTTPResponseCodeSerializer.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
770
771
772
def __init__(self) -> None:
    """Initialize an HTTPResponseCodeSerializer."""
    self.response_code: int | None = None

HTTPResponseSerializer

Bases: SpecificShapeSerializer

Binds a serializable shape to an HTTP response.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
class HTTPResponseSerializer(SpecificShapeSerializer):
    """Binds a serializable shape to an HTTP response."""

    def __init__(
        self,
        payload_codec: Codec,
        http_trait: HTTPTrait,
        omit_empty_payload: bool = True,
    ) -> None:
        """Initialize an HTTPResponseSerializer.

        :param payload_codec: The codec to use to serialize the HTTP payload, if one is
            present.
        :param http_trait: The HTTP trait of the operation being handled.
        :param omit_empty_payload: Whether an empty payload should be omitted.
        """
        self._http_trait = http_trait
        self._payload_codec = payload_codec
        self.result: HTTPResponse | None = None
        self._omit_empty_payload = omit_empty_payload

    @contextmanager
    def begin_struct(self, schema: Schema) -> Iterator[ShapeSerializer]:
        payload: AsyncBytesReader | AsyncBytesProvider
        binding_serializer: HTTPResponseBindingSerializer

        content_type: str | None = self._payload_codec.media_type
        content_length: int | None = None
        content_length_required = False

        binding_matcher = ResponseBindingMatcher(schema)
        if binding_matcher.event_stream_member is not None:
            payload = AsyncBytesProvider()
            content_type = "application/vnd.amazon.eventstream"
            binding_serializer = HTTPResponseBindingSerializer(
                SpecificShapeSerializer(), binding_matcher
            )
            yield binding_serializer
        elif (payload_member := binding_matcher.payload_member) is not None:
            content_length_required = RequiresLengthTrait in payload_member
            if payload_member.shape_type in (ShapeType.BLOB, ShapeType.STRING):
                if (media_type := payload_member.get_trait(MediaTypeTrait)) is not None:
                    content_type = media_type.value
                elif payload_member.shape_type is ShapeType.BLOB:
                    content_type = "application/octet-stream"
                else:
                    content_type = "text/plain"
                payload_serializer = RawPayloadSerializer()
                binding_serializer = HTTPResponseBindingSerializer(
                    payload_serializer, binding_matcher
                )
                yield binding_serializer
                if isinstance(payload_serializer.payload, Sized):
                    content_length = len(payload_serializer.payload)
                payload = AsyncBytesReader(payload_serializer.payload or b"")
            else:
                if (media_type := payload_member.get_trait(MediaTypeTrait)) is not None:
                    content_type = media_type.value
                sync_payload = BytesIO()
                payload_serializer = self._payload_codec.create_serializer(sync_payload)
                binding_serializer = HTTPResponseBindingSerializer(
                    payload_serializer, binding_matcher
                )
                yield binding_serializer
                content_length = sync_payload.tell()
                sync_payload.seek(0)
                payload = AsyncBytesReader(sync_payload)
        else:
            sync_payload = BytesIO()
            payload_serializer = self._payload_codec.create_serializer(sync_payload)
            if binding_matcher.should_write_body(self._omit_empty_payload):
                if binding_matcher.event_stream_member is not None:
                    content_type = "application/vnd.amazon.eventstream"
                with payload_serializer.begin_struct(schema) as body_serializer:
                    binding_serializer = HTTPResponseBindingSerializer(
                        body_serializer, binding_matcher
                    )
                    yield binding_serializer
                content_length = sync_payload.tell()
            else:
                content_type = None
                content_length = None
                binding_serializer = HTTPResponseBindingSerializer(
                    payload_serializer,
                    binding_matcher,
                )
                yield binding_serializer
            sync_payload.seek(0)
            payload = AsyncBytesReader(sync_payload)

        headers = binding_serializer.header_serializer.headers
        if content_type is not None:
            headers.append(("content-type", content_type))

        if content_length is not None:
            headers.append(("content-length", str(content_length)))

        fields = tuples_to_fields(headers)
        if content_length_required and "content-length" not in fields:
            content_length = _compute_content_length(payload)
            if content_length is None:
                raise SerializationError(
                    "This operation requires the the content length of the input "
                    "stream, but it was not provided and was unable to be computed."
                )
            fields.set_field(Field(name="content-length", values=[str(content_length)]))

        status = binding_serializer.response_code_serializer.response_code
        if status is None:
            if binding_matcher.response_status > 0:
                status = binding_matcher.response_status
            else:
                status = self._http_trait.code

        self.result = _HTTPResponse(
            fields=tuples_to_fields(binding_serializer.header_serializer.headers),
            body=payload,
            status=status,
        )

__init__(payload_codec, http_trait, omit_empty_payload=True)

Initialize an HTTPResponseSerializer.

:param payload_codec: The codec to use to serialize the HTTP payload, if one is present. :param http_trait: The HTTP trait of the operation being handled. :param omit_empty_payload: Whether an empty payload should be omitted.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def __init__(
    self,
    payload_codec: Codec,
    http_trait: HTTPTrait,
    omit_empty_payload: bool = True,
) -> None:
    """Initialize an HTTPResponseSerializer.

    :param payload_codec: The codec to use to serialize the HTTP payload, if one is
        present.
    :param http_trait: The HTTP trait of the operation being handled.
    :param omit_empty_payload: Whether an empty payload should be omitted.
    """
    self._http_trait = http_trait
    self._payload_codec = payload_codec
    self.result: HTTPResponse | None = None
    self._omit_empty_payload = omit_empty_payload

HostPrefixSerializer

Bases: SpecificShapeSerializer

Binds properites of a serializable shape to the HTTP URI host.

These properties are also bound to the payload.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
class HostPrefixSerializer(SpecificShapeSerializer):
    """Binds properites of a serializable shape to the HTTP URI host.

    These properties are also bound to the payload.
    """

    def __init__(
        self, payload_serializer: ShapeSerializer, host_prefix_pattern: str
    ) -> None:
        """Initialize a HostPrefixSerializer.

        :param host_prefix_pattern: The pattern to bind properties to.
        :param payload_serializer: The payload serializer to additionally write
            properties to.
        """
        self._prefix_params: dict[str, str] = {}
        self._host_prefix_pattern = host_prefix_pattern
        self._payload_serializer = payload_serializer

    @property
    def host_prefix(self) -> str:
        """The formatted host prefix.

        This must not be accessed before serialization is complete, otherwise an
        exception will be raised.
        """
        return self._host_prefix_pattern.format(**self._prefix_params)

    def write_string(self, schema: Schema, value: str) -> None:
        self._payload_serializer.write_string(schema, value)
        self._prefix_params[schema.expect_member_name()] = urlquote(value, safe=".")

host_prefix property

The formatted host prefix.

This must not be accessed before serialization is complete, otherwise an exception will be raised.

__init__(payload_serializer, host_prefix_pattern)

Initialize a HostPrefixSerializer.

:param host_prefix_pattern: The pattern to bind properties to. :param payload_serializer: The payload serializer to additionally write properties to.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
737
738
739
740
741
742
743
744
745
746
747
748
def __init__(
    self, payload_serializer: ShapeSerializer, host_prefix_pattern: str
) -> None:
    """Initialize a HostPrefixSerializer.

    :param host_prefix_pattern: The pattern to bind properties to.
    :param payload_serializer: The payload serializer to additionally write
        properties to.
    """
    self._prefix_params: dict[str, str] = {}
    self._host_prefix_pattern = host_prefix_pattern
    self._payload_serializer = payload_serializer

RawPayloadSerializer

Bases: SpecificShapeSerializer

Binds properties of serializable shape to an HTTP payload.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
class RawPayloadSerializer(SpecificShapeSerializer):
    """Binds properties of serializable shape to an HTTP payload."""

    payload: "AsyncStreamingBlob | None"
    """The serialized payload.

    This will only be non-null after serialization.
    """

    def __init__(self) -> None:
        """Initialize a RawPayloadSerializer."""
        self.payload: AsyncStreamingBlob | None = None

    def write_string(self, schema: Schema, value: str) -> None:
        self.payload = value.encode("utf-8")

    def write_blob(self, schema: Schema, value: bytes) -> None:
        self.payload = value

    def write_data_stream(self, schema: Schema, value: "AsyncStreamingBlob") -> None:
        self.payload = value

payload = None instance-attribute

The serialized payload.

This will only be non-null after serialization.

__init__()

Initialize a RawPayloadSerializer.

Source code in packages/smithy-http/src/smithy_http/serializers.py 
431
432
433
def __init__(self) -> None:
    """Initialize a RawPayloadSerializer."""
    self.payload: AsyncStreamingBlob | None = None

user_agent

RawStringUserAgentComponent dataclass

UserAgentComponent interface wrapper around str.

Use for User-Agent header components that are not constructed from prefix+name+value but instead are provided as strings. No sanitization is performed.

Source code in packages/smithy-http/src/smithy_http/user_agent.py 
37
38
39
40
41
42
43
44
45
46
47
48
@dataclass(frozen=True, slots=True)
class RawStringUserAgentComponent:
    """UserAgentComponent interface wrapper around ``str``.

    Use for User-Agent header components that are not constructed from prefix+name+value
    but instead are provided as strings. No sanitization is performed.
    """

    value: str

    def __str__(self) -> str:
        return self.value

UserAgentComponent dataclass

Component of a User-Agent header string in the standard format.

Each component consists of a prefix, a name, and a value. In the string representation these are combined in the format prefix/name#value.

This class is considered private and is subject to abrupt breaking changes.

Source code in packages/smithy-http/src/smithy_http/user_agent.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@dataclass(frozen=True, slots=True)
class UserAgentComponent:
    """Component of a User-Agent header string in the standard format.

    Each component consists of a prefix, a name, and a value. In the string
    representation these are combined in the format ``prefix/name#value``.

    This class is considered private and is subject to abrupt breaking changes.
    """

    prefix: str
    name: str
    value: str | None = None

    def __str__(self):
        """Create string like 'prefix/name#value' from a UserAgentComponent."""
        clean_prefix = sanitize_user_agent_string_component(
            self.prefix, allow_hash=True
        )
        clean_name = sanitize_user_agent_string_component(self.name, allow_hash=False)
        if self.value is None or self.value == "":
            return f"{clean_prefix}/{clean_name}"
        clean_value = sanitize_user_agent_string_component(self.value, allow_hash=True)
        return f"{clean_prefix}/{clean_name}#{clean_value}"

__str__()

Create string like 'prefix/name#value' from a UserAgentComponent.

Source code in packages/smithy-http/src/smithy_http/user_agent.py 
25
26
27
28
29
30
31
32
33
34
def __str__(self):
    """Create string like 'prefix/name#value' from a UserAgentComponent."""
    clean_prefix = sanitize_user_agent_string_component(
        self.prefix, allow_hash=True
    )
    clean_name = sanitize_user_agent_string_component(self.name, allow_hash=False)
    if self.value is None or self.value == "":
        return f"{clean_prefix}/{clean_name}"
    clean_value = sanitize_user_agent_string_component(self.value, allow_hash=True)
    return f"{clean_prefix}/{clean_name}#{clean_value}"

sanitize_user_agent_string_component(raw_str, allow_hash=False)

Replaces all not allowed characters in the string with a dash ("-").

Allowed characters are ASCII alphanumerics and !$%&'*+-.^_`|~. If allow_hash is True, "#"``" is also allowed.

:type raw_str: str :param raw_str: The input string to be sanitized.

:type allow_hash: bool :param allow_hash: Whether "#" is considered an allowed character.

Source code in packages/smithy-http/src/smithy_http/user_agent.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def sanitize_user_agent_string_component(raw_str: str, allow_hash: bool = False) -> str:
    """Replaces all not allowed characters in the string with a dash ("-").

    Allowed characters are ASCII alphanumerics and ``!$%&'*+-.^_`|~``. If
    ``allow_hash`` is ``True``, "#"``" is also allowed.

    :type raw_str: str
    :param raw_str: The input string to be sanitized.

    :type allow_hash: bool
    :param allow_hash: Whether "#" is considered an allowed character.
    """
    return "".join(
        c if c in _USERAGENT_ALLOWED_CHARACTERS or (allow_hash and c == "#") else "-"
        for c in raw_str
    )

utils

join_query_params(params, prefix='')

Join a list of query parameter key-value tuples.

:param params: The list of key-value query parameter tuples. :param prefix: An optional query prefix.

Source code in packages/smithy-http/src/smithy_http/utils.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def join_query_params(
    params: Sequence[tuple[str, str | None]], prefix: str = ""
) -> str:
    """Join a list of query parameter key-value tuples.

    :param params: The list of key-value query parameter tuples.
    :param prefix: An optional query prefix.
    """
    query: str = prefix
    for param in params:
        if query:
            query += "&"
        if param[1] is None:
            query += urlquote(param[0], safe="")
        else:
            query += f"{urlquote(param[0], safe='')}={urlquote(param[1], safe='')}"
    return query

split_header(given, handle_unquoted_http_date=False)

Splits a header value into a list of strings.

The format is based on RFC9110's list production found in secion 5.6.1 with the quoted string production found in section 5.6.4. In short:

A list is 1 or more elements surrounded by optional whitespace and separated by commas. Elements may be quoted with double quotes (") to contain leading or trailing whitespace, commas, or double quotes. Inside the the double quotes, a value may be escaped with a backslash (\). Elements that contain no contents are ignored.

If the list is known to contain unquoted IMF-fixdate formatted timestamps, the handle_unquoted_http_date parameter can be set to ensure the list isn't split on the commas inside the timestamps.

:param given: The header value to split. :param handle_unquoted_http_date: Support splitting IMF-fixdate lists without quotes. Defaults to False. :returns: The header value split on commas.

Source code in packages/smithy-http/src/smithy_http/utils.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def split_header(given: str, handle_unquoted_http_date: bool = False) -> list[str]:
    """Splits a header value into a list of strings.

    The format is based on RFC9110's list production found in secion 5.6.1 with
    the quoted string production found in section 5.6.4. In short:

    A list is 1 or more elements surrounded by optional whitespace and separated by
    commas. Elements may be quoted with double quotes (``"``) to contain leading or
    trailing whitespace, commas, or double quotes. Inside the the double quotes, a
    value may be escaped with a backslash (``\\``). Elements that contain no contents
    are ignored.

    If the list is known to contain unquoted IMF-fixdate formatted timestamps, the
    ``handle_unquoted_http_date`` parameter can be set to ensure the list isn't
    split on the commas inside the timestamps.

    :param given: The header value to split.
    :param handle_unquoted_http_date: Support splitting IMF-fixdate lists without
        quotes. Defaults to False.
    :returns: The header value split on commas.
    """
    result: list[str] = []

    i = 0
    while i < len(given):
        if given[i].isspace():
            # Skip any leading space.
            i += 1
        elif given[i] == '"':
            # Grab the contents of the quoted value and append it.
            entry, i = _consume_until(given, i + 1, '"', escape_char="\\")
            result.append(entry)

            if i > len(given) or given[i - 1] != '"':
                raise SmithyError(
                    f"Invalid header list syntax: expected end quote but reached end "
                    f"of value: `{given}`"
                )

            # Skip until the next comma.
            excess, i = _consume_until(given, i, ",")
            if excess.strip():
                raise SmithyError(
                    f"Invalid header list syntax: Found quote contents after "
                    f"end-quote: `{excess}` in `{given}`"
                )
        else:
            entry, i = _consume_until(
                given, i, ",", skip_first=handle_unquoted_http_date
            )
            if stripped := entry.strip():
                result.append(stripped)

    return result