DEV Community: William

gRPC in Python, the FastAPI way: models in, models out

William — Tue, 03 Nov 2026 10:58:05 +0000

Here is what a gRPC handler usually looks like in Python:

# 1. write orders.proto
# 2. python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. orders.proto
# 3. commit orders_pb2.py and orders_pb2_grpc.py, regenerate them at every change
import grpc
import orders_pb2, orders_pb2_grpc


class OrdersServicer(orders_pb2_grpc.OrdersServicer):
    async def Get(self, request, context):
        if request.order_number not in DB:
            await context.abort(grpc.StatusCode.NOT_FOUND, "no such order")
        return orders_pb2.Order(order_number=request.order_number, paid=True)

It works. But the messages are not your domain models, nothing validates them, the generated files drift from the schema, and every new service starts with the same servicer-class ceremony. If you have written a FastAPI endpoint, you know it does not have to be like that.

This tutorial builds the same service with fastgrpc3, an open-source gRPC framework I wrote on top of my Rust protobuf codec, protoruf.
By the end you will have a service, its .proto contract, a typed client, tests, and the option to run it on a Rust HTTP/2 server — without writing a single line of protobuf by hand.

Install

pip install fastgrpc3

Python 3.12 to 3.14. There is no protoc to install: protobuf is compiled and encoded in Rust by protoruf, which ships as a wheel.

A service is a function and two models

# orders.py
from pydantic import BaseModel
from fastgrpc3 import GrpcApp, GrpcError, GrpcService, Insecure, StatusCode


class GetOrder(BaseModel):
    order_number: str


class Order(BaseModel):
    order_number: str
    paid: bool = False


DB = {"A": Order(order_number="A", paid=True)}

service = GrpcService("Orders", package="orders.v1")


@service.unary()
async def Get(req: GetOrder) -> Order:
    if req.order_number not in DB:
        raise GrpcError(StatusCode.NOT_FOUND, "no such order")
    return DB[req.order_number]


app = GrpcApp(address="[::]:50051", tls=Insecure())
app.add_service(service)
app.enable_reflection()  # so that grpcurl can discover the service

The request is validated by Pydantic before your function runs; a payload that does not fit the model is answered with INVALID_ARGUMENT. The return value is checked against the response model. GrpcError becomes a real gRPC status, readable by any client in any language. Note that tls= is mandatory: Insecure() is a choice you make on purpose, not a default you forget.

Run it:

fastgrpc3 run orders:app

And call it from the command line with grpcurl, which reads the contract through server reflection:

grpcurl -plaintext -d '{"order_number": "A"}' localhost:50051 orders.v1.Orders/Get
# {
#   "orderNumber": "A",
#   "paid": true
# }

The `.proto` is generated from your models

Your models are the contract. Field numbers are what keeps protobuf compatible over time, so fastgrpc3 does not leave them to chance: fastgrpc3 lock pins them in a proto.lock file you commit. Adding a field later never renumbers the others, and a removed field's number stays reserved. Then export renders the contract as standard
.proto files, for the teams that call you in Go, Java or TypeScript:

fastgrpc3 lock orders:app                  # write proto.lock (commit it)
fastgrpc3 export orders:app -o proto/      # proto/orders/v1/orders.proto
fastgrpc3 check orders:app                 # CI: fail if proto.lock is out of date
fastgrpc3 breaking orders:app --against main   # CI: fail on a breaking change

// proto/orders/v1/orders.proto
syntax = "proto3";

package orders.v1;

message GetOrder {
  string order_number = 1;
}

message Order {
  string order_number = 1;
  bool paid = 2;
}

service Orders {
  rpc Get(GetOrder) returns (Order);
}

Already have .proto files? It works the other way too: fastgrpc3 generate proto/ -o src/myapp/gen produces Pydantic models, the service contract and a typed client, all passing mypy --strict.

A typed client

