Add codec interface

python-packages/smithy-core/smithy_core/aio/types.py

@@ -5,7 +5,7 @@
 from io import BytesIO
 from typing import Self, cast
 
-from ..interfaces import ByteStream
+from ..interfaces import BytesReader
 from .interfaces import AsyncByteStream, StreamingBlob
 
 # The default chunk size for iterating streams.
@@ -18,7 +18,7 @@ class AsyncBytesReader:
     """A file-like object with an async read method."""
 
     # BytesIO *is* a ByteStream, but mypy will nevertheless complain if it isn't here.
-    _data: ByteStream | AsyncByteStream | AsyncIterable[bytes] | BytesIO | None
+    _data: BytesReader | AsyncByteStream | AsyncIterable[bytes] | BytesIO | None
     _closed = False
 
     def __init__(self, data: StreamingBlob):
@@ -44,7 +44,7 @@ async def read(self, size: int = -1) -> bytes:
         if self._closed or not self._data:
             raise ValueError("I/O operation on closed file.")
 
-        if isinstance(self._data, ByteStream) and not iscoroutinefunction(
+        if isinstance(self._data, BytesReader) and not iscoroutinefunction(
             self._data.read
         ):
             # Python's runtime_checkable can't actually tell the difference between

python-packages/smithy-core/smithy_core/aio/utils.py

@@ -5,7 +5,7 @@
 from typing import TypeVar
 
 from ..exceptions import AsyncBodyException
-from ..interfaces import ByteStream
+from ..interfaces import BytesReader
 from ..interfaces import StreamingBlob as SyncStreamingBlob
 from .interfaces import AsyncByteStream, StreamingBlob
 
@@ -47,7 +47,7 @@ def read_streaming_blob(body: StreamingBlob) -> bytes:
             return body
         case bytearray():
             return bytes(body)
-        case ByteStream():
+        case BytesReader():
             return body.read()
         case _:
             raise AsyncBodyException(

python-packages/smithy-core/smithy_core/codecs.py

@@ -0,0 +1,65 @@
+from io import BytesIO
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+from .interfaces import BytesReader, BytesWriter
+
+if TYPE_CHECKING:
+    from .deserializers import DeserializeableShape, ShapeDeserializer
+    from .serializers import SerializeableShape, ShapeSerializer
+
+
+@runtime_checkable
+class Codec(Protocol):
+    """A protocol for Smithy codecs.
+
+    Smithy codecs are responsible for serializing and deserializing shapes in a
+    particular format.
+    """
+
+    @property
+    def media_type(self) -> str:
+        """The media type that the codec supports."""
+        ...
+
+    def create_serializer(self, sink: BytesWriter) -> "ShapeSerializer":
+        """Create a serializer that writes to the given bytes writer.
+
+        :param sink: The output class to write to.
+        :returns: A serializer that will write to the given output.
+        """
+        ...
+
+    def create_deserializer(self, source: BytesReader) -> "ShapeDeserializer":
+        """Create a deserializer that reads from the given bytes reader.
+
+        :param source: The source to read bytes from.
+        :returns: A deserializer that reads from the given source.
+        """
+        ...
+
+    def serialize(self, shape: "SerializeableShape") -> bytes:
+        """Serialize a shape to bytes.
+
+        :param shape: The shape to serialize.
+        :returns: Bytes representing the shape serialized in the codec's media type.
+        """
+        stream = BytesIO()
+        serializer = self.create_serializer(sink=stream)
+        shape.serialize(serializer=serializer)
+        serializer.flush()
+        stream.seek(0)
+        return stream.read()
+
+    def deserialize[
+        S: DeserializeableShape
+    ](self, source: bytes | BytesReader, shape: type[S]) -> S:
+        """Deserialize bytes into a shape.
+
+        :param source: The bytes to deserialize.
+        :param shape: The shape class to deserialize into.
+        :returns: An instance of the given shape class with the data from the source.
+        """
+        if isinstance(source, bytes):
+            source = BytesIO(source)
+        deserializer = self.create_deserializer(source=source)
+        return shape.deserialize(deserializer=deserializer)

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

@@ -45,12 +45,19 @@ def netloc(self) -> str:
 
 
 @runtime_checkable
-class ByteStream(Protocol):
-    """A file-like object with a read method that returns bytes."""
+class BytesWriter(Protocol):
+    """A protocol for objects that support writing bytes to them."""
 
-    def read(self, size: int = -1) -> bytes: ...
+    def write(self, b: bytes, /) -> int: ...
+
+
+@runtime_checkable
+class BytesReader(Protocol):
+    """A protocol for objects that support reading bytes from them."""
+
+    def read(self, size: int = -1, /) -> bytes: ...
 
 
 # A union of all acceptable streaming blob types. Deserialized payloads will
 # always return a ByteStream, or AsyncByteStream if async is enabled.
-type StreamingBlob = ByteStream | bytes | bytearray
+type StreamingBlob = BytesReader | bytes | bytearray