### Setup Virtual Environment and Install Dependencies Source: https://github.com/danielgtaylor/python-betterproto/blob/master/README.md Installs dependencies and enables the compiler feature within a poetry-managed virtual environment. ```shell poetry install -E compiler ``` -------------------------------- ### Install for Development Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/OVERVIEW.md Set up the project for development, including the compiler, and install dependencies using Poetry. ```bash pip install -e ".[compiler]" cd python-betterproto && poetry install -E compiler ``` -------------------------------- ### Install Betterproto and Compiler Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/OVERVIEW.md Use pip to install the library and the protocol buffer compiler. ```bash pip install "betterproto[compiler]" ``` -------------------------------- ### Install and Compile Protobuf Files Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/OVERVIEW.md Install the betterproto compiler and then use protoc to compile your .proto files. Options can be passed to customize the generated output. ```bash # Install compiler features pip install "betterproto[compiler]" # Compile proto file protoc -I . --python_betterproto_out=./lib example.proto # With options protoc -I . --python_betterproto_out=./lib \ --python_betterproto_opt=pydantic_dataclasses \ --python_betterproto_opt=typing.310 \ example.proto ``` -------------------------------- ### Install Betterproto from PyPI Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/quick-start.rst Install the Betterproto library from the Python Package Index. Use the 'py -3' command on Windows. ```sh python3 -m pip install -U betterproto ``` ```sh py -3 -m pip install -U betterproto ``` -------------------------------- ### Basic gRPC Server Implementation Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md A straightforward implementation of a gRPC server using a betterproto-generated base class. This example shows how to subclass the base class and implement the service methods, then start the server. ```python from grpclib.server import Server from myproto.example import ExampleBase, HelloRequest, HelloResponse class ExampleService(ExampleBase): async def hello( self, request: HelloRequest, ) -> HelloResponse: return HelloResponse(message=f"Hello, {request.name}!") async def main(): server = Server([ExampleService()]) await server.start("0.0.0.0", 50051) print("Server started on port 50051") await server.wait_closed() import asyncio asyncio.run(main()) ``` -------------------------------- ### Core Documents Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/MANIFEST.txt These documents provide an overview, installation guide, and navigation for the project. ```APIDOC ## README.md ### Description Overview of documentation package, navigation guide, quick reference, and installation/version information. ## OVERVIEW.md ### Description Comprehensive project summary, architecture, design patterns, key features, capabilities, and version/dependency information. ## INDEX.md ### Description Complete navigation index, document organization by topic, search guide, API reference by use case, and quick reference commands. ``` -------------------------------- ### Install Betterproto with Compiler Plugin Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/quick-start.rst Install Betterproto along with the protoc plugin for compiling proto files. This is done by specifying the 'compiler' extra. ```sh python3 -m pip install -U "betterproto[compiler]" ``` -------------------------------- ### Install Betterproto Library Only Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/README.md Command to install only the Betterproto library, useful when you only need to use generated code. ```bash pip install betterproto ``` -------------------------------- ### Integration Test Example Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md An example showing how to set up and test a gRPC service using pytest, grpclib, and a client stub. ```APIDOC ## Integration Test Example ```python import pytest from grpclib.client import Channel from grpclib.server import Server from myproto.service import MyServiceBase, MyServiceStub class TestService(MyServiceBase): async def my_method(self, request): return MyResponse(result="test") @pytest.mark.asyncio async def test_grpc_call(): # Start server server = Server([TestService()]) await server.start("127.0.0.1", 50051) try: # Create client channel = Channel("127.0.0.1", 50051) stub = MyServiceStub(channel) # Make call response = await stub.my_method(MyRequest(field="value")) assert response.result == "test" finally: await channel.close() await server.close() ``` ``` -------------------------------- ### Example Usage of Service Stub Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-stub.md Demonstrates how to create a channel, instantiate a service stub, and make both unary-unary and unary-stream RPC calls. ```APIDOC ## Example Usage of Service Stub This section shows a practical example of how to use a generated service stub to interact with a gRPC service. ### Description This code snippet illustrates the typical workflow for a client using a betterproto-generated service stub. It covers channel creation, stub instantiation, making a unary-unary call (`hello`), and a unary-stream call (`hello_stream`), followed by closing the channel. ### Method Asynchronous Python functions using `grpclib`. ### Endpoint Not directly applicable as this is client-side SDK usage, but the underlying gRPC service endpoints are: - `/example.Example/Hello` (Unary-Unary) - `/example.Example/HelloStream` (Unary-Stream) ### Parameters **For `stub.hello` (Unary-Unary):** - `request` (HelloRequest): The request message object. - `timeout` (Optional[float]): The timeout for the RPC in seconds. - `deadline` (Optional[Deadline]): The deadline for the RPC. - `metadata` (Optional[MetadataLike]): Metadata to send with the RPC. **For `stub.hello_stream` (Unary-Stream):** - `request` (HelloRequest): The request message object. - `timeout` (Optional[float]): The timeout for the RPC in seconds. - `deadline` (Optional[Deadline]): The deadline for the RPC. - `metadata` (Optional[MetadataLike]): Metadata to send with the RPC. ### Request Example ```python from grpclib.client import Channel from myservice.example import ExampleStub, HelloRequest, HelloResponse async def main(): channel = Channel(host="localhost", port=50051) stub = ExampleStub(channel, timeout=10.0) # Unary-Unary request response = await stub.hello(HelloRequest(name="Alice")) print(response.greeting) # Unary-Stream request async for message in stub.hello_stream(HelloRequest(name="Bob')): print(message.greeting) await channel.close() import asyncio asyncio.run(main()) ``` ### Response **For `stub.hello` (Unary-Unary):** - `HelloResponse`: The response message object. **For `stub.hello_stream` (Unary-Stream):** - `AsyncIterator[HelloResponse]`: An asynchronous iterator yielding `HelloResponse` messages. #### Response Example (Unary-Unary) ```json { "greeting": "Hello, Alice!" } ``` #### Response Example (Unary-Stream - each message) ```json { "greeting": "Hello, Bob!" } ``` ``` -------------------------------- ### Implement a gRPC Server Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/INDEX.md Create a service class that inherits from `MyServiceBase` and implement the required methods. Start the server on a specific address and port. ```python from grpclib.server import Server from myservice import MyServiceBase class MyService(MyServiceBase): async def my_method(self, request): return MyResponse(...) server = Server([MyService()]) await server.start("0.0.0.0", 50051) ``` -------------------------------- ### gRPC Client Example with Python Betterproto Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/README.md Shows how to set up a gRPC client using grpclib and Betterproto. Requires a running gRPC server and generated service stubs. ```python from grpclib.client import Channel from myservice import MyServiceStub async def main(): channel = Channel(host="localhost", port=50051) stub = MyServiceStub(channel) response = await stub.my_method(request) await channel.close() ``` -------------------------------- ### gRPC Server Example with Python Betterproto Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/README.md Illustrates how to implement a gRPC server using grpclib and Betterproto. Requires a base service class generated by Betterproto. ```python from grpclib.server import Server from myservice import MyServiceBase class MyService(MyServiceBase): async def my_method(self, request): return MyResponse(...) async def main(): server = Server([MyService()]) await server.start("0.0.0.0", 50051) await server.wait_closed() ``` -------------------------------- ### Unit Test Example Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md An example demonstrating how to write a unit test for a service method using pytest. ```APIDOC ## Unit Test Example ```python import pytest from myservice import MyService, MyRequest, MyResponse @pytest.mark.asyncio async def test_my_method(): service = MyService() request = MyRequest(field="value") response = await service.my_method(request) assert response.result == "expected" ``` ``` -------------------------------- ### gRPC Server Implementation Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/quick-start.rst Example of implementing a gRPC server by subclassing the generated base service class and overriding RPC methods. ```python from echo import EchoBase from grpclib.server import Server from typing import AsyncIterator class EchoService(EchoBase): async def echo(self, value: str, extra_times: int) -> "EchoResponse": return value async def echo_stream( self, value: str, extra_times: int ) -> AsyncIterator["EchoStreamResponse"]: for _ in range(extra_times): yield value async def start_server(): HOST = "127.0.0.1" PORT = 1337 server = Server([EchoService()]) await server.start(HOST, PORT) await server.serve_forever() ``` -------------------------------- ### Start gRPC Server and Wait for Shutdown Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Start the server to listen on a specified host and port. The `wait_closed` method keeps the server running until it's shut down. ```python # Start listening await server.start( host="0.0.0.0", port=50051, ) print(f"Server listening on 0.0.0.0:50051") # Wait for shutdown await server.wait_closed() ``` -------------------------------- ### Python Service Stub Usage Example Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-stub.md Demonstrates creating a channel, instantiating a service stub, and making unary-unary and unary-stream RPC requests. Remember to close the channel when done. ```python from grpclib.client import Channel from myservice.example import ExampleStub, HelloRequest, HelloResponse async def main(): # Create channel and stub channel = Channel(host="localhost", port=50051) stub = ExampleStub(channel, timeout=10.0) # Make unary-unary request response = await stub.hello(HelloRequest(name="Alice")) print(response.greeting) # Make unary-stream request async for message in stub.hello_stream(HelloRequest(name="Bob")): print(message.greeting) # Close the channel when done await channel.close() # Run the async function import asyncio asyncio.run(main()) ``` -------------------------------- ### Example MetadataLike Usage Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/types.md Illustrates the two accepted formats for gRPC metadata: a dictionary mapping strings to values, or a list of (key, value) tuples. ```python # Example using a dictionary metadata_dict = {"authorization": "Bearer token"} # Example using a list of tuples metadata_list = [("authorization", "Bearer token")] print(metadata_dict) print(metadata_list) ``` -------------------------------- ### Implement an Async gRPC Server Source: https://github.com/danielgtaylor/python-betterproto/blob/master/README.md Example of an asynchronous gRPC server implementing the Echo service by subclassing the generated EchoBase. ```python import asyncio from echo import EchoBase, EchoRequest, EchoResponse, EchoStreamResponse from grpclib.server import Server from typing import AsyncIterator class EchoService(EchoBase): async def echo(self, echo_request: "EchoRequest") -> "EchoResponse": return EchoResponse([echo_request.value for _ in range(echo_request.extra_times)]) async def echo_stream(self, echo_request: "EchoRequest") -> AsyncIterator["EchoStreamResponse"]: for _ in range(echo_request.extra_times): yield EchoStreamResponse(echo_request.value) async def main(): server = Server([EchoService()]) await server.start("127.0.0.1", 50051) await server.wait_closed() if __name__ == '__main__': loop = asyncio.get_event_loop() loop.run_until_complete(main()) ``` -------------------------------- ### Unit Test Example for MyService Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Demonstrates a basic unit test for a service method using pytest and asyncio. Ensure MyService, MyRequest, and MyResponse are correctly imported. ```python import pytest from myservice import MyService, MyRequest, MyResponse @pytest.mark.asyncio async def test_my_method(): service = MyService() request = MyRequest(field="value") response = await service.my_method(request) assert response.result == "expected" ``` -------------------------------- ### Define gRPC Service in Proto Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/quick-start.rst Example of defining gRPC services, including unary and streaming RPCs, within a .proto file. ```proto syntax = "proto3"; package echo; message EchoRequest { string value = 1; // Number of extra times to echo uint32 extra_times = 2; } message EchoResponse { repeated string values = 1; } message EchoStreamResponse { string value = 1; } service Echo { rpc Echo(EchoRequest) returns (EchoResponse); rpc EchoStream(EchoRequest) returns (stream EchoStreamResponse); } ``` -------------------------------- ### Test Case Directory Structure Example Source: https://github.com/danielgtaylor/python-betterproto/blob/master/tests/README.md Illustrates the typical file organization for a single test case, including the protobuf definition, optional JSON data, and optional custom pytest file. ```bash bool/ bool.proto bool.json # optional test_bool.py # optional ``` -------------------------------- ### Implement an Async gRPC Client Source: https://github.com/danielgtaylor/python-betterproto/blob/master/README.md Example of an asynchronous gRPC client using grpclib to call Echo and EchoStream RPC methods. ```python import asyncio import echo from grpclib.client import Channel async def main(): channel = Channel(host="127.0.0.1", port=50051) service = echo.EchoStub(channel) response = await service.echo(echo.EchoRequest(value="hello", extra_times=1)) print(response) async for response in service.echo_stream(echo.EchoRequest(value="hello", extra_times=1)): print(response) # don't forget to close the channel when done! channel.close() if __name__ == "__main__": loop = asyncio.get_event_loop() loop.run_until_complete(main()) ``` -------------------------------- ### Define a Protocol Buffer Message Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/quick-start.rst Example of a simple Protocol Buffer message definition in proto3 syntax. ```proto syntax = "proto3"; package hello; // Greeting represents a message you can tell a user. message Greeting { string message = 1; } ``` -------------------------------- ### Generated Python Service Stub Example Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-stub.md Illustrates the structure of a generated service stub inheriting from `ServiceStub`, showing implementations for unary-unary and unary-stream RPC methods. ```python class ExampleStub(ServiceStub): async def hello( self, request: HelloRequest, *, timeout: Optional[float] = None, deadline: Optional[Deadline] = None, metadata: Optional[MetadataLike] = None, ) -> HelloResponse: return await self._unary_unary( "/example.Example/Hello", request, HelloResponse, timeout=timeout, deadline=deadline, metadata=metadata, ) async def hello_stream( self, request: HelloRequest, *, timeout: Optional[float] = None, deadline: Optional[Deadline] = None, metadata: Optional[MetadataLike] = None, ) -> AsyncIterator[HelloResponse]: async for response in self._unary_stream( "/example.Example/HelloStream", request, HelloResponse, timeout=timeout, deadline=deadline, metadata=metadata, ): yield response ``` -------------------------------- ### Async gRPC Client Implementation Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/quick-start.rst Example of an asynchronous gRPC client using grpclib to interact with a defined service. Includes both unary and streaming calls. ```python import asyncio from grpclib.client import Channel import echo async def main(): channel = Channel(host="127.0.0.1", port=50051) service = echo.EchoStub(channel) response = await service.echo(value="hello", extra_times=1) print(response) async for response in service.echo_stream(value="hello", extra_times=1): print(response) # don't forget to close the channel when you're done! channel.close() asyncio.run(main()) # python 3.7 only # outputs EchoResponse(values=['hello', 'hello']) EchoStreamResponse(value='hello') EchoStreamResponse(value='hello') ``` -------------------------------- ### Integration Test Example for gRPC Service Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Shows how to perform an integration test for a gRPC service using grpclib. This involves setting up a test server and a client to make calls. Ensure MyServiceBase, MyServiceStub, MyRequest, and MyResponse are correctly imported. ```python import pytest from grpclib.client import Channel from grpclib.server import Server from myproto.service import MyServiceBase, MyServiceStub class TestService(MyServiceBase): async def my_method(self, request): return MyResponse(result="test") @pytest.mark.asyncio async def test_grpc_call(): # Start server server = Server([TestService()]) await server.start("127.0.0.1", 50051) try: # Create client channel = Channel("127.0.0.1", 50051) stub = MyServiceStub(channel) # Make call response = await stub.my_method(MyRequest(field="value")) assert response.result == "test" finally: await channel.close() await server.close() ``` -------------------------------- ### Start and Close gRPC Server/Channel Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/README.md Ensure your gRPC server is running using `await server.start(...)` and that the client channel is properly closed after use with `await channel.close()`. Verify the client is using the correct host and port. ```python await server.start(...) ``` ```python await channel.close() ``` -------------------------------- ### Bidirectional Streaming Method Example Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-stub.md Demonstrates how to implement a bidirectional streaming RPC method in a generated stub. This method sends a stream of requests and receives a stream of responses. ```python async def bidirectional_method( self, request_iterator: AsyncIterable[BiRequest], ) -> AsyncIterator[BiResponse]: async for response in self._stream_stream( "/package.Example/BiMethod", request_iterator, BiRequest, BiResponse, ): yield response ``` -------------------------------- ### MetadataLike Type Alias Example Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-stub.md Illustrates the usage of the MetadataLike type alias for gRPC metadata. It can be represented as a dictionary or a list of tuples. ```python MetadataLike = Union[ Mapping[str, Value], Collection[Tuple[str, Value]] ] ``` ```python # As dict metadata = {"authorization": "Bearer token"} # As list of tuples metadata = [("authorization", "Bearer token"), ("custom", "value")] ``` -------------------------------- ### JSON Test Data Example Source: https://github.com/danielgtaylor/python-betterproto/blob/master/tests/README.md Provides sample JSON data used to validate the generated protobuf message. This JSON should correspond to the fields defined in the .proto file. ```json { "value": true } ``` -------------------------------- ### Proto Definition for Well-Known Types Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/migrating.rst Example proto definition demonstrating the use of Google's well-known types like Duration, Timestamp, and wrappers. ```proto syntax = "proto3"; import "google/protobuf/duration.proto"; import "google/protobuf/timestamp.proto"; import "google/protobuf/wrappers.proto"; message Test { google.protobuf.BoolValue maybe = 1; google.protobuf.Timestamp ts = 2; google.protobuf.Duration duration = 3; } ``` -------------------------------- ### Pydantic Validation Example Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/OVERVIEW.md Instantiate a Pydantic-integrated message and handle potential validation errors. This assumes your proto definition includes validation rules. ```python from pydantic import ValidationError try: msg = MyMessage(count=-1) # Validates if rules defined except ValidationError as e: print(f"Validation error: {e}") ``` -------------------------------- ### Implement gRPC Server in Python Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/README.md Implements a gRPC server by extending the generated service base class and defining asynchronous methods. The server needs to be started with a host and port. ```python class MyService(MyServiceBase): async def my_method(self, request): return MyResponse(...) server = Server([MyService()]) await server.start("0.0.0.0", 50051) ``` -------------------------------- ### Generated Code with Python 3.10+ Type Syntax Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/configuration.md Example of generated Python code using Python 3.10+ native type syntax (e.g., list[str], str | None). Requires Python 3.10+. ```python class MyMessage(Message): items: list[str] = string_field(1) count: str | None = string_field(2) ``` -------------------------------- ### Basic Server Implementation Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Demonstrates how to implement a gRPC server by subclassing the generated base class and providing concrete method implementations. ```APIDOC ## Basic Server Implementation ```python from grpclib.server import Server from myproto.example import ExampleBase, HelloRequest, HelloResponse class ExampleService(ExampleBase): async def hello(self, request: HelloRequest) -> HelloResponse: return HelloResponse(message=f"Hello, {request.name}!") async def main(): server = Server([ExampleService()]) await server.start("0.0.0.0", 50051) print("Server started on port 50051") await server.wait_closed() import asyncio asyncio.run(main()) ``` ``` -------------------------------- ### Create gRPC Server Instance Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Instantiate a `Server` object with your service handlers. Multiple services can be registered by passing them as a list. ```python from grpclib.server import Server # Create service instance service = MyService() # Create server with service handlers server = Server([service]) ``` ```python server = Server([ ServiceOne(), ServiceTwo(), ServiceThree(), ]) ``` -------------------------------- ### Get Encoded Message Size Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/message.md Get the size in bytes of the encoded Protobuf representation of a message instance by calling the built-in len() function on it. This can be useful for pre-allocating buffers or estimating bandwidth usage. ```python msg = MyMessage(value="hello") size = len(msg) assert size == len(bytes(msg)) ``` -------------------------------- ### __len__() Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/message.md Get the size in bytes of the encoded Protobuf representation. ```APIDOC ## __len__() -> int ### Description Get the size in bytes of the encoded Protobuf representation. ### Returns `int` — The size of the encoded message in bytes ### Example ```python msg = MyMessage(value="hello") size = len(msg) assert size == len(bytes(msg)) ``` ``` -------------------------------- ### __bytes__() Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/message.md Get the binary encoded Protobuf representation of the message instance. ```APIDOC ## __bytes__() -> bytes ### Description Get the binary encoded Protobuf representation of the message instance. ### Returns `bytes` — The binary encoded Protobuf representation ### Example ```python msg = MyMessage(value="hello") binary = bytes(msg) assert isinstance(binary, bytes) ``` ``` -------------------------------- ### decode_varint Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/utilities.md Decodes a single varint value from a byte buffer starting at a specified position. ```APIDOC ## decode_varint(buffer: bytes, pos: int) -> Tuple[int, int] ### Description Decode a single varint value from a byte buffer. ### Parameters #### Path Parameters - **buffer** (bytes) - Required - The byte buffer to decode from - **pos** (int) - Required - The position in the buffer to start decoding ### Returns - **Tuple[int, int]** — A tuple of (decoded_value, new_position) ### Request Example ```python buffer = b'\xac\x02' value, pos = betterproto.decode_varint(buffer, 0) assert value == 300 assert pos == 2 ``` ``` -------------------------------- ### Create a gRPC Client Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/INDEX.md Instantiate a gRPC channel and a service stub to make calls to a gRPC server. Ensure the `MyServiceStub` is generated from your proto file. ```python from grpclib.client import Channel from myservice import MyServiceStub channel = Channel(host="localhost", port=50051) stub = MyServiceStub(channel) response = await stub.my_method(request) ``` -------------------------------- ### Generated Python Message Class Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/quick-start.rst Example of a Python class generated by Betterproto from a .proto file, using dataclasses. ```python # Generated by the protocol buffer compiler. DO NOT EDIT! # sources: example.proto # plugin: python-betterproto from dataclasses import dataclass import betterproto @dataclass class Greeting(betterproto.Message): """Greeting represents a message you can tell a user.""" message: str = betterproto.string_field(1) ``` -------------------------------- ### Get Enum Length Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/enum.md Determine the number of defined members in an enum. This is useful for validation or setting array sizes. ```python class Color(Enum): RED = 0 GREEN = 1 BLUE = 2 assert len(Color) == 3 ``` -------------------------------- ### Enum Immutability Example Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/enum.md Demonstrates that enum members and the enum class itself are immutable after definition. Attempting to modify them will raise an AttributeError. ```python class Color(Enum): RED = 0 Color.RED.name = "CRIMSON" # Raises AttributeError Color.YELLOW = 3 # Raises AttributeError ``` -------------------------------- ### Perform Async gRPC Calls Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/OVERVIEW.md Shows how to establish a gRPC channel, create a client stub, and make various types of asynchronous calls, including unary-unary and server-streaming. Remember to close the channel when done. ```python # Create client channel = Channel(host="localhost", port=50051) stub = MyServiceStub(channel, timeout=10.0) # Unary-unary call response = await stub.my_method(request) # Server-streaming call async for response in stub.streaming_method(request): process(response) # Clean up await channel.close() ``` -------------------------------- ### Create, Serialize, and Parse Messages Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/OVERVIEW.md Demonstrates how to instantiate message objects, serialize them to binary, parse them back from binary, and convert them to and from JSON and dictionaries. ```python # Create a message msg = MyMessage(field1="value", field2=42) # Serialize to binary binary = bytes(msg) # Parse from binary msg2 = MyMessage().parse(binary) # Convert to/from JSON json_str = msg.to_json() msg3 = MyMessage().from_json(json_str) # Convert to/from dict d = msg.to_dict() msg4 = MyMessage.from_dict(d) ``` -------------------------------- ### Generate Test Assets and Run Tests Source: https://github.com/danielgtaylor/python-betterproto/blob/master/README.md Generates necessary assets from .proto files for tests and then executes the test suite. ```shell poe generate poe test ``` -------------------------------- ### Compile with gRPC Services using Betterproto Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/configuration.md This command compiles .proto files to generate both message classes and gRPC service client stubs and server base classes. Ensure the output directory is correctly specified. ```bash protoc \ -I . \ --python_betterproto_out=./lib \ --python_betterproto_out=./lib \ example.proto ``` -------------------------------- ### Get Enum Members Dictionary Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/enum.md Access all defined members of an enum as a read-only dictionary. Useful for checking existence or iterating through members. ```python class Status(Enum): ACTIVE = 1 INACTIVE = 2 members = Status.__members__ assert members["ACTIVE"] == Status.ACTIVE assert "INACTIVE" in members ``` -------------------------------- ### Compile Proto File with grpcio-tools Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/quick-start.rst Compile a .proto file using grpcio-tools, which also integrates with Betterproto for code generation. ```sh pip install grpcio-tools python -m grpc_tools.protoc -I . --python_betterproto_out=lib example.proto ``` -------------------------------- ### Run Concurrent HTTP and gRPC Servers with FastAPI Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Illustrates how to run both a FastAPI HTTP server and a grpclib gRPC server concurrently using asyncio. ```python import asyncio from fastapi import FastAPI from grpclib.server import Server async def run_servers(): # Start HTTP server (FastAPI) http_server_task = asyncio.create_task( uvicorn.run(app, host="0.0.0.0", port=8000) ) # Start gRPC server grpc_server = Server([MyService()]) grpc_server_task = asyncio.create_task( grpc_server.start("0.0.0.0", 50051) ) # Wait for both await asyncio.gather(http_server_task, grpc_server_task) ``` -------------------------------- ### from_string(name: str) -> Self Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/enum.md Get the enum member by name. This method is useful for retrieving an enum member when you only know its string representation. ```APIDOC ## `from_string(name: str) -> Self` Get the enum member by name. ### Parameters #### Query Parameters - **name** (str) - Required - The name of the enum member ### Returns `Enum` — The enum member with the given name ### Raises `ValueError` — If the name does not correspond to a member ### Example ```python class Color(Enum): RED = 0 GREEN = 1 green = Color.from_string("GREEN") assert green == Color.GREEN ``` ``` -------------------------------- ### Integrate FastAPI with grpclib Server Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Shows how to integrate a grpclib server with a FastAPI application using event handlers for startup and shutdown. ```python from fastapi import FastAPI from grpclib.server import Server app = FastAPI() @app.on_event("startup") async def startup(): server = Server([MyService()]) await server.start("0.0.0.0", 50051) app.state.grpc_server = server @app.on_event("shutdown") async def shutdown(): await app.state.grpc_server.close() ``` -------------------------------- ### Integrate Starlette with grpclib Server Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Demonstrates how to integrate a grpclib server with a Starlette application using its lifespan management for startup and shutdown. ```python from starlette.applications import Starlette from grpclib.server import Server class GRPCService(ServiceBase): # ... implement methods pass async def lifespan(app): # Startup server = Server([GRPCService()]) await server.start("0.0.0.0", 50051) app.state.grpc_server = server yield # Shutdown await server.close() app = Starlette(lifespan=lifespan) ``` -------------------------------- ### Decode Varint from Buffer Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/utilities.md Decodes a varint value from a byte buffer starting at a specified position. Returns the decoded value and the new position in the buffer. ```python buffer = b'\xac\x02' value, pos = betterproto.decode_varint(buffer, 0) assert value == 300 assert pos == 2 ``` -------------------------------- ### Run Full Test Suite with Tox Source: https://github.com/danielgtaylor/python-betterproto/blob/master/README.md Executes the complete test suite, mimicking the environment used in CI, by running tox. ```shell poe full-test ``` -------------------------------- ### MessageSource Type Alias Example Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-stub.md Shows how to use the MessageSource type alias for request message streams. It accepts either a regular iterable or an async iterable. ```python MessageSource = Union[ Iterable[IProtoMessage], AsyncIterable[IProtoMessage] ] ``` ```python # Regular iterable def sync_messages(): yield request1 yield request2 # Async iterable async def async_messages(): yield request1 yield request2 ``` -------------------------------- ### Instantiate ServiceStub with Custom Metadata Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-stub.md Initialize a ServiceStub with custom metadata headers, such as authorization tokens, to be included in all outgoing requests. ```python stub = ExampleStub( channel, metadata={"authorization": "Bearer token123"} ) ``` -------------------------------- ### Content Coverage Summary Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/MANIFEST.txt Summary of the types of content covered in the documentation, including exported classes, functions, type definitions, configuration options, and code examples. ```APIDOC ## Exported Classes - Message - Enum - ServiceStub - Casing ## Exported Functions - 50+ total - 17 field definition functions - 8 utility functions - 4 casing functions - 8 varint functions - 5 parsing functions - Plus service-generated methods ## Type Definitions - 25+ total - FieldMetadata - ProtoClassMetadata - ParsedField - Type aliases - Well-known type mappings ## Configuration Options - 4 total - pydantic_dataclasses - typing.direct - typing.root - typing.310 ## Code Examples - 150+ total - Message serialization - gRPC clients and servers - Field definitions - Enum usage - Error handling - Integration patterns ``` -------------------------------- ### Registering a Generated Service Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Implement the generated service base class and pass an instance to the grpclib Server. The server automatically detects and registers the service. ```python from grpclib.server import Server from myproto.service import MyServiceBase class MyService(MyServiceBase): # Implement methods pass # Server automatically detects and registers the service server = Server([MyService()]) ``` -------------------------------- ### Keep Methods Short - Good vs. Avoid Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Compares a good practice of keeping service methods short by delegating complex logic to helper methods, versus the anti-pattern of including extensive logic directly within a single method. ```python # Good async def create_user(self, request: CreateUserRequest) -> User: return await self._create_and_validate(request) async def _create_and_validate(self, request): # Complex logic here pass # Avoid async def create_user(self, request: CreateUserRequest) -> User: # 500 lines of validation, database, caching, etc. pass ``` -------------------------------- ### Instantiate ServiceStub with Channel and Timeout Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-stub.md Create an instance of a generated gRPC service stub, providing the communication channel and an optional request timeout. Remember to close the channel when done. ```python from grpclib.client import Channel from myproto.example import ExampleStub channel = Channel(host="localhost", port=50051) stub = ExampleStub(channel, timeout=10.0) # Later, close the channel channel.close() ``` -------------------------------- ### ServiceStub Constructor Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-stub.md Initializes a ServiceStub with a gRPC channel and optional request parameters like timeout, deadline, and metadata. ```APIDOC ## ServiceStub Constructor ### Description Initializes a ServiceStub with a gRPC channel and optional request parameters like timeout, deadline, and metadata. ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Request Example ```python from grpclib.client import Channel from myproto.example import ExampleStub channel = Channel(host="localhost", port=50051) stub = ExampleStub(channel, timeout=10.0) # Later, close the channel channel.close() ``` ### Response #### Success Response (200) None #### Response Example None ``` -------------------------------- ### Build Script for Multiple Codebases Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/configuration.md This bash script compiles all proto files within a project, supporting multiple input directories and output locations. It also includes Pydantic support and custom include paths. ```bash #!/bin/bash OUTPUT_DIR="./generated" PROTO_DIR="./proto" # Compile all proto files protoc \ -I $PROTO_DIR \ -I /usr/local/include \ --python_betterproto_out=$OUTPUT_DIR \ --python_betterproto_opt=pydantic_dataclasses \ $PROTO_DIR/**/*.proto ``` -------------------------------- ### Generated Code with Direct Typing Imports Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/configuration.md Example of generated Python code using direct imports from the 'typing' module (e.g., List[str]). This is the default behavior. ```python from typing import ( List, Optional, Union, ) class MyMessage(Message): items: List[str] = string_field(1) count: Optional[int] = int32_field(2) ``` -------------------------------- ### Serialize Message to Binary Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/message.md Get the binary encoded Protobuf representation of a message instance by calling the built-in bytes() function on it. This is useful for sending messages over a network or storing them. ```python msg = MyMessage(value="hello") binary = bytes(msg) assert isinstance(binary, bytes) ``` -------------------------------- ### Get Oneof Field Value in Betterproto Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/errors.md Retrieve the name and value of the currently set field within a oneof group in a Betterproto message. Returns None, None if no field is set. ```python def get_oneof_value(msg, group_name): field_name, value = betterproto.which_one_of(msg, group_name) if not field_name: return None, None return field_name, value ``` -------------------------------- ### Serialize and Parse Message with Betterproto Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/README.md Demonstrates how to create, serialize, and parse a message using Betterproto. Ensure the dataclass inherits from Message and uses appropriate field decorators. ```python from betterproto import Message, string_field from dataclasses import dataclass @dataclass class Greeting(Message): message: str = string_field(1) # Create and serialize greeting = Greeting(message="Hello, World!") binary = bytes(greeting) # Parse greeting2 = Greeting().parse(binary) assert greeting2.message == "Hello, World!" ``` -------------------------------- ### Get Enum Member by Name Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/enum.md Use the from_string method to retrieve an enum member by its string name. This method raises a ValueError if the provided name does not correspond to any defined enum member. ```python class Color(Enum): RED = 0 GREEN = 1 green = Color.from_string("GREEN") assert green == Color.GREEN ``` -------------------------------- ### Use Generated Message Class Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/quick-start.rst Demonstrates how to instantiate, set values, serialize, and deserialize a Betterproto-generated message class. ```python >>> from lib import Greeting >>> test = Greeting() >>> test Greeting(message='') >>> test.message = "Hey!" >>> test Greeting(message="Hey!") >>> bytes(test) b'\n\x04Hey!' >>> Greeting().parse(serialized) Greeting(message="Hey!") ``` -------------------------------- ### Generated gRPC Service Base Class Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md This is an example of a base class generated by the betterproto compiler for a gRPC service defined in a proto file. It outlines the methods that a server implementation must provide. ```python class ExampleBase: async def hello( self, request: HelloRequest, ) -> HelloResponse: raise NotImplementedError() async def stream( self, request: StreamRequest, ) -> AsyncIterator[StreamResponse]: raise NotImplementedError() ``` -------------------------------- ### Pytest Test Results Interpretation Source: https://github.com/danielgtaylor/python-betterproto/blob/master/tests/README.md Shows an example of pytest output indicating the status of various test cases, including passed (.), intentionally failed (x), and unexpectedly passed (X) tests. ```text betterproto/tests/test_inputs.py ..x...x..x...x.X........xx........x.....x.......x.xx....x...................... [ 84%] - . — PASSED - x — XFAIL: expected failure - X — XPASS: expected failure, but still passed ``` -------------------------------- ### Basic Message Compilation with Betterproto Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/configuration.md Use this command to generate Python message classes from a .proto file. The output will be placed in the specified directory. ```bash protoc -I . --python_betterproto_out=./lib example.proto ``` -------------------------------- ### Instantiate and Use Generated Message Source: https://github.com/danielgtaylor/python-betterproto/blob/master/README.md Demonstrates creating, setting values, serializing, and parsing a generated Betterproto message. ```python >>> from lib.hello import Greeting >>> test = Greeting() >>> test Greeting(message='') >>> test.message = "Hey!" >>> test Greeting(message="Hey!") >>> serialized = bytes(test) >>> serialized b'\n\x04Hey!' >>> another = Greeting().parse(serialized) >>> another Greeting(message="Hey!") >>> another.to_dict() {"message": "Hey!"} >>> another.to_json(indent=2) '{ "message": "!"\n}' ``` -------------------------------- ### try_value(value: int = 0) -> Self Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/enum.md Get the enum member for a given value, or create an unknown member. If the value does not correspond to a known member, an unknown member with the specified value is returned. ```APIDOC ## `try_value(value: int = 0) -> Self` Get the enum member for a given value, or create an unknown member. ### Parameters #### Query Parameters - **value** (int) - Optional - The enum value to look up. Defaults to 0. ### Returns `Enum` — The corresponding enum member, or a new unknown member if not found ### Example ```python class Color(Enum): RED = 0 GREEN = 1 red = Color.try_value(0) assert red == Color.RED unknown = Color.try_value(99) assert unknown.value == 99 assert unknown.name is None ``` ``` -------------------------------- ### Invoke Protoc with grpc_tools Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/configuration.md Alternatively, use `grpc_tools.protoc` to invoke the betterproto plugin, with options like `pydantic_dataclasses`. ```bash python -m grpc_tools.protoc \ -I . \ --python_betterproto_out=. \ --python_betterproto_opt=pydantic_dataclasses \ example.proto ``` -------------------------------- ### Get Detailed Message Representation (`__repr__`) Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/message.md The `__repr__` method provides a developer-friendly string representation of a message object. It displays all fields that have non-default values, making it easy to inspect the message's state. ```python msg = MyMessage(value="hello") print(repr(msg)) ``` -------------------------------- ### Get Type Information for Debugging in Betterproto Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/errors.md Inspect message structure and field metadata for debugging field access errors. This includes retrieving field numbers, proto types, and oneof group memberships. ```python # Get all field metadata for field_name, meta in msg._betterproto.meta_by_field_name.items(): print(f"{field_name}: number={meta.number}, type={meta.proto_type}") # Check oneof groups for group_name, fields in msg._betterproto.oneof_field_by_group.items(): print(f"Oneof group '{group_name}' has fields: {[f.name for f in fields]}") ``` -------------------------------- ### Control Typing Imports with Betterproto (Qualified Access) Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/configuration.md Use the `typing.root` option for qualified typing access, which can be useful in complex project structures. ```bash # Use qualified typing access protoc \ -I . \ --python_betterproto_out=./lib \ --python_betterproto_opt=typing.root \ example.proto ``` -------------------------------- ### Protoc Compiler Command Format Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/configuration.md Standard format for invoking the protoc compiler with betterproto options. ```bash protoc \ -I . \ --python_betterproto_out= \ --python_betterproto_opt= \ --python_betterproto_opt= \ example.proto ``` -------------------------------- ### Control Typing Imports with Betterproto (Python 3.10+) Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/configuration.md Use the `typing.310` option to leverage Python 3.10+ type hinting syntax for cleaner imports. ```bash # Use Python 3.10+ syntax protoc \ -I . \ --python_betterproto_out=./lib \ --python_betterproto_opt=typing.310 \ example.proto ``` -------------------------------- ### Get Enum Member by Value with Unknown Handling Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/enum.md Use the try_value method to retrieve an enum member by its integer value. If the value does not match any defined member, an unknown member with the specified value and no name is returned. ```python class Color(Enum): RED = 0 GREEN = 1 red = Color.try_value(0) assert red == Color.RED unknown = Color.try_value(99) assert unknown.value == 99 assert unknown.name is None ``` -------------------------------- ### gRPC Service Definition and Generated Base Class Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Illustrates a gRPC service definition in proto and the corresponding Python base class generated by betterproto. ```APIDOC ## Proto Service Definition ```protobuf service Example { rpc Hello(HelloRequest) returns (HelloResponse); rpc Stream(StreamRequest) returns (stream StreamResponse); } ``` ## Generated Python Base Class ```python class ExampleBase: async def hello(self, request: HelloRequest) -> HelloResponse: raise NotImplementedError() async def stream(self, request: StreamRequest) -> AsyncIterator[StreamResponse]: raise NotImplementedError() ``` ``` -------------------------------- ### Compile Proto File with Betterproto Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/quick-start.rst Compile a .proto file using the protoc compiler with the betterproto plugin. This generates Python code for the defined messages. ```sh mkdir hello protoc -I . --python_betterproto_out=lib example.proto ``` -------------------------------- ### Generated Code with Root Typing Import Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/configuration.md Example of generated Python code importing the 'typing' module as a whole and using qualified names (e.g., typing.List[str]). Use this to avoid name conflicts with types like 'Optional' or 'List'. ```python import typing class MyMessage(Message): items: typing.List[str] = string_field(1) count: typing.Optional[int] = int32_field(2) ``` -------------------------------- ### Managing Service State with Instance Variables Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/api-reference/service-base.md Maintain state within your service instance by using instance variables. This allows services to store and retrieve data across multiple requests. ```python class StatefulService(ServiceBase): def __init__(self): self.state = {} async def set_state(self, request: SetRequest) -> Empty: self.state[request.key] = request.value return Empty() async def get_state(self, request: GetRequest) -> Value: return Value(value=self.state.get(request.key, "")) ``` -------------------------------- ### JSON Serialization with Casing Control Source: https://github.com/danielgtaylor/python-betterproto/blob/master/docs/quick-start.rst Demonstrates using to_json and from_json methods for JSON serialization. Shows how to control field name casing, defaulting to CAMEL case. ```python import betterproto from dataclasses import dataclass @dataclass class MyMessage(betterproto.Message): a_long_field_name: str = betterproto.string_field(1) ``` -------------------------------- ### Compile Proto File with Pydantic Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/OVERVIEW.md Compile a .proto file into Python code using Pydantic dataclasses for enhanced data validation. ```bash protoc -I . --python_betterproto_out=./lib \ --python_betterproto_opt=pydantic_dataclasses example.proto ``` -------------------------------- ### gRPC Service Client in Python Source: https://github.com/danielgtaylor/python-betterproto/blob/master/_autodocs/OVERVIEW.md Establish a gRPC channel and create a service client stub to make remote procedure calls. Remember to close the channel when done. ```python from grpclib.client import Channel from myproto.service import ServiceStub channel = Channel(host="localhost", port=50051) stub = ServiceStub(channel) response = await stub.method(request) await channel.close() ```