Fix codec type

Signed-off-by: Anuraag Agrawal <anuraaga@gmail.com>

Author: anuraaga

.gitignore

@@ -110,15 +110,7 @@ venv.bak/
 *.so
 *.dylib
 
-# Test binary, build with `go test -c`
-*.test
-
-# Output of the go coverage tool, specifically when used with LiteIDE
-*.out
-
 .vscode
 
-out/
-
 # GitHub Actions temporary files
 .github/.tmp/

conformance/test/client.py

@@ -475,6 +475,7 @@ def send_unary_request_sync(
                         if test_request.cancel:
                             match test_request.cancel.cancel_timing:
                                 case Oneof("after_close_send_ms", after_close_send_ms):
+                                    await request_closed.wait()
                                     await asyncio.sleep(after_close_send_ms / 1000.0)
                                     task.cancel()
                         await task

src/connectrpc/_client_async.py

@@ -333,8 +333,7 @@ async def _send_request_unary(
                         f"message is larger than configured max {self._read_max_bytes}",
                     )
 
-                response = ctx.method.output()
-                return self._codec.decode(resp.content, response)
+                return self._codec.decode(resp.content, ctx.method.output)
             raise ConnectWireError.from_response(resp).to_exception()
         except (TimeoutError, asyncio.TimeoutError) as e:
             raise ConnectError(Code.DEADLINE_EXCEEDED, "Request timed out") from e

src/connectrpc/_client_sync.py

@@ -330,8 +330,7 @@ def _send_request_unary(self, request: REQ, ctx: RequestContext[REQ, RES]) -> RE
                         f"message is larger than configured max {self._read_max_bytes}",
                     )
 
-                response = ctx.method.output()
-                return self._codec.decode(resp.content, response)
+                return self._codec.decode(resp.content, ctx.method.output)
             raise ConnectWireError.from_response(resp).to_exception()
         except TimeoutError as e:
             raise ConnectError(Code.DEADLINE_EXCEEDED, "Request timed out") from e

src/connectrpc/_codec.py

@@ -57,7 +57,7 @@ def encode(self, message: T_contra) -> bytes:
         """Marshals the given message."""
         ...
 
-    def decode(self, data: bytes | bytearray, message: U) -> U:
+    def decode(self, data: bytes | bytearray, message_class: type[U]) -> U:
         """Unmarshals the given message."""
         ...
 
@@ -71,8 +71,8 @@ def name(self) -> str:
     def encode(self, message: Message) -> bytes:
         return message.to_binary()
 
-    def decode(self, data: bytes | bytearray, message: V) -> V:
-        return message.__class__.from_binary(data)  # TODO: fix type
+    def decode(self, data: bytes | bytearray, message_class: type[V]) -> V:
+        return message_class.from_binary(data)
 
 
 class ProtoJSONCodec(Codec[Message, V]):
@@ -88,8 +88,8 @@ def name(self) -> str:
     def encode(self, message: Message) -> bytes:
         return message.to_json(registry=self._registry).encode()
 
-    def decode(self, data: bytes | bytearray, message: V) -> V:
-        return message.__class__.from_json(data, registry=self._registry)
+    def decode(self, data: bytes | bytearray, message_class: type[V]) -> V:
+        return message_class.from_json(data, registry=self._registry)
 
 
 _proto_binary_codec = ProtoBinaryCodec()

src/connectrpc/_envelope.py

@@ -75,8 +75,7 @@ def _read_messages(self) -> Iterator[_RES]:
                 if self.handle_end_message(prefix_byte, message_data):
                     return
 
-                res = self._message_class()
-                res = self._codec.decode(message_data, res)
+                res = self._codec.decode(message_data, self._message_class)
                 yield res
 
             if len(self._buffer) < 5:

src/connectrpc/_server_async.py

@@ -330,7 +330,7 @@ async def _read_get_request(
             message = compression.decompress(message)
 
         # Get the appropriate decoder for the endpoint
-        return codec.decode(message, endpoint.method.input())
+        return codec.decode(message, endpoint.method.input)
 
     async def _read_post_request(
         self,
@@ -363,7 +363,7 @@ async def _read_post_request(
                 f"message is larger than configured max {self._read_max_bytes}",
             )
 
-        return codec.decode(req_body, endpoint.method.input())
+        return codec.decode(req_body, endpoint.method.input)
 
     async def _handle_stream(
         self,

src/connectrpc/_server_sync.py

@@ -336,7 +336,7 @@ def _handle_post_request(
                 )
 
             try:
-                return codec.decode(req_body, endpoint.method.input()), codec
+                return codec.decode(req_body, endpoint.method.input), codec
             except Exception as e:
                 raise ConnectError(
                     Code.INVALID_ARGUMENT, f"Failed to decode request body: {e!s}"
@@ -396,7 +396,7 @@ def _handle_get_request(
             # Handle GET request with proto decoder
             try:
                 # TODO - Use content type from queryparam
-                request = codec.decode(message, endpoint.method.input())
+                request = codec.decode(message, endpoint.method.input)
                 return request, codec
             except Exception as e:
                 raise ConnectError(

src/connectrpc/compat/_codec.py

@@ -26,9 +26,8 @@ def name(self) -> str:
     def encode(self, message: Message) -> bytes:
         return message.SerializeToString()
 
-    def decode(self, data: bytes | bytearray, message: V) -> V:
-        message.ParseFromString(data)  # ty:ignore[invalid-argument-type] type is incorrect
-        return message
+    def decode(self, data: bytes | bytearray, message_class: type[V]) -> V:
+        return message_class.FromString(data)  # ty:ignore[invalid-argument-type] type is incorrect
 
 
 class ProtoJSONCodec(Codec[Message, V]):
@@ -43,7 +42,8 @@ def name(self) -> str:
     def encode(self, message: Message) -> bytes:
         return MessageToJson(message).encode()
 
-    def decode(self, data: bytes | bytearray, message: V) -> V:
+    def decode(self, data: bytes | bytearray, message_class: type[V]) -> V:
+        message = message_class()
         MessageFromJson(data, message)  # ty:ignore[invalid-argument-type] type is incorrect
         return message
 

test/test_codec.py

@@ -40,8 +40,9 @@ def encode(self, message: Message) -> bytes:
             case _:
                 raise ValueError(f"unexpected message type: {type(message)}")
 
-    def decode(self, data: bytes | bytearray, message: Message) -> Message:
+    def decode(self, data: bytes | bytearray, message_class: type[Message]) -> Message:
         s = data.decode()
+        message = message_class()
         match message:
             case Size():
                 message.inches = int(s)