Add client protocol interface

Author: JordonPhillips

packages/smithy-core/src/smithy_core/aio/interfaces/__init__.py

@@ -1,12 +1,19 @@
 #  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 #  SPDX-License-Identifier: Apache-2.0
 from collections.abc import AsyncIterable
-from typing import Protocol, runtime_checkable
+from typing import Protocol, runtime_checkable, TYPE_CHECKING, Any
 
-from ...interfaces import URI
+from ...interfaces import URI, Endpoint
 from ...interfaces import StreamingBlob as SyncStreamingBlob
 
 
+if TYPE_CHECKING:
+    from ...schemas import APIOperation
+    from ...shapes import ShapeID
+    from ...serializers import SerializeableShape
+    from ...deserializers import DeserializeableShape
+
+
 @runtime_checkable
 class AsyncByteStream(Protocol):
     """A file-like object with an async read method."""
@@ -67,3 +74,67 @@ class ClientTransport[I: Request, O: Response](Protocol):
     async def send(self, request: I) -> O:
         """Send a request over the transport and receive the response."""
         ...
+
+
+class ClientProtocol[I: Request, O: Response](Protocol):
+    """A protocol used by a client to communicate with a server."""
+
+    @property
+    def id(self) -> "ShapeID":
+        """The ID of the protocol."""
+        ...
+
+    def serialize_request[
+        OperationInput: "SerializeableShape",
+        OperationOutput: "DeserializeableShape",
+    ](
+        self,
+        *,
+        operation: APIOperation[OperationInput, OperationOutput],
+        input: OperationInput,
+        endpoint: URI,
+        context: dict[str, Any],
+    ) -> I:
+        """Serialize an operation input into a transport request.
+
+        :param operation: The operation whose request is being serialized.
+        :param input: The input shape to be serialized.
+        :param endpoint: The base endpoint to serialize.
+        :param context: A context bag for the request.
+        """
+        ...
+
+    def set_service_endpoint(
+        self,
+        *,
+        request: I,
+        endpoint: Endpoint,
+    ) -> I:
+        """Update the endpoint of a transport request.
+
+        :param request: The request whose endpoint should be updated.
+        :param endpoint: The endpoint to set on the request.
+        """
+        ...
+
+    async def deserialize_response[
+        OperationInput: "SerializeableShape",
+        OperationOutput: "DeserializeableShape",
+    ](
+        self,
+        *,
+        operation: APIOperation[OperationInput, OperationOutput],
+        request: I,
+        response: O,
+        error_registry: Any,  # TODO: add error registry
+        context: dict[str, Any],  # TODO: replace with a typed context bag
+    ) -> OperationOutput:
+        """Deserializes the output from the tranport response or throws an exception.
+
+        :param operation: The operation whose response is being deserialized.
+        :param request: The transport request that was sent for this response.
+        :param response: The response to deserialize.
+        :param error_registry: A TypeRegistry used to deserialize errors.
+        :param context: A context bag for the request.
+        """
+        ...

packages/smithy-core/src/smithy_core/interfaces/__init__.py

@@ -83,3 +83,19 @@ def is_streaming_blob(obj: Any) -> TypeGuard[StreamingBlob]:
     :param obj: The object to inspect.
     """
     return isinstance(obj, bytes | bytearray) or is_bytes_reader(obj)
+
+
+# TODO: update HTTP package and existing endpoint implementations to use this.
+class Endpoint(Protocol):
+    """A resolved endpoint."""
+
+    uri: URI
+    """The endpoint URI."""
+
+    # TODO: replace this with a typed context bag
+    properties: dict[str, Any]
+    """Properties required to interact with the endpoint.
+
+    For example, in some AWS use cases this might contain HTTP headers to add to each
+    request.
+    """

packages/smithy-core/src/smithy_core/schemas.py

@@ -1,12 +1,17 @@
 #  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 #  SPDX-License-Identifier: Apache-2.0
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field, replace
-from typing import NotRequired, Required, Self, TypedDict, overload, Any
+from typing import NotRequired, Required, Self, TypedDict, overload, Any, TYPE_CHECKING
 
 from .exceptions import ExpectationNotMetException, SmithyException
 from .shapes import ShapeID, ShapeType
-from .traits import Trait, DynamicTrait
+from .traits import Trait, DynamicTrait, IdempotencyTokenTrait, StreamingTrait
+
+
+if TYPE_CHECKING:
+    from .serializers import SerializeableShape
+    from .deserializers import DeserializeableShape
 
 
 @dataclass(kw_only=True, frozen=True, init=False)
@@ -263,3 +268,54 @@ class MemberSchema(TypedDict):
     target: Required[Schema]
     index: Required[int]
     traits: NotRequired[list["Trait | DynamicTrait"]]
+
+
+@dataclass(kw_only=True, frozen=True)
+class APIOperation[I: "SerializeableShape", O: "DeserializeableShape"]:
+    """A modeled Smithy operation."""
+
+    input: type[I]
+    """The input type of the operation."""
+
+    output: type[O]
+    """The output type of the operation."""
+
+    schema: Schema
+    """The schema of the operation."""
+
+    input_schema: Schema
+    """The schema of the operation's input shape."""
+
+    output_schema: Schema
+    """The schema of the operation's output shape."""
+
+    # TODO: Add a type registry for errors
+    error_registry: Any
+    """A TypeRegistry used to create errors."""
+
+    effective_auth_schemes: Sequence[ShapeID]
+    """A list of effective auth schemes for the operation."""
+
+    @property
+    def idempotency_token_member(self) -> Schema | None:
+        """The input schema member that serves as the idempotency token."""
+        for member in self.input_schema.members.values():
+            if member.get_trait(IdempotencyTokenTrait) is not None:
+                return member
+        return None
+
+    @property
+    def input_stream_member(self) -> Schema | None:
+        """The input schema member that contains an event stream or data stream."""
+        for member in self.input_schema.members.values():
+            if member.get_trait(StreamingTrait) is not None:
+                return member
+        return None
+
+    @property
+    def output_stream_member(self) -> Schema | None:
+        """The output schema member that contains an event stream or data stream."""
+        for member in self.output_schema.members.values():
+            if member.get_trait(StreamingTrait) is not None:
+                return member
+        return None

packages/smithy-core/src/smithy_core/traits.py

@@ -195,6 +195,12 @@ def value(self) -> str:
         return self.document_value  # type: ignore
 
 
+@dataclass(init=False, frozen=True)
+class IdempotencyTokenTrait(Trait, id=ShapeID("smithy.api#IdempotencyToken")):
+    def __post_init__(self):
+        assert self.document_value is None
+
+
 # TODO: Get all this moved over to the http package
 @dataclass(init=False, frozen=True)
 class HTTPTrait(Trait, id=ShapeID("smithy.api#http")):