整了一个 rpc 框架,支持普通函数、生成器函数以及它们的异步版本。
https://github.com/abersheeran/rpc.py
感觉好像不缺什么了?欢迎大家来用。
Server side:
import uvicorn
from rpcpy import RPC
app = RPC(mode="ASGI")
@app.register
async def none() -> None:
return
@app.register
async def sayhi(name: str) -> str:
return f"hi {name}"
@app.register
async def yield_data(max_num: int):
for i in range(max_num):
yield i
if __name__ == "__main__":
uvicorn.run(app, interface="asgi3", port=65432)
OR
import uvicorn
from rpcpy import RPC
app = RPC()
@app.register
def none() -> None:
return
@app.register
def sayhi(name: str) -> str:
return f"hi {name}"
@app.register
def yield_data(max_num: int):
for i in range(max_num):
yield i
if __name__ == "__main__":
uvicorn.run(app, interface="wsgi", port=65432)
Client side:
import httpx
from rpcpy.client import Client
app = Client( httpx.Client(), base_url="http://127.0.0.1:65432/")
@app.remote_call
def none() -> None:
...
@app.remote_call
def sayhi(name: str) -> str:
...
@app.remote_call
def yield_data(max_num: int):
yield
OR
import httpx
from rpcpy.client import Client
app = Client( httpx.AsyncClient(), base_url="http://127.0.0.1:65432/")
@app.remote_call
async def none() -> None:
...
@app.remote_call
async def sayhi(name: str) -> str:
...
@app.remote_call
async def yield_data(max_num: int):
yield