pjrpc 1.1.1

eh3客户端请求>

使用pjrpc客户机的方法非常简单直观。方法可以使用代理对象按名称调用 或者发送手工制作的pjrpc.common.Requestclass对象。通知请求可以使用 pjrpc.client.AbstractClient.notify方法或发送一个没有id的pjrpc.common.Request对象

^{pr2}$

异步客户端api看起来几乎相同：

importpjrpcfrompjrpc.client.backendimportaiohttpaspjrpc_clientclient=pjrpc_client.Client('http://localhost/api/v1')response=awaitclient.send(pjrpc.Request('sum',params=[1,2],id=1))print(f"1 + 2 = {response.result}")result=awaitclient('sum',a=1,b=2)print(f"1 + 2 = {result}")result=awaitclient.proxy.sum(1,2)print(f"1 + 2 = {result}")awaitclient.notify('tick')

批处理请求

也支持批处理请求。您可以手工生成pjrpc.common.BatchRequest请求，然后将其发送到 服务器。结果是一个pjrpc.common.BatchResponse实例，您可以迭代以获得所有结果或 每个索引：

importpjrpcfrompjrpc.client.backendimportrequestsaspjrpc_clientclient=pjrpc_client.Client('http://localhost/api/v1')batch_response=awaitclient.batch.send(pjrpc.BatchRequest(pjrpc.Request('sum',[2,2],id=1),pjrpc.Request('sub',[2,2],id=2),pjrpc.Request('div',[2,2],id=3),pjrpc.Request('mult',[2,2],id=4),))print(f"2 + 2 = {batch_response[0].result}")print(f"2 - 2 = {batch_response[1].result}")print(f"2 / 2 = {batch_response[2].result}")print(f"2 * 2 = {batch_response[3].result}")

对于第一种方法，也有几种替代方法（注意，结果是 不再是pjrpc.common.BatchResponse对象，而是“普通”方法调用结果的元组）：

• 使用链调用表示法：

result=awaitclient.batch('sum',2,2)('sub',2,2)('div',2,2)('mult',2,2).call()print(f"2 + 2 = {result[0]}")print(f"2 - 2 = {result[1]}")print(f"2 / 2 = {result[2]}")print(f"2 * 2 = {result[3]}")

• 使用订阅运算符：

result=awaitclient.batch[('sum',2,2),('sub',2,2),('div',2,2),('mult',2,2),]print(f"2 + 2 = {result[0]}")print(f"2 - 2 = {result[1]}")print(f"2 / 2 = {result[2]}")print(f"2 * 2 = {result[3]}")

• 使用代理链调用：

result=awaitclient.batch.proxy.sum(2,2).sub(2,2).div(2,2).mult(2,2).call()print(f"2 + 2 = {result[0]}")print(f"2 - 2 = {result[1]}")print(f"2 / 2 = {result[2]}")print(f"2 * 2 = {result[3]}")

使用哪一个由您决定，但是请注意，如果任何请求返回错误，则其他请求的结果 会迷路的。在这种情况下，可以使用第一种方法迭代所有响应并获得 像这样的成功者：

importpjrpcfrompjrpc.client.backendimportrequestsaspjrpc_clientclient=pjrpc_client.Client('http://localhost/api/v1')batch_response=client.batch.send(pjrpc.BatchRequest(pjrpc.Request('sum',[2,2],id=1),pjrpc.Request('sub',[2,2],id=2),pjrpc.Request('div',[2,2],id=3),pjrpc.Request('mult',[2,2],id=4),))forresponseinbatch_response:ifresponse.is_success:print(response.result)else:print(response.error)

批通知：

importpjrpcfrompjrpc.client.backendimportrequestsaspjrpc_clientclient=pjrpc_client.Client('http://localhost/api/v1')client.batch.notify('tick').notify('tack').notify('tick').notify('tack').call()

服务器

{a7}流行框架{a7}， flask和消息代理，如{a11} 和aio_pika。在

运行基于aiohttp的JSON-RPC服务器是一个非常简单的过程。只需定义方法，将它们添加到 注册并运行服务器：

