abersheeran/rpc.py

A fast and powerful RPC framework based on ASGI/WSGI.

rpc.py

An fast and powerful RPC framework based on ASGI/WSGI. Based on WSGI/ASGI, you can deploy the rpc.py server to any server and use http2 to get better performance. And based on httpx's support for multiple http protocols, the client can also use http/1.0, http/1.1 or http2.

You can freely use ordinary functions and asynchronous functions for one-time response. You can also use generator functions or asynchronous generator functions to stream responses.

Install

Install from PyPi:

pip install rpc.py

# need use client
pip install rpc.py[client]

# need use pydantic type hint or OpenAPI docs
pip install rpc.py[type]

# need use msgpack to serializer
pip install rpc.py[msgpack]

# need use CBOR to serializer
pip install rpc.py[cbor]

# or install all dependencies
pip install rpc.py[full]

Install from github:

pip install git+https://github.com/abersheeran/[email protected]

Usage

Server side:

Use ASGI mode to register async def...

from typing import AsyncGenerator
from typing_extensions import TypedDict

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) -> AsyncGenerator[int, None]:
    for i in range(max_num):
        yield i


D = TypedDict("D", {"key": str, "other-key": str})


@app.register
async def query_dict(value: str) -> D:
    return {"key": value, "other-key": value}


if __name__ == "__main__":
    uvicorn.run(app, interface="asgi3", port=65432)

OR

Use WSGI mode to register def...

from typing import Generator
from typing_extensions import TypedDict

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) -> Generator[int, None, None]:
    for i in range(max_num):
        yield i


D = TypedDict("D", {"key": str, "other-key": str})


@app.register
def query_dict(value: str) -> D:
    return {"key": value, "other-key": value}


if __name__ == "__main__":
    uvicorn.run(app, interface="wsgi", port=65432)

Client side:

Notice: Regardless of whether the server uses the WSGI mode or the ASGI mode, the client can freely use the asynchronous or synchronous mode.

Use httpx.Client() mode to register def...

from typing import Generator
from typing_extensions import TypedDict

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) -> Generator[int, None, None]:
    yield


D = TypedDict("D", {"key": str, "other-key": str})


@app.remote_call
def query_dict(value: str) -> D:
    ...

OR

Use httpx.AsyncClient() mode to register async def...

from typing import AsyncGenerator
from typing_extensions import TypedDict

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) -> AsyncGenerator[int, None]:
    yield


D = TypedDict("D", {"key": str, "other-key": str})


@app.remote_call
async def query_dict(value: str) -> D:
    ...

Server as client

You can also write two copies of code in one place. Just make sure that server.register is executed before client.remote_call.

import httpx
from rpcpy import RPC
from rpcpy.client import Client

server = RPC()
client = Client(httpx.Client(), base_url="http://127.0.0.1:65432/")


@client.remote_call
@server.register
def sayhi(name: str) -> str:
    return f"hi {name}"


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, interface="wsgi", port=65432)

Sub-route

If you need to deploy the rpc.py server under example.com/sub-route/*, you need to set RPC(prefix="/sub-route/") and modify the Client(base_path=https://example.com/sub-route/).

Serialization

Currently supports three serializers, JSON, Msgpack and CBOR. JSON is used by default. You can override the default JSONSerializer with parameters.

from rpcpy.serializers import MsgpackSerializer, CBORSerializer

RPC(
    ...,
    response_serializer=MsgpackSerializer(),
)
# Or
Client(
    ...,
    request_serializer=CBORSerializer(),
)

Type hint and OpenAPI Doc

Thanks to the great work of pydantic, which makes rpc.py allow you to use type annotation to annotate the types of function parameters and response values, and perform type verification and JSON serialization . At the same time, it is allowed to generate openapi documents for human reading.

OpenAPI Documents

If you want to open the OpenAPI document, you need to initialize RPC like this RPC(openapi={"title": "TITLE", "description": "DESCRIPTION", "version": "v1"}).

Then, visit the "{prefix}openapi-docs" of RPC and you will be able to see the automatically generated OpenAPI documentation. (If you do not set the prefix, the prefix is "/")

Limitations

Currently, file upload is not supported, but you can do this by passing a bytes object.

rpc.py's Issues

How to fix userwarning

UserWarning: Unclosed <httpx.AsyncClient object at 0x10ed159d0>. See https://www.python-httpx.org/async/#opening-and-closing-clients for details.

versions:
Successfully installed httpcore-0.12.3 httpx-0.16.1 msgpack-1.0.2 pydantic-1.7.3 rfc3986-1.4.0 rpc.py-0.4.3 sniffio-1.2.0

---

use https://github1s.com/abersheeran/rpc.py/blob/master/examples/async_client.py and async_server.py
my script:

from rpcclient import sayhi
import asyncio


async def run():
    result = await sayhi("quantex")
    print(result)
    return result


if __name__ == '__main__':
    asyncio.run(run())

openapi-docs怎么打不开呢

服务端分多个文件可以吗

不同文件对应一个sub route

可以定义客户端的错误提示吗

目前客户端调用函数时，如果参数类型不对，提示如下:

httpx.HTTPStatusError: Server error '500 Internal Server Error' for url 'http://127.0.0.1:65432/sayhi'
For more information check: https://httpstatuses.com/500

错误提示可否像服务端一样明确一点:

pydantic.error_wrappers.ValidationError: 1 validation error for sayhi
name
  value is not a valid integer (type=type_error.integer)

Unauthenticated Remote Code Execution vulnerability

Unfortunately, I have not received any replies from the maintainer within a time frame of two weeks, so I am disclosing this vulnerability that is still existent in the latest version, with no patch available:

https://medium.com/@elias.hohl/remote-code-execution-0-day-in-rpc-py-709c76690c30

Support CBOR serializer

Related links: https://cbor2.readthedocs.io/en/latest/usage.html

Add EventResponse to support Generator/Asynchronous Generator

example

@app.register
def generator_function():
    yield 1
    yield 2

@app.register
async def generator_function():
    yield 1
    yield 2

[Security] Workflow test.yml is using vulnerable action actions/checkout

The workflow test.yml is referencing action actions/checkout using references v1. However this reference is missing the commit a6747255bd19d7a757dbdda8c654a9f84db19839 which may contain fix to the some vulnerability.
The vulnerability fix that is missing by actions version could be related to:
(1) CVE fix
(2) upgrade of vulnerable dependency
(3) fix to secret leak and others.
Please consider to update the reference to the action.

Need more test, coverage -> 100%

install the packets

are you sure we can install ... like this ?

pip install rpc.py[full]

zsh: no matches found: rpc.py[full]