### Run Eliza Chatbot Example Source: https://github.com/connectrpc/connect-python/blob/main/README.md Instructions to start the Eliza Chatbot server using Uvicorn and run the client. Ensure you are in the 'example' directory. ```bash cd example uvicorn example.eliza_service:app --port 8080 ``` ```bash python -m example.eliza_client ``` -------------------------------- ### Verify Development Environment Setup Source: https://github.com/connectrpc/connect-python/blob/main/CONTRIBUTING.md After setting up your development environment, run this command to verify that everything is working correctly. ```bash uv run poe check ``` -------------------------------- ### Install Connect-Python Runtime Source: https://github.com/connectrpc/connect-python/blob/main/README.md Install the connectrpc runtime library using pip. Alternatively, use uv or poetry for package management. ```bash pip install connectrpc ``` ```bash # Using uv uv add connectrpc ``` ```bash # Using poetry poetry add connectrpc ``` -------------------------------- ### Basic Async Server Usage Source: https://github.com/connectrpc/connect-python/blob/main/README.md Example of implementing a Connect-Python service and creating an ASGI application. This service handles 'say_hello' requests. ```python from connectrpc.request import RequestContext from your_service_pb2 import HelloRequest, HelloResponse from your_service_connect import HelloService, HelloServiceASGIApplication class MyHelloService(HelloService): async def say_hello(self, request: HelloRequest, ctx: RequestContext) -> HelloResponse: return HelloResponse(message=f"Hello, {request.name}!") # Create ASGI app app = HelloServiceASGIApplication(MyHelloService()) # Run with any ASGI server like uvicorn and hypercorn: # uvicorn server:app --port 8080 ``` -------------------------------- ### Install Dependencies with uv Source: https://github.com/connectrpc/connect-python/blob/main/DEVELOPMENT.md Install project dependencies using 'uv'. This command synchronizes the environment with the project's requirements. ```bash uv sync ``` -------------------------------- ### Basic Async Client Usage Source: https://github.com/connectrpc/connect-python/blob/main/README.md Example of creating an asynchronous Connect-Python client and making a unary RPC call. Ensure 'your_service_pb2' and 'your_service_connect' are generated. ```python from your_service_pb2 import HelloRequest, HelloResponse from your_service_connect import HelloServiceClient # Create async client async def main(): async with HelloServiceClient("https://api.example.com") as client: # Make a unary RPC call response = await client.say_hello(HelloRequest(name="World")) print(response.message) # "Hello, World!" ``` -------------------------------- ### Run Conformance Tests with Poe Source: https://github.com/connectrpc/connect-python/blob/main/DEVELOPMENT.md Execute the Connect conformance tests using 'poe test-conformance'. This requires Go to be installed and verifies compatibility with the Connect protocol. ```bash uv run poe test-conformance ``` -------------------------------- ### Enable GET Request Support for Idempotent Methods Source: https://github.com/connectrpc/connect-python/blob/main/README.md Define methods with `idempotency_level = NO_SIDE_EFFECTS` in your proto files to automatically enable GET request support on the client. ```proto service YourService { // This method will support both GET and POST requests rpc GetData(Request) returns (Response) { option idempotency_level = NO_SIDE_EFFECTS; } } ``` -------------------------------- ### Configure Connect-Python Code Generation with buf Source: https://github.com/connectrpc/connect-python/blob/main/README.md Configure buf.gen.yaml for generating Python protobuf and Connect stub code. This setup uses remote plugins for Python and Connect. ```yaml version: v2 plugins: - remote: buf.build/protocolbuffers/python out: . - remote: buf.build/protocolbuffers/pyi out: . - remote: buf.build/connectrpc/python out: . ``` -------------------------------- ### Generate Protobuf Code with Poe Source: https://github.com/connectrpc/connect-python/blob/main/DEVELOPMENT.md Generate protobuf code for examples and tests using 'poe generate'. This command is essential for working with protobuf definitions. ```bash uv run poe generate ``` -------------------------------- ### Basic Synchronous Client Usage Source: https://github.com/connectrpc/connect-python/blob/main/README.md Example of creating a synchronous Connect-Python client and making a unary RPC call. This is suitable for applications not requiring asynchronous operations. ```python from your_service_pb2 import HelloRequest from your_service_connect import HelloServiceClientSync # Create sync client def main(): with HelloServiceClientSync("https://api.example.com") as client: # Make a unary RPC call response = client.say_hello(HelloRequest(name="World")) print(response.message) # "Hello, World!" if __name__ == "__main__": main() ``` -------------------------------- ### Client Usage of GET Requests Source: https://github.com/connectrpc/connect-python/blob/main/README.md Clients can automatically use GET requests for methods marked as idempotent. ```python # The client will use GET for idempotent methods response = await client.get_data(request) ``` -------------------------------- ### Server Streaming RPC Implementation and Usage Source: https://github.com/connectrpc/connect-python/blob/main/README.md Example of a server implementing a server streaming RPC and a client consuming it. The server yields multiple responses for a single request. ```python # Server implementation async def make_hats(self, req: Size, ctx: RequestContext) -> AsyncIterator[Hat]: for i in range(3): yield Hat(size=req.inches + i, color=["red", "green", "blue"][i]) # Client usage async for hat in client.make_hats(Size(inches=12)): print(f"Received: {hat}") ``` -------------------------------- ### Client Streaming RPC Implementation and Usage Source: https://github.com/connectrpc/connect-python/blob/main/README.md Example of a server implementing a client streaming RPC and a client sending multiple requests for a single response. The client sends a stream of requests, and the server aggregates them into a single response. ```python # Server implementation async def collect_sizes(self, reqs: AsyncIterator[Size], ctx: RequestContext) -> Summary: total = 0 count = 0 async for size in reqs: total += size.inches count += 1 return Summary(total=total, average=total/count if count else 0) # Client usage async def send_sizes(): for i in range(5): yield Size(inches=i * 2) summary = await client.collect_sizes(send_sizes()) ``` -------------------------------- ### Integrate Connect-Python with ASGI CORS Middleware Source: https://github.com/connectrpc/connect-python/blob/main/README.md Connect-python integrates with any ASGI CORS middleware. This example shows integration with Starlette's CORSMiddleware. ```python from starlette.middleware.cors import CORSMiddleware from starlette.applications import Starlette app = Starlette() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["GET", "POST"], allow_headers=["*"], ) # Mount your Connect application ``` -------------------------------- ### Bidirectional Streaming RPC Implementation and Usage Source: https://github.com/connectrpc/connect-python/blob/main/README.md Example of a server and client engaging in bidirectional streaming, similar to a chatbot. Both client and server can send multiple messages independently. ```python # Server implementation (like the Eliza chatbot) async def converse(self, reqs: AsyncIterator[ConverseRequest], ctx: RequestContext) -> AsyncIterator[ConverseResponse]: async for req in reqs: # Process and respond to each message reply = process_message(req.sentence) yield ConverseResponse(sentence=reply) # Client usage async def chat(): yield ConverseRequest(sentence="Hello") yield ConverseRequest(sentence="How are you?") async for response in client.converse(chat()): print(f"Response: {response.sentence}") ``` -------------------------------- ### Clone connect-python Repository Source: https://github.com/connectrpc/connect-python/blob/main/DEVELOPMENT.md Clone the connect-python repository to your local machine. This is the first step in setting up the development environment. ```bash git clone https://github.com/connectrpc/connect-python cd connect-python ``` -------------------------------- ### Configure Connect-Python Code Generation with Local Plugin Source: https://github.com/connectrpc/connect-python/blob/main/README.md Configure code generation using a local protoc-gen-connectrpc plugin. The output directory is specified by 'out'. ```yaml - local: .venv/bin/protoc-gen-connectrpc out: . ``` -------------------------------- ### Fork and Clone connect-python Repository Source: https://github.com/connectrpc/connect-python/blob/main/CONTRIBUTING.md Follow these steps to fork and clone the connect-python repository for development. ```bash gh repo fork connectrpc/connect-python --clone cd connect-python ``` -------------------------------- ### Specify Protobuf Edition in Proto File Source: https://github.com/connectrpc/connect-python/blob/main/README.md Use the `edition` keyword argument to specify the Protobuf Edition, such as "2024", for your service definitions. ```proto edition = "2024"; package your.service; service YourService { rpc YourMethod(Request) returns (Response); } ``` -------------------------------- ### Implement WSGI Service with Connect-Python Source: https://github.com/connectrpc/connect-python/blob/main/README.md Use ConnectWSGIApplication for synchronous Python applications. Integrate with WSGI servers like Gunicorn. Supports synchronous methods and streaming responses (excluding bidirectional). ```python from connectrpc.request import RequestContext from connectrpc.server import ConnectWSGIApplication from your_service_pb2 import Request, Response from your_service_connect import YourService, YourServiceWSGIApplication class YourServiceImpl(YourService): def your_method(self, request: Request, ctx: RequestContext) -> Response: # Synchronous implementation return Response(message="Hello from WSGI") # WSGI also supports streaming (except full-duplex bidirectional) def stream_data(self, request: Request, ctx: RequestContext) -> Iterator[Response]: for i in range(3): yield Response(message=f"Message {i}") # Create WSGI application app = YourServiceWSGIApplication(YourServiceImpl()) # Run with gunicorn: gunicorn server:app ``` -------------------------------- ### Run Unit Tests with Poe Source: https://github.com/connectrpc/connect-python/blob/main/DEVELOPMENT.md Execute the project's unit tests using 'poe test'. This command verifies the functionality of individual code components. ```bash uv run poe test ``` -------------------------------- ### Server and Client Instrumentation with OpenTelemetryInterceptor Source: https://github.com/connectrpc/connect-python/blob/main/connectrpc-otel/README.md Integrate OpenTelemetryInterceptor into your connect-python application for both server and client-side tracing. For server-side, pass an instance to ElizaServiceWSGIApplication. For client-side, pass an instance with `client=True` to ElizaServiceClientSync. ```python from connectrpc_otel import OpenTelemetryInterceptor from eliza_connect import ElizaServiceWSGIApplication, ElizaServiceClientSync from ._service import MyElizaService app = ElizaServiceWSGIApplication(MyElizaService(), interceptors=[OpenTelemetryInterceptor()]) def make_request(): client = ElizaServiceClientSync("http://localhost:8080", interceptors=[OpenTelemetryInterceptor(client=True)]) resp = client.Say(SayRequest(sentence="Hello!")) print(resp) ``` -------------------------------- ### Run All Checks with Poe Source: https://github.com/connectrpc/connect-python/blob/main/DEVELOPMENT.md Execute all project checks, including linting and formatting, using the 'poe check' command. This ensures code quality and consistency. ```bash uv run poe check ``` -------------------------------- ### Format Code with Poe Source: https://github.com/connectrpc/connect-python/blob/main/DEVELOPMENT.md Format the project's code according to the defined style standards using 'poe format'. This command is typically run before committing changes. ```bash uv run poe format ``` -------------------------------- ### Bump to Minor Version Source: https://github.com/connectrpc/connect-python/blob/main/RELEASE.md Use this command for releases that include new features or breaking changes before version 1.0. ```bash uv run poe bump --version minor ``` -------------------------------- ### Access All Poe Targets Source: https://github.com/connectrpc/connect-python/blob/main/README.md View and run all available development targets defined in the `poe` configuration. This provides access to various tasks like linting, testing, and formatting. ```sh uv run poe ``` -------------------------------- ### Bump to Patch Version Source: https://github.com/connectrpc/connect-python/blob/main/RELEASE.md Use this command for releases containing only bug fixes. ```bash uv run poe bump --version patch ``` -------------------------------- ### Configure Message Size Limits for Client Source: https://github.com/connectrpc/connect-python/blob/main/README.md Set `read_max_bytes` when initializing the client to limit the maximum size of messages the client will read. ```python # Client with message size limit client = YourServiceClient( "https://api.example.com", read_max_bytes=1024 * 1024 ) ``` -------------------------------- ### Implement Server-Side Interceptor in Connect-Python Source: https://github.com/connectrpc/connect-python/blob/main/README.md Add cross-cutting concerns like authentication or logging using server-side interceptors. Implement the MetadataInterceptor protocol for on_start and on_end hooks. ```python from connectrpc.interceptor import MetadataInterceptor from connectrpc.request import RequestContext class LoggingInterceptor: """Implements the MetadataInterceptor protocol.""" async def on_start(self, ctx: RequestContext) -> None: print(f"Handling {ctx.method().name} request") async def on_end(self, token: None, ctx: RequestContext, error: Exception | None) -> None: if error: print(f"Failed {ctx.method().name}: {error}") else: print(f"Completed {ctx.method().name} request") # Add to your application app = HelloServiceASGIApplication( MyHelloService(), interceptors=[LoggingInterceptor()] ) ``` -------------------------------- ### Configure Client-Side Interceptors in Connect-Python Source: https://github.com/connectrpc/connect-python/blob/main/README.md Enhance client functionality with interceptors for request/response processing. Commonly used for authentication and retry logic. ```python client = HelloServiceClient( "https://api.example.com", interceptors=[AuthInterceptor(), RetryInterceptor()] ) ``` -------------------------------- ### Create a Feature Branch Source: https://github.com/connectrpc/connect-python/blob/main/CONTRIBUTING.md Before making changes, create a new feature branch from an up-to-date `main` branch. ```bash git checkout -b your-feature-branch ``` -------------------------------- ### Configure Message Size Limits for ASGI Application Source: https://github.com/connectrpc/connect-python/blob/main/README.md Set `read_max_bytes` when creating the ASGI application to limit the maximum size of incoming messages, protecting against resource exhaustion. ```python # ASGI application with 1MB limit app = YourServiceASGIApplication( service, read_max_bytes=1024 * 1024 # 1MB ) ``` -------------------------------- ### Sign Commits with Developer Certificate of Origin Source: https://github.com/connectrpc/connect-python/blob/main/CONTRIBUTING.md All commits must be signed off to affirm compliance with the Developer Certificate of Origin. Configure your git identity to match your GitHub account, then use the `-s` flag when committing. ```bash git commit -s -m "your commit message" ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.