fastgrpc2 0.3.1

FastAPI-like developer experience for gRPC: pydantic-first, async, built on grpc.aio

fastgrpc2

A FastAPI-style developer experience for gRPC — pydantic-first, async, and typed.

Write an async handler that takes a plain Pydantic request and returns a Pydantic response. fastgrpc2 derives the protobuf contract from your annotations, (de)serializes over real gRPC/HTTP-2, injects your dependencies, and never makes you touch grpc.aio, a _pb2 stub, or a hand-written .proto. It's a thin layer on top of the grpc.aio C-core.

Documentation: https://fastgrpc.ru

Install

pip install fastgrpc2

Requires Python 3.10+. Pulls in grpcio, protobuf, and pydantic.

Example

Server:

from pydantic import BaseModel
from fastgrpc2 import GrpcApp, GrpcService, Context, GrpcError, StatusCode

class GetOrderRequest(BaseModel):
    order_number: str

class OrderReply(BaseModel):
    order_number: str
    status: str

service = GrpcService("Orders")

@service.unary("GetOrder")
async def get_order(request: GetOrderRequest, ctx: Context) -> OrderReply:
    if not request.order_number:
        raise GrpcError(StatusCode.INVALID_ARGUMENT, "order_number required")
    return OrderReply(order_number=request.order_number, status="paid")

app = GrpcApp("orders", address="0.0.0.0:50051")
app.add_service(service)
# import asyncio; asyncio.run(app.run())

Client:

from fastgrpc2 import GrpcClient, Channel

class OrderClient(GrpcClient):
    service = "Orders"
    get_order = GrpcClient.unary("GetOrder", GetOrderRequest, OrderReply)

async with Channel("127.0.0.1:50051", default_timeout=5.0) as ch:
    client = OrderClient(ch)
    reply = await client.get_order(GetOrderRequest(order_number="A-100"))
    print(reply.status)  # -> "paid"

Features

• All four call types — @service.unary / unary_stream / stream_unary / stream_stream, inferred from the handler signature.
• Pydantic ↔ protobuf — scalars, nested models, list, Optional (proto3 presence), Enum, datetime/date; anything without a clean proto mapping falls back to a JSON carrier, so a model never fails to register. app.export_proto() dumps the .proto.
• Context — metadata, deadline, trailers, and peer, kept out of the message body.
• Errors — GrpcError + the full StatusCode enum + rich google.rpc-style details.
• Client — a reusable Channel and a typed GrpcClient; per-call timeout / metadata.
• Security — TLS / MTLS / Insecure builders and BearerToken.
• DI & middleware — Depends with dependency_overrides, and interceptors.
• Ops — enable_health() (grpc.health.v1), graceful stop(grace), lifespan, Compression, message-size limits, RetryPolicy, Keepalive, and deadline propagation across service hops.
• Testing — an in-process TestClient.

Performance

fastgrpc2 is a thin layer: the transport is the gRPC C-core, the codec is protobuf upb (C), and validation is pydantic-core (Rust). The framework adds single-digit % over a hand-written grpc.aio servicer running the identical handler. Its ceiling is Python's grpc.aio — ergonomics at low cost, not the throughput of compiled languages. See the benchmarks.

License

MIT © Ruslan Kiradiev