connect-python

from connectrpc.client import ConnectClient

from connectrpc.servers.asgi import ConnectASGI

from connectrpc.client import ConnectError

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())