Avoid unnecessary copies in IdentityCompression and reuse protocol singletons

src/connectrpc/_compression.py

@@ -16,11 +16,11 @@ def name(self) -> str:
 
     def compress(self, data: bytes | bytearray | memoryview) -> bytes:
         """Return data as-is without compression."""
-        return bytes(data)
+        return data if isinstance(data, bytes) else bytes(data)
 
     def decompress(self, data: bytes | bytearray | memoryview) -> bytes:
         """Return data as-is without decompression."""
-        return bytes(data)
+        return data if isinstance(data, bytes) else bytes(data)
 
 
 _identity = IdentityCompression()

src/connectrpc/_protocol_server.py

@@ -16,13 +16,18 @@
     from ._protocol import ServerProtocol
 
 
+_CONNECT_PROTOCOL = ConnectServerProtocol()
+_GRPC_PROTOCOL = GRPCServerProtocol()
+_GRPC_WEB_PROTOCOL = GRPCWebServerProtocol()
+
+
 def negotiate_server_protocol(content_type: str) -> ServerProtocol:
     if content_type == GRPC_CONTENT_TYPE_DEFAULT or content_type.startswith(
         GRPC_CONTENT_TYPE_PREFIX
     ):
-        return GRPCServerProtocol()
+        return _GRPC_PROTOCOL
     if content_type == GRPC_WEB_CONTENT_TYPE_DEFAULT or content_type.startswith(
         GRPC_WEB_CONTENT_TYPE_PREFIX
     ):
-        return GRPCWebServerProtocol()
-    return ConnectServerProtocol()
+        return _GRPC_WEB_PROTOCOL
+    return _CONNECT_PROTOCOL

test/test_compression.py

@@ -4,6 +4,7 @@
 from pyqwest import Client, SyncClient
 from pyqwest.testing import ASGITransport, WSGITransport
 
+from connectrpc._compression import IdentityCompression
 from connectrpc.client import ResponseMetadata
 from connectrpc.compression.brotli import BrotliCompression
 from connectrpc.compression.gzip import GzipCompression
@@ -88,3 +89,27 @@ def make_hat(self, request, ctx):
     assert res.size == 10
     assert res.color == "blue"
     assert meta.headers.get("content-encoding") == encoding
+
+
+class TestIdentityCompression:
+    def test_name(self):
+        assert IdentityCompression().name() == "identity"
+
+    def test_bytes(self):
+        data = b"hello"
+        compression = IdentityCompression()
+        compressed = compression.compress(data)
+        assert compressed is data
+        decompressed = compression.decompress(compressed)
+        assert decompressed is data
+
+    @pytest.mark.parametrize("ctor", [bytearray, memoryview])
+    def test_not_bytes(self, ctor: type[bytearray | memoryview]) -> None:
+        data = ctor(b"hello")
+        compression = IdentityCompression()
+        compressed = compression.compress(data)
+        assert compressed == b"hello"
+        assert isinstance(compressed, bytes)
+        decompressed = compression.decompress(compressed)
+        assert decompressed == b"hello"
+        assert isinstance(decompressed, bytes)