Hello DEV Community! Introducing PydanticRPC: Build gRPC & Connect RPC Services Without Manually Writing Protobuf Files

This is my inaugural DEV post, introducing PydanticRPC, a Python library automating the creation of gRPC and Connect RPC services from Pydantic models. No more manual .proto file creation!

GitHub - PydanticRPC

---

Overview

Python REST APIs often leverage frameworks like FastAPI or Flask. However, for optimized data transfer or a schema-first approach, gRPC or Connect RPC are compelling alternatives. Traditional RPC workflows involve defining .proto files, code generation (using protoc or buf), and integration—a process often cumbersome and demanding.

PydanticRPC streamlines this. Define your RPC data structures with Pydantic models; PydanticRPC dynamically generates Protobuf definitions and launches the server.

---

What is PydanticRPC?

Key features:

• Automated Protobuf Generation: Creates .proto files directly from your Python classes and Pydantic models.
• Dynamic Code Generation: Uses grpcio-tools to generate server/client stubs and integrates your Python classes seamlessly.
• Broad Support: Compatible with gRPC, gRPC-Web (via Sonora), Connect RPC (via Connecpy), and asynchronous (asyncio) operations, including server-streaming methods.

Essentially: "Define a Python class with Pydantic models, and instantly get an RPC service—no .proto files required!"

---

Installation

Install via PyPI:

<code class="language-bash">pip install pydantic-rpc</code>

---

Usage: Creating a gRPC Service

Use pydantic_rpc.Server to create a gRPC server.

Synchronous Example

<code class="language-python"># server.py
from pydantic_rpc import Server, Message

class HelloRequest(Message):
    name: str

class HelloReply(Message):
    message: str

class Greeter:
    def say_hello(self, request: HelloRequest) -> HelloReply:
        return HelloReply(message=f"Hello, {request.name}!")

if __name__ == "__main__":
    server = Server()
    server.run(Greeter())</code>

Message aliases pydantic.BaseModel. Greeter exposes its methods. Server().run(Greeter()) generates the .proto file and starts a gRPC server (localhost:50051 by default).

Asynchronous Example

For async servers, use AsyncIOServer:

<code class="language-python">import asyncio
from pydantic_rpc import AsyncIOServer, Message

class HelloRequest(Message):
    name: str

class HelloReply(Message):
    message: str

class Greeter:
    async def say_hello(self, request: HelloRequest) -> HelloReply:
        return HelloReply(message=f"Hello, {request.name}!")

if __name__ == "__main__":
    server = AsyncIOServer()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(server.run(Greeter()))</code>

server.run(Greeter()) is a coroutine, run within your event loop.

---

Usage: Response Streaming

PydanticRPC supports server-streaming responses (currently async gRPC only). The example below uses pydantic_ai for Olympic trivia, showcasing both standard and streaming methods:

<code class="language-python">import asyncio
from typing import AsyncIterator

# ... (imports and class definitions as shown in the original) ...

if __name__ == "__main__":
    s = AsyncIOServer()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(s.run(OlympicsAgent()))</code>

ask is a unary RPC; ask_stream is server-streaming, yielding results incrementally. PydanticRPC generates a .proto file defining both, launching an async gRPC server.

---

Usage: Creating a Connect RPC Service

Integrate with Connecpy for Connect RPC in an ASGI app:

<code class="language-bash">pip install pydantic-rpc</code>

Pydantic handles validation. Integrate this app into your existing ASGI framework (FastAPI, Starlette).

---

Usage: Creating a gRPC-Web Service

Serve gRPC-Web in WSGI or ASGI applications:

<code class="language-python"># server.py
from pydantic_rpc import Server, Message

class HelloRequest(Message):
    name: str

class HelloReply(Message):
    message: str

class Greeter:
    def say_hello(self, request: HelloRequest) -> HelloReply:
        return HelloReply(message=f"Hello, {request.name}!")

if __name__ == "__main__":
    server = Server()
    server.run(Greeter())</code>

Coexist gRPC-Web and REST endpoints.

---

Conclusion

PydanticRPC simplifies gRPC, gRPC-Web, and Connect RPC development from Pydantic models, including server-streaming. Explore the PydanticRPC GitHub repository for more details. Feedback welcome!

The above is the detailed content of Hello DEV Community! Introducing PydanticRPC: Build gRPC & Connect RPC Services Without Manually Writing Protobuf Files. For more information, please follow other related articles on the PHP Chinese website!