give context to input streams

Author: JordonPhillips

codegen/core/src/main/java/software/amazon/smithy/python/codegen/ClientGenerator.java

@@ -199,12 +199,12 @@ async def _input_stream(
                                 config: $4T,
                                 operation_name: str,
                             ) -> Any:
-                                request_future = Future[$2T]()
+                                request_future = Future[InterceptorContext[Any, Any, $2T, Any]]()
                                 awaitable_output = asyncio.create_task(self._execute_operation(
                                     input, plugins, serialize, deserialize, config, operation_name,
                                     request_future=request_future
                                 ))
-                                transport_request = await request_future
+                                request_context = await request_future
                                 ${5C|}
 
                             async def _output_stream(
@@ -235,14 +235,14 @@ async def _duplex_stream(
                                 operation_name: str,
                                 event_deserializer: Callable[[ShapeDeserializer], Any],
                             ) -> Any:
-                                request_future = Future[$2T]()
+                                request_future = Future[InterceptorContext[Any, Any, $2T, Any]]()
                                 response_future = Future[$3T]()
                                 awaitable_output = asyncio.create_task(self._execute_operation(
                                     input, plugins, serialize, deserialize, config, operation_name,
                                     request_future=request_future,
                                     response_future=response_future
                                 ))
-                                transport_request = await request_future
+                                request_context = await request_future
                                 ${7C|}
                             """,
                     pluginSymbol,
@@ -264,7 +264,7 @@ async def _execute_operation(
                             deserialize: Callable[[$3T, $5T], Awaitable[Output]],
                             config: $5T,
                             operation_name: str,
-                            request_future: Future[$2T] | None = None,
+                            request_future: Future[InterceptorContext[Any, Any, $2T, Any]] | None = None,
                             response_future: Future[$3T] | None = None,
                         ) -> Output:
                             try:
@@ -292,7 +292,7 @@ async def _handle_execution(
                             deserialize: Callable[[$3T, $5T], Awaitable[Output]],
                             config: $5T,
                             operation_name: str,
-                            request_future: Future[$2T] | None,
+                            request_future: Future[InterceptorContext[Any, Any, $2T, Any]] | None,
                             response_future: Future[$3T] | None,
                         ) -> Output:
                             logger.debug('Making request for operation "%s" with parameters: %s', operation_name, input)
@@ -409,7 +409,7 @@ await seek(0)
                                         retry_strategy.record_success(token=retry_token)
                                         if response_future is not None:
                                             response_future.set_result(
-                                                context_with_response.transport_response,  # type: ignore
+                                                context_with_response.transport_response  # type: ignore
                                             )
                                         break
                             except Exception as e:
@@ -433,7 +433,7 @@ async def _handle_attempt(
                             context: InterceptorContext[Input, None, $2T, None],
                             config: $5T,
                             operation_name: str,
-                            request_future: Future[$2T] | None,
+                            request_future: Future[InterceptorContext[Any, Any, $2T, Any]] | None,
                         ) -> InterceptorContext[Input, Output, $2T, $3T | None]:
                             try:
                                 # assert config.interceptors is not None
@@ -599,7 +599,7 @@ async def _handle_attempt(
                                     request=context_with_response.transport_request,
                                     request_config=request_config,
                                 ))
-                                request_future.set_result(context_with_response.transport_request)
+                                request_future.set_result(context_with_response)
                                 context_with_response._transport_response = await response_task
                             else:
                                 context_with_response._transport_response = await config.http_client.send(

codegen/core/src/main/java/software/amazon/smithy/python/codegen/integrations/RestJsonProtocolGenerator.java

@@ -403,7 +403,7 @@ public void wrapInputStream(GenerationContext context, PythonWriter writer) {
                         return AWSInputEventStream[Any, Any](
                             payload_codec=codec,
                             awaitable_output=awaitable_output,
-                            async_writer=transport_request.body,  # type: ignore
+                            async_writer=request_context.transport_request.body,  # type: ignore
                         )
                         """);
     }
@@ -443,7 +443,7 @@ public void wrapDuplexStream(GenerationContext context, PythonWriter writer) {
                         codec = JSONCodec(default_timestamp_format=TimestampFormat.EPOCH_SECONDS)
                         return AWSDuplexEventStream[Any, Any, Any](
                             payload_codec=codec,
-                            async_writer=transport_request.body,  # type: ignore
+                            async_writer=request_context.transport_request.body,  # type: ignore
                             awaitable_output=awaitable_output,
                             awaitable_response=response_future,
                             deserializer=event_deserializer,  # type: ignore