pydanticrpc , perpustakaan Python mengautomasikan penciptaan perkhidmatan GRPC dan menyambung RPC dari model Pydantic. Tiada lagi manual. Penciptaan fail proto!
github - pydanticrpc
Gambaran Keseluruhan
Python Rest Apis sering memanfaatkan kerangka seperti Fastapi atau Flask. Walau bagaimanapun, untuk pemindahan data yang dioptimumkan atau pendekatan skema-pertama, GRPC atau Connect RPC adalah alternatif yang menarik. Aliran kerja RPC tradisional melibatkan mendefinisikan fail proto, penjanaan kod (menggunakan atau protoc
), dan integrasi -proses yang sering rumit dan menuntut. buf
pydanticrpc Streamlines ini. Tentukan struktur data RPC anda dengan model Pydantic; PydanticRPC secara dinamik menjana definisi protobuf dan melancarkan pelayan.
Apakah pydanticrpc?
Ciri -ciri Utama:
grpcio-tools
Pasang melalui PYPI:
pip install pydantic-rpc
Gunakan
untuk membuat pelayan GRPC.
pydantic_rpc.Server
Contoh Synchronous
# 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
mendedahkan kaedahnya. pydantic.BaseModel
menghasilkan fail .proto dan memulakan pelayan GRPC (localhost: 50051 secara lalai). Greeter
Server().run(Greeter())
contoh asynchronous
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())
pydanticrpc menyokong respons streaming pelayan (kini Async GRPC sahaja). Contoh di bawah menggunakan
untuk trivia Olimpik, mempamerkan kaedah standard dan streaming:
pydantic_ai
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()))
adalah pelayan-streaming, menghasilkan hasil secara berperingkat. Pydanticrpc menjana fail .proto yang menentukan kedua -duanya, melancarkan pelayan GRPC async. ask
ask_stream
Bersepadu dengan Connecpy untuk Connect RPC dalam aplikasi ASGI: Pydantic mengendalikan pengesahan. Mengintegrasikan ini Penggunaan: Membuat Perkhidmatan GRPC-WEB Hidangkan GRPC-WEB dalam aplikasi WSGI atau ASGI: wujud bersama GRPC-WEB dan REST endpoints. Kesimpulan PydanticRPC memudahkan GRPC, GRPC-WEB, dan menghubungkan pembangunan RPC dari model Pydantic, termasuk penyejuk pelayan. Terokai repositori GitHub Pydanticrpc untuk maklumat lanjut. Maklum balas selamat datang! Atas ialah kandungan terperinci Hello Dev Community! Memperkenalkan PydanticRPC: Membina perkhidmatan RPC GRPC & Sambungkan Tanpa Menulis Fail Protobuf secara manual. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!pip install pydantic-rpc
app
ke dalam rangka kerja ASGI sedia ada (Fastapi, Starlette).
# 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())