Call read_data_stream on data streams

Author: JordonPhillips

codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/MemberDeserializerGenerator.java

@@ -28,6 +28,7 @@
 import software.amazon.smithy.model.shapes.StructureShape;
 import software.amazon.smithy.model.shapes.TimestampShape;
 import software.amazon.smithy.model.shapes.UnionShape;
+import software.amazon.smithy.model.traits.StreamingTrait;
 import software.amazon.smithy.python.codegen.GenerationContext;
 import software.amazon.smithy.python.codegen.SymbolProperties;
 import software.amazon.smithy.python.codegen.writer.PythonWriter;
@@ -100,7 +101,11 @@ private void writeSchema() {
 
     @Override
     public Void blobShape(BlobShape shape) {
-        writeDeserializer(shape);
+        if (shape.hasTrait(StreamingTrait.class)) {
+            writeDeserializer("data_stream");
+        } else {
+            writeDeserializer(shape);
+        }
         return null;
     }
 

packages/smithy-http/src/smithy_http/deserializers.py

@@ -4,8 +4,11 @@
 from base64 import b64decode
 from collections.abc import Callable
 from decimal import Decimal
-from typing import TYPE_CHECKING
+from inspect import iscoroutinefunction
+from typing import TYPE_CHECKING, Any, TypeGuard
 
+from smithy_core.aio.interfaces import AsyncByteStream
+from smithy_core.aio.types import AsyncBytesReader
 from smithy_core.codecs import Codec
 from smithy_core.deserializers import ShapeDeserializer, SpecificShapeDeserializer
 from smithy_core.exceptions import UnsupportedStreamError
@@ -295,7 +298,14 @@ def read_blob(self, schema: Schema) -> bytes:
         return self._consume_payload()
 
     def read_data_stream(self, schema: Schema) -> "AsyncStreamingBlob":
-        return self._payload
+        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):