Proto first

When the contract already exists as .proto files, or is shared with other languages, generate the Python side from it:

fastgrpc3 generate proto/ -o src/myapp/gen          # -I third_party/ for extra imports
fastgrpc3 generate proto/ -o src/myapp/gen --check  # in CI: fail if not up to date

PROTO_PATH is a directory (every .proto under it) or a file. The options can live in pyproject.toml, so that a bare fastgrpc3 generate does it:

pyproject.toml

[tool.fastgrpc3.generate]
proto_paths = ["proto"]
include = ["third_party"]
output = "src/myapp/gen"

Only proto3 files are supported.

What is generated

The layout follows the .proto files; every directory is a package:

src/myapp/gen/
├── __init__.py
├── _descriptor.py               DESCRIPTOR: the compiled descriptor set
└── orders/v1/
    ├── __init__.py
    ├── orders_models.py         one ProtoModel per message, one ProtoEnum per enum
    └── orders_service.py        the contract (Orders) and a typed client (OrdersClient)

Imported files are generated too, except the well-known types, which map to Python types (see Type mapping). Generated files start with # Generated by fastgrpc3; a later generate rewrites them and deletes those no longer produced. The code passes mypy --strict and ruff.

For this .proto:

syntax = "proto3";
package orders.v1;
import "google/protobuf/timestamp.proto";

enum OrderStatus { ORDER_STATUS_UNSPECIFIED = 0; ORDER_STATUS_PAID = 1; }

message GetOrderRequest { string order_number = 1; }

message Order {
  string order_number = 1;
  OrderStatus status = 2;
  google.protobuf.Timestamp created_at = 3;
  optional string note = 4;
}

service Orders {
  rpc GetOrder(GetOrderRequest) returns (Order);
  rpc Watch(GetOrderRequest) returns (stream Order);
}

orders_models.py holds ordinary code-first models:

class OrderStatus(ProtoEnum, proto="orders.v1.OrderStatus"):
    UNSPECIFIED = 0
    PAID = 1


class Order(ProtoModel, proto="orders.v1.Order"):
    order_number: str = field(1, default="")
    status: OrderStatus = field(2, default=OrderStatus.UNSPECIFIED)
    created_at: datetime.datetime | None = field(3, default=None)
    note: str | None = field(4, default=None)

A nested type (message Order { message Line {} }) is the class Order_Line, also reachable as Order.Line: both names are the same class, typed for mypy.

and orders_service.py the contract and the client:

Orders = ServiceDef(
    "Orders",
    "orders.v1",
    (
        MethodDef("GetOrder", _orders_v1_orders.GetOrderRequest, _orders_v1_orders.Order),
        MethodDef(
            "Watch",
            _orders_v1_orders.GetOrderRequest,
            _orders_v1_orders.Order,
            server_streaming=True,
        ),
    ),
)


class OrdersClient(GrpcClient):
    """Client of orders.v1.Orders."""

    package = "orders.v1"
    service = "Orders"

    get_order = GrpcClient.unary(
        "GetOrder", _orders_v1_orders.GetOrderRequest, _orders_v1_orders.Order
    )
    watch = GrpcClient.unary_stream(
        "Watch", _orders_v1_orders.GetOrderRequest, _orders_v1_orders.Order
    )

Comments of the .proto become docstrings and description=.

Implement the contract

GrpcService.implement(contract) checks every handler against the contract: an unknown method, a different request or response model, or a different call kind fails at import; a missing handler fails when the app starts.

from collections.abc import AsyncIterator

from fastgrpc3 import GrpcApp, GrpcService, Insecure

from myapp.gen.orders.v1.orders_models import GetOrderRequest, Order, OrderStatus
from myapp.gen.orders.v1.orders_service import Orders

service = GrpcService.implement(Orders)


@service.unary("GetOrder")
async def get_order(req: GetOrderRequest) -> Order:
    return Order(order_number=req.order_number, status=OrderStatus.PAID)


@service.unary_stream("Watch")
async def watch(req: GetOrderRequest) -> AsyncIterator[Order]:
    yield Order(order_number=req.order_number)


app = GrpcApp(tls=Insecure())
app.add_service(service)

The handlers are written exactly as in code first: Context, Depends(...), errors and the rest are described in Server.

Call it

from myapp.gen.orders.v1.orders_service import OrdersClient

async with Channel("orders:50051", tls=TLS(ca="ca.pem")) as channel:
    order = await OrdersClient(channel).get_order(GetOrderRequest(order_number="A"))

Mixing both modes

A server can serve generated and code-first services side by side, and a code-first model can use a generated one as a field type. The generated models carry their file and package (file=, proto=), so the contract compiled at startup keeps the original .proto layout: rendering generated models gives back descriptors equivalent to the original files (same files, packages, messages, fields, reserved, enums and services; options and comments are not kept).

_descriptor.py holds the descriptor as compiled by generate. Schema.from_descriptor(registry, DESCRIPTOR) binds models to it without compiling, and refuses models that no longer match it (generated modules older than the descriptor, or edited by hand).