python?json-rpc?規(guī)范源碼閱讀
json-rpc 源碼閱讀
JSON-RPC是一個無狀態(tài)且輕量級的遠(yuǎn)程過程調(diào)用(RPC)協(xié)議。JSON-RPC應(yīng)用很廣泛,比如以太坊的API。JSON-RPC的python實現(xiàn)較多,我選擇了Exploding Labs 提供的python版本。主要是其它庫都比較古老,而e-labs的實現(xiàn)采用最新版本python,支持類型系統(tǒng),還有一些函數(shù)式編程的范式,代碼也很簡潔,值得學(xué)習(xí)。
e-labs的JSON-RPC分成客戶端和服務(wù)端兩個庫,分別是jsonrpcclient和jsonrpcserver, 代碼版本如下表:
名稱 | 版本 |
---|---|
jsonrpcclient | 4.0.2 |
jsonrpcserver | 5.0.9 |
準(zhǔn)備好代碼后,我們可以開始json-rpc的源碼閱讀,本文包括下面幾個部分:
- JSON-RPC規(guī)范
- jsonrpcclient的實現(xiàn)
- jsonrpcserver的實現(xiàn)
- 小結(jié)
- 小技巧
JSON-RPC規(guī)范
JSON-RPC規(guī)范,我這里借用jsonrpcserver中的驗證規(guī)則文件簡單介紹一下,文件如下:
# request-schema.json { "$schema": "http://json-schema.org/draft-04/schema#", "description": "A JSON RPC 2.0 request", "oneOf": [ { "description": "An individual request", "$ref": "#/definitions/request" }, { "description": "An array of requests", "type": "array", "items": { "$ref": "#/definitions/request" }, "minItems": 1 } ], "definitions": { "request": { "type": "object", "required": [ "jsonrpc", "method" ], "properties": { "jsonrpc": { "enum": [ "2.0" ] }, "method": { "type": "string" }, "id": { "type": [ "string", "number", "null" ], "note": [ "While allowed, null should be avoided: http://www.jsonrpc.org/specification#id1", "While allowed, a number with a fractional part should be avoided: http://www.jsonrpc.org/specification#id2" ] }, "params": { "type": [ "array", "object" ] } }, "additionalProperties": false } } }
文件描述了JSON-RPC的規(guī)則,如下:
- json-rpc請求可以是單個的request對象,也是是批量的request對象數(shù)組
- 每個request對象需要符合:
- 必填字段jsonrpc,值枚舉類型。目前2.0,其實就是版本號。(之前有1.0版本)
- 必填字段method, 字符串類型。遠(yuǎn)程函數(shù)的名稱。
- id字段,支持字符串,數(shù)字或者空。為空表示通知無需回應(yīng)(result)。id確保響應(yīng)可以一一對應(yīng)到請求上。
- params字段,支持?jǐn)?shù)組或者字典。
JSON-RPC響應(yīng)部分的規(guī)則是:
- jsonrpc字段,值為2.0
- result字段,值為調(diào)用結(jié)果
- error字段,值為異常信息,包括code,message和data三個字段,規(guī)范定義了詳細(xì)的錯誤清單。
- id同請求的id
- result和error二選一
強烈建議大家閱讀參考鏈接中的規(guī)范原文,介紹的非常清晰,中文翻譯也很到位,有助于對JSON-RPC規(guī)范完全理解。
jsonrpcclient的實現(xiàn)
模塊文件 | 功能描述 |
---|---|
id_generators.py | id生成器 |
requests.py | 請求信息封裝 |
response.py | 響應(yīng)信息封裝 |
sentinels.py | 定義NOID,用于通知類請求 |
utils.py | 一些工具函數(shù) |
examples | 一些示例 |
從示例可以知道JSON-RPC,可以使用不同的底層協(xié)議比如http,websocket和tcp(zeromq實現(xiàn))等。我們看最簡單的基于http實現(xiàn)的實例:
from jsonrpcclient import request, parse, Ok import logging import requests response = requests.post("http://localhost:5000/", json=request("ping")) parsed = parse(response.json()) if isinstance(parsed, Ok): print(parsed.result) else: logging.error(parsed.message)
這段api展示了:
- jsonrpcclient只是封裝請求request和響應(yīng)Ok,數(shù)據(jù)請求的發(fā)送由不同協(xié)議提供,這里使用requests,另外還有aiohttp的實現(xiàn)等。
- resquest函數(shù)封裝請求,parse解析響應(yīng)
- 正常的結(jié)果展示result信息,錯誤的結(jié)果展示message信息
request代碼很簡單, 封裝請求成符合JSON-RPC規(guī)范的字符串:
# requests.py def request_pure( id_generator: Iterator[Any], method: str, params: Union[Dict[str, Any], Tuple[Any, ...]], id: Any, ) -> Dict[str, Any]: return { "jsonrpc": "2.0", "method": method, **( {"params": list(params) if isinstance(params, tuple) else params} if params else {} ), "id": id if id is not NOID else next(id_generator), } def request_impure( id_generator: Iterator[Any], method: str, params: Union[Dict[str, Any], Tuple[Any, ...], None] = None, id: Any = NOID, ) -> Dict[str, Any]: return request_pure( id_generator or id_generators.decimal(), method, params or (), id ) request_natural = partial(request_impure, id_generators.decimal()) ... request = request_natural
所以示例中的請求,可以等價下面的curl命令:
$ curl -X POST http://localhost:5001 -d '{"jsonrpc": "2.0", "method": "ping", "params": {}, "id": 1}'
response處理也很簡單:
# response.py class Ok(NamedTuple): result: Any id: Any def __repr__(self) -> str: return f"Ok(result={self.result!r}, id={self.id!r})" class Error(NamedTuple): code: int message: str data: Any id: Any def __repr__(self) -> str: return f"Error(code={self.code!r}, message={self.message!r}, data={self.data!r}, id={self.id!r})" Response = Union[Ok, Error]
定義Response類型,是Ok或者Error。Ok和Error是兩個可命名元祖。
parse就是將結(jié)果json字典解析成對應(yīng)的Response:
def to_result(response: Dict[str, Any]) -> Response: return ( Ok(response["result"], response["id"]) if "result" in response else Error( response["error"]["code"], response["error"]["message"], response["error"].get("data"), response["id"], ) ) def parse(response: Deserialized) -> Union[Response, Iterable[Response]]: return ( map(to_result, response) if isinstance(response, list) else to_result(response) )
也可以直接使用parse_json函數(shù),從json字符串生成結(jié)果:
parse_json = compose(parse, json.loads)
這里的map,componse等都是函數(shù)式編程。在server中函數(shù)式編程使用的更多,可見作者非常喜歡函數(shù)式編程的思想
jsonrpcserver的實現(xiàn)
jsonrpcclient實現(xiàn)非常簡單,jsonrpcserver的實現(xiàn)會略微復(fù)雜點,但是還是可以很好的理解的,我們一起繼續(xù)。jsonrpcserver的主要模塊如下:
模塊 | 描述 |
---|---|
main.py/async_main.py | main文件,分別是同步和異步版本 |
dispatcher.py/async_dispatcher.py | rpc服務(wù)的分配器實現(xiàn) |
methods.py | rpc函數(shù)的裝飾器 |
request.py | 請求處理 |
response.py | 響應(yīng)處理 |
result.py | 結(jié)果處理 |
examplse | 一些示例 |
通用,我們先從示例入手,看看api的使用。下面是flask版本:
# flask_server.py from flask import Flask, Response, request from jsonrpcserver import method, Result, Success, dispatch app = Flask(__name__) @method def ping() -> Result: return Success("pong") @app.route("/", methods=["POST"]) def index(): return Response( dispatch(request.get_data().decode()), content_type="application/json" ) if __name__ == "__main__": app.run()
從示例我們可以知道,rpc服務(wù)其實就2大步驟:
- 使用method裝飾ping函數(shù),使它支持rpc調(diào)用,ping函數(shù)返回的是一個特點的Result數(shù)據(jù)結(jié)構(gòu)
- 所有rpc調(diào)用的http-url都是根目錄,服務(wù)使用dispatch調(diào)度rpc請求
先看第一步rpc裝飾器:
# methods.py Method = Callable[..., Result] Methods = Dict[str, Method] global_methods = dict() def method( f: Optional[Method] = None, name: Optional[str] = None ) -> Callable[..., Any]: """A decorator to add a function into jsonrpcserver's internal global_methods dict. The global_methods dict will be used by default unless a methods argument is passed to `dispatch`. Functions can be renamed by passing a name argument: @method(name=bar) def foo(): ... """ def decorator(func: Method) -> Method: nonlocal name global_methods[name or func.__name__] = func return func return decorator(f) if callable(f) else cast(Method, decorator)
- 將所有的rpc函數(shù)都封裝到global_methods字典中
- 函數(shù)需要返回Result類型
第2步中,main模塊提供了dispatch的api,主要就是下面的函數(shù):
# main.py def dispatch_to_response( request: str, methods: Optional[Methods] = None, *, context: Any = NOCONTEXT, deserializer: Callable[[str], Deserialized] = json.loads, validator: Callable[[Deserialized], Deserialized] = default_validator, post_process: Callable[[Response], Any] = identity, ) -> Union[Response, List[Response], None]: """Takes a JSON-RPC request string and dispatches it to method(s), giving Response namedtuple(s) or None. This is a public wrapper around dispatch_to_response_pure, adding globals and default values to be nicer for end users. Args: request: The JSON-RPC request string. methods: Dictionary of methods that can be called - mapping of function names to functions. If not passed, uses the internal global_methods dict which is populated with the @method decorator. context: If given, will be passed as the first argument to methods. deserializer: Function that deserializes the request string. validator: Function that validates the JSON-RPC request. The function should raise an exception if the request is invalid. To disable validation, pass lambda _: None. post_process: Function that will be applied to Responses. Returns: A Response, list of Responses or None. Examples: >>> dispatch('{"jsonrpc": "2.0", "method": "ping", "id": 1}') '{"jsonrpc": "2.0", "result": "pong", "id": 1}' """ return dispatch_to_response_pure( deserializer=deserializer, validator=validator, post_process=post_process, context=context, methods=global_methods if methods is None else methods, request=request, )
- request 請求的函數(shù)名稱
- methods 可供調(diào)用的函數(shù)集合,默認(rèn)就是之前rpc裝飾器中存儲的global_methods
- deserializer 請求的反序列化函數(shù),validator請求驗證器
- post_process響應(yīng)處理函數(shù)
post_process主要就是根據(jù)結(jié)果類型,分別取不同的字段并序列化:
def to_serializable_one(response: ResponseType) -> Union[Deserialized, None]: return ( serialize_error(response._error) if isinstance(response, Left) else serialize_success(response._value) )
dispatch的實現(xiàn),主要是下面2個函數(shù)dispatch_request和call,前者查找rpc函數(shù),后者執(zhí)行rpc函數(shù)。dispatch_request內(nèi)容如下:
def dispatch_request( methods: Methods, context: Any, request: Request ) -> Tuple[Request, Result]: """Get the method, validates the arguments and calls the method. Returns: A tuple containing the Result of the method, along with the original Request. We need the ids from the original request to remove notifications before responding, and create a Response. """ return ( request, get_method(methods, request.method) .bind(partial(validate_args, request, context)) .bind(partial(call, request, context)), )
這里使用了oslash這個函數(shù)式編程庫,我們可以簡單的使用unix的管道思想去理解:
- 使用get_method查找rpc響應(yīng)函數(shù)
- 使用validate_args驗證rpc請求
- 使用call執(zhí)行rpc調(diào)用
- 3個步驟依次執(zhí)行,前者的返回值會作為后綴的參數(shù)
重中之重是call函數(shù),原理非常簡單:
def call(request: Request, context: Any, method: Method) -> Result: """Call the method. Handles any exceptions raised in the method, being sure to return an Error response. Returns: A Result. """ try: result = method(*extract_args(request, context), **extract_kwargs(request)) # validate_result raises AssertionError if the return value is not a valid # Result, which should respond with Internal Error because its a problem in the # method. validate_result(result) # Raising JsonRpcError inside the method is an alternative way of returning an error # response. except JsonRpcError as exc: return Left(ErrorResult(code=exc.code, message=exc.message, data=exc.data)) # Any other uncaught exception inside method - internal error. except Exception as exc: logger.exception(exc) return Left(InternalErrorResult(str(exc))) return result
- 使用args和kwargs動態(tài)執(zhí)行rpc函數(shù),并將結(jié)果進(jìn)行返回
- 捕獲異常,返回標(biāo)準(zhǔn)錯誤
這里的Left是函數(shù)式編程中的概念,我們可以從response的實現(xiàn),簡單了解一下:
# response.py class SuccessResult(NamedTuple): result: Any = None class ErrorResult(NamedTuple): code: int message: str data: Any = NODATA # The spec says this value may be omitted # Union of the two valid result types Result = Either[ErrorResult, SuccessResult] def Success(*args: Any, **kwargs: Any) -> Either[ErrorResult, SuccessResult]: return Right(SuccessResult(*args, **kwargs)) def Error(*args: Any, **kwargs: Any) -> Either[ErrorResult, SuccessResult]: return Left(ErrorResult(*args, **kwargs))
SuccessResult和ErrorResult是python的兩個標(biāo)準(zhǔn)對象;Result是oslash中定義的聯(lián)合對象,在ErrorResult, SuccessResult中二選一,有些類似rust中的Option;Right封裝了正確的結(jié)果,Left封裝了錯誤的結(jié)果。
這一部分需要一些函數(shù)式編程的基礎(chǔ),如果不太理解,推薦閱讀參考鏈接。
小結(jié)
我們一起學(xué)習(xí)了JSON-RPC規(guī)范,并且了解了Exploding Labs如何使用 現(xiàn)代python 實現(xiàn)該規(guī)范,也接觸了一些函數(shù)式編程的方式。
小技巧
業(yè)務(wù)有時候需要自己實現(xiàn)一個簡單的自增id,我們也許會用全局變量來做:
start = 0 def gen1(): start +=1 return count # 調(diào)用 id = gen1()
全局變量會形成一些污染,利用閉包的特性,我們可以優(yōu)化成這樣:
def gen2(): start = 0 def incr(): start +=1 return count return incr gen = gen2() # 調(diào)用 id = gen()
json-rpc里提供了使用yeild關(guān)鍵字實現(xiàn)的版本:
def hexadecimal(start: int = 1) -> Iterator[str]: """ Incremental hexadecimal numbers. e.g. 1, 2, 3, .. 9, a, b, etc. Args: start: The first value to start with. """ while True: yield "%x" % start start += 1
參考鏈接
以上就是python json-rpc 規(guī)范源碼閱讀的詳細(xì)內(nèi)容,更多關(guān)于python json-rpc 規(guī)范的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
python使用cartopy在地圖中添加經(jīng)緯線的示例代碼
gridlines可以根據(jù)坐標(biāo)系,自動繪制網(wǎng)格線,這對于普通繪圖來說顯然不必單獨拿出來說說,但在地圖中,經(jīng)緯線幾乎是必不可少的,本文將給大家介紹了python使用cartopy在地圖中添加經(jīng)緯線的方法,需要的朋友可以參考下2024-01-01Python使用type動態(tài)創(chuàng)建類操作示例
這篇文章主要介紹了Python使用type動態(tài)創(chuàng)建類操作,結(jié)合實例形式詳細(xì)分析了Python使用type動態(tài)創(chuàng)建類的具體原理、實現(xiàn)方法與操作注意事項,需要的朋友可以參考下2020-02-02Python實現(xiàn)ElGamal加密算法的示例代碼
ElGamal加密算法是一個基于迪菲-赫爾曼密鑰交換的非對稱加密算法。這篇文章通過示例代碼給大家介紹Python實現(xiàn)ElGamal加密算法的相關(guān)知識,感興趣的朋友一起看看吧2020-06-06為Python的Tornado框架配置使用Jinja2模板引擎的方法
Jinja2是人氣Web框架Flask中的內(nèi)置模板引擎,而且與Django的模板引擎比較類似,這里我們就來看一下為Python的Tornado框架配置使用Jinja2模板引擎的方法2016-06-06