# client.py
import asyncio
from fastgrpc3 import Channel, GrpcClient, GrpcError, Insecure
from orders import GetOrder, Order


class OrdersClient(GrpcClient):
    package = "orders.v1"
    service = "Orders"
    get = GrpcClient.unary("Get", GetOrder, Order)


async def main() -> None:
    async with Channel("localhost:50051", tls=Insecure()) as channel:
        orders = OrdersClient(channel)
        print(await orders.get(GetOrder(order_number="A")))  # order_number='A' paid=True
        try:
            await orders.get(GetOrder(order_number="Z"))
        except GrpcError as error:
            print(error.code.name, error.message)  # NOT_FOUND no such order


if __name__ == "__main__":
    asyncio.run(main())

What you get back is an Order, not a protobuf message: your editor and mypy know its fields.

Dependencies, FastAPI-style

Real handlers need a database session, the current user, a settings object. Depends works the way you expect, including cleanup after the response for yield dependencies:

from collections.abc import AsyncIterator
from fastgrpc3 import Context, Depends


async def session() -> AsyncIterator[dict[str, Order]]:
    yield DB  # open a real session here, close it after the response


def current_user(ctx: Context) -> str:
    return str(ctx.metadata.get("x-user", "anonymous"))


@service.unary()
async def Pay(
    req: GetOrder,
    db: dict[str, Order] = Depends(session),
    user: str = Depends(current_user),
) -> Order:
    order = db[req.order_number]
    order.paid = True
    return order

Sub-dependencies, a per-call cache, @blocking for providers that do blocking I/O, and app.dependency_overrides for tests are all there. So are hooks, middlewares and exception handlers around every call.

Tests against a real server

# test_orders.py
from fastgrpc3.testing import TestClient
from orders import GetOrder, app
from client import OrdersClient


async def test_get() -> None:
    async with TestClient(app) as tc:
        order = await tc.client(OrdersClient).get(GetOrder(order_number="A"))
    assert order.paid

TestClient starts the app on a free local port: you test the real wire, not a mock.

Ready for production

A few lines turn the service into something you can deploy:

from fastgrpc3 import TLS

app = GrpcApp(address="[::]:50051", tls=TLS(cert="server.pem", key="server.key"))
app.enable_health()     # grpc.health.v1: Kubernetes probes, grpc-health-probe
app.enable_reflection()
app.enable_docs()       # Swagger UI on :8080 (pip install "fastgrpc3[docs]")

fastgrpc3 run orders:app --workers 4   # one process per core

And then: one flag, 14× more calls per core

Everything above runs on grpc.aio, the standard asyncio server. It is solid, but on one CPU core it tops out around 4,400 calls per second, and most of that time is spent in the transport, not in your code.

So fastgrpc3 ships a second transport, written in Rust on tonic: HTTP/2, gRPC framing, TLS and flow control run in Rust, and your Python handlers receive the calls in batches. Same handlers, same dependencies, same errors:

pip install "fastgrpc3[tonic]"
fastgrpc3 run orders:app --transport tonic

On one core, unary calls, 64 concurrent clients:

	calls/s	CPU per call	p99
fastgrpc3 on tonic	62,240	13 µs	2.0 ms
fastgrpc3 on `grpc.aio`	4,430	226 µs	15.9 ms
`grpc.aio` + generated stubs, no framework	4,571	219 µs	18.0 ms

That is 14× the throughput, 17× less CPU per call and an 8× lower tail latency — and the 62,240 is the load generator's ceiling, not the server's. On grpc.aio, the framework itself costs about 3 % over hand-written stubs. How the Rust transport gets there is a story of its own, which I tell in a separate article.

Under the hood: protoruf

None of this needs generated _pb2 classes because the protobuf layer is protoruf: it compiles .proto files in memory and its codec goes straight from protobuf bytes to Pydantic models, in Rust. Encoding a 20-field message takes about 1 µs — 55× faster than going through google.protobuf and Pydantic.

Try it

pip install "fastgrpc3[tonic]"

