smithy-lang · stobrien89 · Sep 9, 2026
@@ -1,8 +1,17 @@
 #  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 #  SPDX-License-Identifier: Apache-2.0
-from collections.abc import Mapping, Sequence
+from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass, field, replace
-from typing import TYPE_CHECKING, Any, NotRequired, Required, Self, TypedDict, overload
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    NotRequired,
+    Required,
+    Self,
+    TypedDict,
+    cast,
+    overload,
+)
 
 from .exceptions import ExpectationNotMetError, SmithyError
 from .shapes import ShapeID, ShapeType
@@ -95,6 +104,8 @@ def __init__(
         if member_index is not None:
             object.__setattr__(self, "member_index", member_index)
 
+        object.__setattr__(self, "_extensions", None)
+
     @property
     def member_name(self) -> str | None:
         """The name of the member, if the shape is the MEMBER type."""
@@ -174,6 +185,33 @@ def expect_trait(self, t: "type[Trait] | ShapeID") -> "Trait | DynamicTrait":
         id = t if isinstance(t, ShapeID) else t.id
         return self.traits[id]
 
+    def get_extension[T](self, extension: "SchemaExtension[T]") -> T:
+        """Get or lazily build metadata associated with this schema.
+
+        Extension descriptors are intended to be shared across all codec and protocol
+        instances. Values are cached per schema after construction. Concurrent cache
+        misses may construct the same value more than once, but subsequent calls return
+        the published cached value.
+
+        :param extension: The shared extension descriptor.
+        :returns: The cached extension value for this schema.
+        """
+        extensions = cast(
+            "dict[object, Any] | None",
+            getattr(self, "_extensions", None),
+        )
+        if extensions is None:
+            value = extension.provider(self)
+            object.__setattr__(self, "_extensions", {extension: value})
+            return value
+
+        try:
+            return extensions[extension]
+        except KeyError:
+            value = extension.provider(self)
+            extensions[extension] = value
+            return value
+
     def __contains__(self, item: Any):
         """Returns whether the schema has the given member or trait."""
         match item:
@@ -271,6 +309,14 @@ def member(
         )
 
 
+@dataclass(frozen=True, slots=True, eq=False)
+class SchemaExtension[T]:
+    """A shared provider of lazily cached schema metadata."""
+
+    provider: Callable[[Schema], T]
+    """Build the extension value for a schema."""
+
+
 class MemberSchema(TypedDict):
     """A simplified schema for members.
 

@@ -1,11 +1,11 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
-from dataclasses import replace
+from dataclasses import asdict, fields, replace
 from typing import Any
 
 import pytest
 from smithy_core.exceptions import ExpectationNotMetError
-from smithy_core.schemas import Schema
+from smithy_core.schemas import Schema, SchemaExtension
 from smithy_core.shapes import ShapeID, ShapeType
 from smithy_core.traits import (
     DynamicTrait,
@@ -47,6 +47,37 @@ def test_get_unknown_trait_by_id():
     assert schema.get_trait(SensitiveTrait.id) is None
 
 
+def test_schema_extension_is_built_once_and_cached() -> None:
+    calls = 0
+
+    def build_extension(schema: Schema) -> tuple[ShapeID, int]:
+        nonlocal calls
+        calls += 1
+        return schema.id, calls
+
+    extension = SchemaExtension(build_extension)
+    schema = Schema(id=ID, shape_type=ShapeType.STRUCTURE)
+
+    first = schema.get_extension(extension)
+    second = schema.get_extension(extension)
+
+    assert first == (ID, 1)
+    assert second is first
+    assert calls == 1
+
+
+def test_schema_extension_cache_is_not_dataclass_state() -> None:
+    extension = SchemaExtension(lambda schema: schema)
+    schema = Schema(id=ID, shape_type=ShapeType.STRUCTURE)
+
+    assert schema.get_extension(extension) is schema
+    assert "_extensions" not in {schema_field.name for schema_field in fields(schema)}
+    assert "_extensions" not in asdict(schema)
+
+    replaced = replace(schema)
+    assert replaced.get_extension(extension) is replaced
+
+
 def test_members_list():
     member_name = "baz"
     member = Schema(

@@ -22,12 +22,12 @@
 from smithy_core.interfaces import StreamingBlob as SyncStreamingBlob
 from smithy_core.prelude import DOCUMENT
 from smithy_core.schemas import APIOperation
-from smithy_core.serializers import SerializeableShape
+from smithy_core.serializers import SerializeableShape, SerializeableStruct
 from smithy_core.shapes import ShapeID
 from smithy_core.traits import EndpointTrait, HTTPTrait
 
 from ..deserializers import HTTPResponseDeserializer
-from ..serializers import HTTPRequestSerializer
+from ..serializers import HTTPBindingSerializer, HTTPRequestSerializer
 from .interfaces import HTTPErrorIdentifier, HTTPRequest, HTTPResponse
 
 
@@ -98,22 +98,31 @@ def serialize_request[
         endpoint: URI,
         context: TypedProperties,
     ) -> HTTPRequest:
-        # TODO(optimization): request binding cache like done in SJ
+        if isinstance(input, SerializeableStruct):
+            serializer = HTTPBindingSerializer(
+                payload_codec=self.payload_codec,
+                schema=operation.input_schema,
+                http_trait=operation.schema.expect_trait(HTTPTrait),
+                endpoint_trait=operation.schema.get_trait(EndpointTrait),
+            )
+            try:
+                input.serialize_members(serializer)
+            except BaseException as error:
+                serializer.abort(type(error), error, error.__traceback__)
+                raise
+            return serializer.build_request()
+
         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:
+        input.serialize(serializer)
+        if serializer.result is None:
             raise ExpectationNotMetError(
                 "Expected request to be serialized, but was None"
             )
-
-        return request
+        return serializer.result
 
     async def deserialize_response[
         OperationInput: "SerializeableShape",
@@ -142,7 +151,6 @@ async def deserialize_response[
         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),

@@ -16,8 +16,6 @@
 from smithy_core.schemas import Schema
 from smithy_core.shapes import ShapeType
 from smithy_core.traits import (
-    HTTPHeaderTrait,
-    HTTPPrefixHeadersTrait,
     HTTPTrait,
     MediaTypeTrait,
     TimestampFormatTrait,
@@ -26,8 +24,9 @@
 from smithy_core.utils import ensure_utc, strict_parse_bool, strict_parse_float
 
 from .aio.interfaces import HTTPResponse
-from .bindings import Binding, ResponseBindingMatcher
+from .bindings import Binding
 from .interfaces import Field, Fields
+from .schema_extensions import HTTP_BINDING_SCHEMA_EXTENSION
 from .utils import split_header
 
 if TYPE_CHECKING:
@@ -39,7 +38,7 @@
 
 
 class HTTPResponseDeserializer(SpecificShapeDeserializer):
-    """Binds :py:class:`HTTPResponse` properties to a DeserializableShape."""
+    """Deserialize HTTP response bindings through the shape deserializer contract."""
 
     # Note: caller will have to read the body if it's async and not streaming
     def __init__(
@@ -66,40 +65,43 @@ def __init__(
     def read_struct(
         self, schema: Schema, consumer: Callable[[Schema, ShapeDeserializer], None]
     ) -> None:
-        binding_matcher = ResponseBindingMatcher(schema)
+        binding_metadata = schema.get_extension(HTTP_BINDING_SCHEMA_EXTENSION)
 
-        for member in schema.members.values():
-            match binding_matcher.match(member):
+        for member, binding, name, is_list in binding_metadata.response_bound_members:
+            match binding:
                 case Binding.HEADER:
-                    trait = member.expect_trait(HTTPHeaderTrait)
-                    header = self._response.fields.entries.get(trait.key.lower())
+                    assert name is not None  # noqa: S101
+                    header = self._response.fields.entries.get(name)
                     if header is not None:
-                        if member.shape_type is ShapeType.LIST:
+                        if is_list:
                             consumer(member, HTTPHeaderListDeserializer(header))
                         else:
-                            consumer(member, HTTPHeaderDeserializer(header.as_string()))
+                            consumer(
+                                member,
+                                HTTPHeaderDeserializer(header.as_string()),
+                            )
                 case Binding.PREFIX_HEADERS:
-                    trait = member.expect_trait(HTTPPrefixHeadersTrait)
+                    assert name is not None  # noqa: S101
                     consumer(
                         member,
-                        HTTPHeaderMapDeserializer(self._response.fields, trait.prefix),
+                        HTTPHeaderMapDeserializer(self._response.fields, name),
                     )
                 case Binding.STATUS:
                     consumer(
-                        member, HTTPResponseCodeDeserializer(self._response.status)
+                        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)
+                    if (
+                        binding_metadata.event_stream_member is None
+                        and self._should_read_payload(member)
+                    ):
+                        deserializer = self._create_payload_deserializer(member)
+                        consumer(member, deserializer)
                 case _:
                     pass
 
-        if binding_matcher.has_body and not self._has_empty_body(
+        if binding_metadata.has_response_body and not self._has_empty_body(
             self._response, self._body
         ):
             deserializer = self._create_body_deserializer()

@@ -0,0 +1,96 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from dataclasses import dataclass
+
+from smithy_core.schemas import Schema, SchemaExtension
+from smithy_core.shapes import ShapeType
+from smithy_core.traits import HTTPHeaderTrait, HTTPPrefixHeadersTrait
+
+from .bindings import Binding, RequestBindingMatcher, ResponseBindingMatcher
+
+type ResponseBoundMember = tuple[Schema, Binding, str | None, bool]
+
+
+@dataclass(frozen=True, slots=True)
+class HTTPBindingSchemaMetadata:
+    """Precomputed request and response HTTP binding metadata for a schema."""
+
+    request_bindings: tuple[Binding, ...]
+    """Request binding route for each member index."""
+
+    response_bindings: tuple[Binding, ...]
+    """Response binding route for each member index."""
+
+    has_request_body: bool
+    """Whether the request structure contains document-body members."""
+
+    has_response_body: bool
+    """Whether the response structure contains document-body members."""
+
+    payload_member: Schema | None
+    """Member bound to the complete HTTP payload, if present."""
+
+    event_stream_member: Schema | None
+    """Member bound to an event stream, if present."""
+
+    response_bound_members: tuple[ResponseBoundMember, ...]
+    """Non-body response bindings in schema-member order."""
+
+    response_status: int
+    """Default response status derived from the structure traits."""
+
+    def should_write_request_body(self, omit_empty_payload: bool) -> bool:
+        """Return whether a request document body should be opened."""
+        return self.has_request_body or (
+            not omit_empty_payload and self.payload_member is None
+        )
+
+    def should_write_response_body(self, omit_empty_payload: bool) -> bool:
+        """Return whether a response document body should be opened."""
+        return self.has_response_body or (
+            not omit_empty_payload and self.payload_member is None
+        )
+
+
+def _build_http_binding_schema_metadata(schema: Schema) -> HTTPBindingSchemaMetadata:
+    request_matcher = RequestBindingMatcher(schema)
+    response_matcher = ResponseBindingMatcher(schema)
+    members = tuple(schema.members.values())
+    request_bindings = tuple(request_matcher.bindings)
+    response_bindings = tuple(response_matcher.bindings)
+
+    response_bound_members: list[ResponseBoundMember] = []
+    for member, binding in zip(members, response_bindings, strict=True):
+        match binding:
+            case Binding.HEADER:
+                trait = member.expect_trait(HTTPHeaderTrait)
+                response_bound_members.append(
+                    (
+                        member,
+                        binding,
+                        trait.key.lower(),
+                        member.shape_type is ShapeType.LIST,
+                    )
+                )
+            case Binding.PREFIX_HEADERS:
+                trait = member.expect_trait(HTTPPrefixHeadersTrait)
+                response_bound_members.append((member, binding, trait.prefix, False))
+            case Binding.STATUS | Binding.PAYLOAD:
+                response_bound_members.append((member, binding, None, False))
+            case _:
+                pass
+
+    return HTTPBindingSchemaMetadata(
+        request_bindings=request_bindings,
+        response_bindings=response_bindings,
+        has_request_body=request_matcher.has_body,
+        has_response_body=response_matcher.has_body,
+        payload_member=request_matcher.payload_member,
+        event_stream_member=request_matcher.event_stream_member,
+        response_bound_members=tuple(response_bound_members),
+        response_status=response_matcher.response_status,
+    )
+
+
+HTTP_BINDING_SCHEMA_EXTENSION = SchemaExtension(_build_http_binding_schema_metadata)
+"""Shared HTTP binding extension used by every HTTP protocol instance."""