connect-python

import asyncio from connectrpc.client import ConnectClient, ConnectError from connectrpc.servers.asgi import ConnectASGI from connectrpc.protocol import Code from dataclasses import dataclass # Assume greet_pb2 and greet_connect have been generated from greet.proto: # protoc -I. --python_out=. --pyi_out=. --grpc_python_out=. --connect_python_out=. greet.proto # --- Minimal Proto Definition (conceptual for quickstart) --- # syntax = "proto3"; # package greet.v1; # message GreetRequest { string name = 1; } # message GreetResponse { string greeting = 1; } # service GreetService { rpc Greet(GreetRequest) returns (GreetResponse); } # --- Mock generated classes for quickstart --- # In a real app, these would come from `import greet_pb2`, `import greet_connect` @dataclass class GreetRequest: name: str @dataclass class GreetResponse: greeting: str class GreetServiceBase: async def greet(self, request: GreetRequest) -> GreetResponse: raise NotImplementedError() class GreetServiceAsyncClient: def __init__(self, client: ConnectClient, base_url: str): self._client = client; self._base_url = base_url async def greet(self, request: GreetRequest) -> GreetResponse: response = await self._client.unary(f"{self._base_url}/greet.v1.GreetService/Greet", request, GreetResponse) return response # --- Server Implementation --- class MyGreetService(GreetServiceBase): async def greet(self, request: GreetRequest) -> GreetResponse: print(f"Server received: {request.name}") return GreetResponse(greeting=f"Hello, {request.name}!") async def run_server_and_client(): # Setup server server_app = ConnectASGI(services=[MyGreetService()]) # For this quickstart, we'll simulate a server request handler. # In a real app, you'd run this with uvicorn: `uvicorn server_module:server_app` print("\n--- Running Server (simulated) ---") async def handle_server_request(path: str, request_data: bytes): # This is a simplified simulation of ASGI handling mock_scope = {"type": "http", "method": "POST", "path": path, "headers": [(b"content-type", b"application/json")]} mock_receive = asyncio.Queue() await mock_receive.put({"type": "http.request", "body": request_data, "more_body": False}) mock_send = asyncio.Queue() await server_app(mock_scope, mock_receive.get, mock_send.put) response_events = [] while True: event = await mock_send.get() response_events.append(event) if event["type"] == "http.response.body" and not event.get("more_body", False): break status_code = next(e["status"] for e in response_events if e["type"] == "http.response.start") body = b"".join(e["body"] for e in response_events if e["type"] == "http.response.body") return status_code, body # Setup client # For this quickstart, we'll use a mocked client that talks directly to the server_app class MockConnectClient(ConnectClient): async def unary(self, path: str, request, response_class): # Serialize request as JSON for simulation import json request_data = json.dumps({"name": request.name}).encode('utf-8') status_code, response_body = await handle_server_request(path, request_data) if status_code != 200: raise ConnectError(Code(status_code), details=response_body.decode('utf-8')) # Deserialize response from JSON for simulation response_dict = json.loads(response_body.decode('utf-8')) return response_class(greeting=response_dict['greeting']) mock_connect_client = MockConnectClient() client = GreetServiceAsyncClient(mock_connect_client, "/greet.v1.GreetService") # Make a client request print("\n--- Running Client ---") try: req = GreetRequest(name="Alice") res = await client.greet(req) print(f"Client received: {res.greeting}") except ConnectError as e: print(f"Client error: {e.code} - {e.details}") if __name__ == "__main__": asyncio.run(run_server_and_client())