Connect RPC for Python (Legacy package name)

0.9.0 · deprecated · verified Tue Apr 14

Connect-Python provides a server and client runtime library for the Connect RPC protocol in Python, supporting all three RPC patterns (unary, server streaming, client streaming, bidirectional streaming) and interoperability with gRPC. Version 0.9.0 is the final release under this package name; future development and updates have transitioned to the 'connectrpc' package. The library is actively maintained under its new name with a focus on stability and feature parity.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates a basic unary (request-response) RPC using Connect-Python. It conceptually shows a `GreetService` with a `Greet` method. In a real application, you would first define your service in a `.proto` file (e.g., `greet.proto`), then generate Python code using `protoc` with the Connect-Python plugin. The example simulates a server responding to a client request. For production, the server (`server_app`) would be run with an ASGI server like Uvicorn.

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

view raw JSON →