Bonjour la communauté Dev! Présentation de PyndanticRPC: Créer des services GRPC et connecter RPC sans écrire manuellement les fichiers Protobuf

Ceci est mon post inaugural de DEV, introduisant pydanticrpc , une bibliothèque Python automatisant la création de services RPC GRPC et connectait des modèles pydantes. Plus de manuelle .proto Création de fichiers!

github - pydanticrpc

---

Présentation

Les API de repos python exploitent souvent des cadres comme Fastapi ou Flask. Cependant, pour un transfert de données optimisé ou une approche d'abord en schéma, le GRPC ou le RPC connectent sont des alternatives convaincantes. Les flux de travail RPC traditionnels impliquent de définir des fichiers .proto, une génération de code (en utilisant protoc ou buf) et une intégration - un processus souvent lourd et exigeant.

pydanticrpc rationalise cela. Définissez vos structures de données RPC avec des modèles pydantiques; Pydanticrpc génère dynamiquement des définitions de protobuf et lance le serveur.

---

Qu'est-ce que pydanticrpc?

Caractéristiques clés:

• Génération de protobuf automatisée: Crée des fichiers .proto directement à partir de vos classes Python et de vos modèles pydatiques.
• Génération de code dynamique: utilise grpcio-tools pour générer des talons de serveur / client et intègre vos classes Python de manière transparente.
• Support large: Compatible avec GRPC, GRPC-WEB (via Sonora), connecter RPC (via Connecpy) et les opérations asynchrones (asyncio), y compris les méthodes de diffusion de serveurs.

Essentiellement: "Définissez une classe Python avec des modèles pydantiques et obtenez instantanément un service RPC - aucun fichier. Proto requis!"

---

Installation

Installer via PYPI:

pip install pydantic-rpc

---

Utilisation: création d'un service GRPC

Utilisez pydantic_rpc.Server pour créer un serveur GRPC.

Exemple synchrone

# 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())

Message alias pydantic.BaseModel. Greeter expose ses méthodes. Server().run(Greeter()) génère le fichier .proto et démarre un serveur GRPC (localhost: 50051 par défaut).

Exemple asynchrone

pour les serveurs asynchrones, utilisez AsyncIOServer:

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()))

server.run(Greeter()) est une coroutine, exécutez dans votre boucle d'événement.

---

Utilisation: Réponse Streaming

pyndanticrpc prend en charge les réponses de diffusion de serveurs (actuellement Async Grpc uniquement). L'exemple ci-dessous utilise pydantic_ai pour les anecdotes olympiques, présentant des méthodes standard et de streaming:

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()))

ask est un RPC unaire; ask_stream est la diffusion de serveurs, ce qui donne des résultats progressivement. Pydanticrpc génère un fichier .proto définissant les deux, lançant un serveur GRPC asynchrone.

---

Utilisation: Création d'un service RPC Connect

Intégrez à Connecpy pour Connect RPC dans une application ASGI:

pip install pydantic-rpc

Validation des manches pydantiques. Intégrez ceci app dans votre framework ASGI existant (Fastapi, Starlette).

---

Utilisation: Création d'un service GRPC-Web

Servir GRPC-Web dans les applications WSGI ou ASGI:

# 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())

coexiste GRPC-Web et points de terminaison de repos.

---

Conclusion

pydanticrpc simplifie le développement de GRPC, GRPC-Web et Connect RPC à partir de modèles pydantiques, y compris la diffusion de serveurs. Explorez le référentiel Github PydanticRPC pour plus de détails. Retour bienvenue!

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!