Docs: https://edwinalkins.github.io/fastgrpc3/
GitHub: https://github.com/EdwinAlkins/fastgrpc3 (and the transport: https://github.com/EdwinAlkins/fastgrpc3-transport)

fastgrpc3 is at version 0.1: if you build something with it, or if it does not fit your use case, open an issue and tell me — and a ⭐ on GitHub helps other Python developers find it.

Protobuf without `protoc`: the ultimate developer experience for Python

William — Wed, 17 Jun 2026 21:51:15 +0000

Install protoc, generate _pb2.py files, juggle Parse and SerializeToString… Sounds familiar? With protoruf, all of that disappears. On-the-fly compilation, zero generated files, and native Pydantic integration. Here's how I swapped the complexity of google.protobuf for absolute simplicity.

If you've ever used Protobuf in Python, you know the routine:

Manually install protoc (and pray the version matches everywhere, especially in CI).
Generate Python code with protoc --python_out=., then version or regenerate that file every time the schema changes.
Write procedural code full of Parse, SerializeToString, MessageToJson calls, with no native data validation.

It works, but it's clunky. And every schema change means repeating the whole ritual.

Protoruf was built to fix this at the root. It's a Python library written in Rust that compiles .proto files in memory, converts data in a single line, and integrates naturally with Pydantic for validation.

The result? A radically smoother developer experience, where you can modify your Protobuf schema and keep coding immediately — without ever leaving your editor.

Before: the classic `google.protobuf` workflow

# Step 1 – Install protoc (a CI nightmare)
sudo apt-get install protobuf-compiler   # Linux
brew install protobuf                    # macOS
# Windows: download an .exe, add to PATH…

# Step 2 – Generate static Python code
protoc --python_out=. message.proto
# → A message_pb2.py file appears.
# You must version it or regenerate it on every build.

# Step 3 – Write the code
from google.protobuf import json_format
from message_pb2 import Message  # The generated file

msg = Message()
json_str = '{"id": "123", "content": "Hello"}'
json_format.Parse(json_str, msg)           # Parse JSON → Protobuf

binary = msg.SerializeToString()           # Serialize

msg2 = Message()
msg2.ParseFromString(binary)               # Deserialize
result = json_format.MessageToJson(msg2)   # Back to JSON

Pain points:

Dependency on protoc (manual installation, version mismatches).
Mandatory code generation → files to version or integrate into builds.
Schema change = regenerate everywhere.
No upfront data validation (everything is plain dict).

After: the `protoruf` simplicity

# A single command, end of story.
pip install protoruf
# No protoc, no external compiler.

from protoruf import compile_proto, json_to_protobuf, protobuf_to_json

# 1. Built-in compilation, in memory (Rust)
descriptor = compile_proto("message.proto")

# 2. JSON → Protobuf in 1 line
json_str = '{"id": "123", "content": "Hello"}'
binary = json_to_protobuf(json_str, descriptor, message_type="message.Message")

# 3. Protobuf → JSON in 1 line
result = protobuf_to_json(binary, descriptor, message_type="message.Message", pretty=True)
print(result)

What changes:

No more protoc — everything is bundled.
Zero generated files → your repo stays clean.
Schema update? Just re-run your script.
Fewer lines, less verbosity, more clarity.

Bonus: native Pydantic integration

protoruf provides direct functions to convert a Pydantic model to a Protobuf message and back. No more manual dict fiddling.

from pydantic import BaseModel
from protoruf import pydantic_to_protobuf, protobuf_to_pydantic

class Message(BaseModel):
    id: str
    content: str

msg = Message(id="123", content="Hello")

# Direct conversion, no manual JSON dump
binary = pydantic_to_protobuf(msg, descriptor, "message.Message")

# Back to a Pydantic instance
result = protobuf_to_pydantic(binary, descriptor, Message, "message.Message")
print(result.content)  # Hello

You get automatic validation from Pydantic both on input and output, with zero extra code.