importuuidfromaiohttpimportwebimportpjrpc.serverfrompjrpc.server.integrationimportaiohttpmethods=pjrpc.server.MethodRegistry()@methods.add(context='request')asyncdefadd_user(request:web.Request,user:dict):user_id=uuid.uuid4().hexrequest.app['users'][user_id]=userreturn{'id':user_id,**user}jsonrpc_app=aiohttp.Application('/api/v1')jsonrpc_app.dispatcher.add_methods(methods)jsonrpc_app.app['users']={}if__name__=="__main__":web.run_app(jsonrpc_app.app,host='localhost',port=8080)

参数验证

通常除了哑方法参数验证外，还需要实现更“深入”的验证并提供 向客户提供全面的错误描述。幸运的是，pjrpc基于 pydantic库，使用python类型注释进行验证。 看看下面的例子：所有你需要注释的方法参数（或者描述更复杂的类型，如果 必要的）。pjrpc将验证方法参数并向客户端返回信息错误。在

importenumimportuuidfromtypingimportListimportpydanticfromaiohttpimportwebimportpjrpc.serverfrompjrpc.server.validatorsimportpydanticasvalidatorsfrompjrpc.server.integrationimportaiohttpmethods=pjrpc.server.MethodRegistry()validator=validators.PydanticValidator()classContactType(enum.Enum):PHONE='phone'EMAIL='email'classContact(pydantic.BaseModel):type:ContactTypevalue:strclassUser(pydantic.BaseModel):name:strsurname:strage:intcontacts:List[Contact]@methods.add(context='request')@validator.validateasyncdefadd_user(request:web.Request,user:User):user_id=uuid.uuid4()request.app['users'][user_id]=userreturn{'id':user_id,**user.dict()}classJSONEncoder(pjrpc.common.JSONEncoder):defdefault(self,o):ifisinstance(o,uuid.UUID):returno.hexifisinstance(o,enum.Enum):returno.valuereturnsuper().default(o)jsonrpc_app=aiohttp.Application('/api/v1',json_encoder=JSONEncoder)jsonrpc_app.dispatcher.add_methods(methods)jsonrpc_app.app['users']={}if__name__=="__main__":web.run_app(jsonrpc_app.app,host='localhost',port=8080)

错误处理

pjrpc实现protocol specification中列出的所有错误 它可以在pjrpc.common.exceptions模块中找到，因此错误处理非常简单且“python方式”：

importpjrpcfrompjrpc.client.backendimportrequestsaspjrpc_clientclient=pjrpc_client.Client('http://localhost/api/v1')try:result=client.proxy.sum(1,2)exceptpjrpc.MethodNotFoundase:print(e)

默认错误列表可以很容易地扩展。创建继承自 pjrpc.exc.JsonRpcError，并定义错误代码和描述消息。pjrpc将自动 正在为您反序列化自定义错误：

importpjrpcfrompjrpc.client.backendimportrequestsaspjrpc_clientclassUserNotFound(pjrpc.exc.JsonRpcError):code=1message='user not found'client=pjrpc_client.Client('http://localhost/api/v1')try:result=client.proxy.get_user(user_id=1)exceptUserNotFoundase:print(e)

在服务器端，一切都非常简单：

importuuidimportflaskimportpjrpcfrompjrpc.serverimportMethodRegistryfrompjrpc.server.integrationimportflaskasintegrationapp=flask.Flask(__name__)methods=pjrpc.server.MethodRegistry()classUserNotFound(pjrpc.exc.JsonRpcError):code=1message='user not found'@methods.adddefadd_user(user:dict):user_id=uuid.uuid4().hexflask.current_app.users[user_id]=userreturn{'id':user_id,**user}@methods.adddefget_user(self,user_id:str):user=flask.current_app.users.get(user_id)ifnotuser:raiseUserNotFound(data=user_id)returnuserjson_rpc=integration.JsonRPC('/api/v1')json_rpc.dispatcher.add_methods(methods)app.users={}json_rpc.init_app(app)if__name__=="__main__":app.run(port=80)