python json-rpc 规范源码阅读_python_

# 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 } } } 

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) 

# 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 -X POST http://localhost:5001 -d '{"jsonrpc": "2.0", "method": "ping", "params": {}, "id": 1}' 

# 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] 

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 = compose(parse, json.loads) 

# 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() 

# 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) 

# 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, ) 

def to_serializable_one(response: ResponseType) -> Union[Deserialized, None]: return ( serialize_error(response._error) if isinstance(response, Left) else serialize_success(response._value) ) 

def dispatch_request( methods
                
提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                

                

                                            
上一篇：Python随机值生成的常用方法总结_python_
                                                                
下一篇：Python ARP扫描与欺骗实现全程详解_python_