Bonus 2: Blazing fast with DescriptorCache

Because protoruf is built on Rust (prost-reflect), it is incredibly fast. But for high-throughput workloads (like processing thousands of messages in a loop), protoruf offers a DescriptorCache.

By decoding the descriptor pool once and reusing it, protoruf achieves 5x to 10x faster serialization/deserialization compared to google.protobuf.

from protoruf import compile_proto, DescriptorCache

# Decode the pool ONCE
cache = DescriptorCache(compile_proto("message.proto"))

# Reuse it everywhere (Thread-safe!)
for json_str in massive_data_stream:
    binary = cache.json_to_protobuf(json_str, "message.Message")
    # Process binary...

Benchmarks

Transparency matters, so here are the numbers from my local machine (100,000 messages, same payload for both libraries).

Simple usage (free functions, no cache)

This is the "just call json_to_protobuf" approach — no setup, no cache. The descriptor is decoded on every call.

Operation	`google.protobuf`	`protoruf` (free functions)
Serialization (JSON→Proto)	1.83 s (54.7k msg/s)	2.07 s (48.2k msg/s)
Parsing (Proto→JSON)	1.62 s (61.8k msg/s)	1.88 s (53.2k msg/s)

In the simplest usage, protoruf is about 13–16% slower because it decodes the descriptor set from scratch on every call. That's the trade-off for not needing code generation upfront.

Hot loop with DescriptorCache

Now the same conversion, but with the descriptor pool decoded once outside the loop (just like you'd do in a real service).

Operation	`google.protobuf`	`protoruf` (DescriptorCache)
Serialization (JSON→Proto)	1.89 s (53.0k msg/s)	0.36 s (280k msg/s)
Parsing (Proto→JSON)	1.65 s (60.5k msg/s)	0.16 s (612k msg/s)

With the cache, protoruf is 5.3× faster on writes and 10.1× faster on reads. That's the power of decoding once and reusing — and it's the recommended setup for any service processing more than a handful of messages.

The best part? You get both modes with the same library. Start simple, and when you need the extra throughput, just wrap your descriptor in DescriptorCache.

Why the developer experience is transformed

Aspect	`google.protobuf`	`protoruf`
Installation	`protoc` + `pip`	`pip install protoruf`
Code generation	Yes (`_pb2.py` files)	None (in-memory compilation)
Schema changes	Regenerate everywhere, redeploy	Do nothing, just re-run
CI / CD	Nightmare (`protoc` versioning)	Zero configuration
Data validation	Manual	Native Pydantic integration
Lines of code	~15 lines	~5 lines
Performance (Hot loops)	Baseline	Up to 10x faster (with Cache)

With protoruf, the time you used to spend configuring protoc and maintaining generated files goes directly into your real code. Changing a Protobuf schema is no longer a chore — it's just another line of code.

Try it in 30 seconds

pip install protoruf

Create a message.proto:

syntax = "proto3";
package message;
message Message {
  string id = 1;
  string content = 2;
  int32 priority = 3;
}

Then run:

from protoruf import compile_proto, json_to_protobuf, protobuf_to_json

descriptor = compile_proto("message.proto")

pb = json_to_protobuf('{"id":"42","content":"Hello"}', descriptor, "message.Message")
print(protobuf_to_json(pb, descriptor, "message.Message", pretty=True))

No hidden steps, no stray files. You code, it compiles.

In summary

Protoruf is the choice of productivity and peace of mind for all your Protobuf exchanges in Python.

No more protoc, no more generated files, no more verbose gymnastics.

You keep the power of Protobuf, with a Pythonic API and instant Pydantic integration.

🔗 Links:

📖 Documentation: http://edwinalkins.github.io/protoruf
💻 Source Code: http://github.com/EdwinAlkins/protoruf
📦 PyPI: http://pypi.org/project/protoruf

Feedback, issues, and PRs are more than welcome! Let me know in the comments what you think.