基于 WSGI/ASGI 双协议的 rpc 框架

[abersheeran]

整了一个 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