jsonrpc/jsonrpc/__init__.py

import json
import logging
from inspect import signature

PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
INTERNAL_ERROR = -32603
GENERIC_APPLICATION_ERROR = -32000


class JsonRpcException(Exception):
    """Исключение
    """
    def __init__(self, id, code, message):
        self.id = id
        self.code = code
        self.message = message

    def as_dict(self):
        """Класс в JSON словарь
        """
        result = {
            'jsonrpc':'2.0',
            'id': self.id,
            'error': {
                'code': self.code,
                'message':self.message
            }
        }
        return result

    def __str__(self):
        return json.dumps(self.as_dict())


class JSONRPC:
    """Основной класс JSON-RPC
    """
    def __init__(self):
        self.methods = {}

    def method(self, name: str):
        """Декоратор метода
        """
        assert len(name) > 0, 'Не указано имя метода'
        # logging.info('{} {}: {}'.format(
        #     func.__module__,
        #     func.__name__,
        #     func.__doc__
        # ))
        # funcname = func.__name__
        # modulename = func.__module__
        def wrap(func):
            # print(func)
            self.methods[name] = func
            return func
        return wrap

    def description(self, name: str):
        """Описание процедуры
        """
        if name not in self.methods:
            return None
        func = self.methods[name]
        sig = signature(func)
        # for key in sig.parameters:
        #     print(sig.parameters[key].annotation)
        result = {
            'name': getattr(func, '__name__', None),
            'summary': getattr(func, '__doc__', None),
            'params': [
                {'name': k, 'type': sig.parameters[k].annotation.__name__}
                for k in sig.parameters
            ],
            'return': sig.return_annotation.__name__,
        }
        return result

    def example(self, name: str):
        """Пример
        """
        if name not in self.methods:
            return None
        func = self.methods[name]
        sig = signature(func)
        params = {}
        for key in sig.parameters:
            params[key] = ''
        result = {
            "jsonrpc": "2.0",
            "method": name,
            "params": params,
            "id": 1
        }
        return result

    def validate(self, query):
        """Валидация запроса
        """
        keys = query.keys()
        logging.info(f'keys: {keys}')
        if 'id' not in query:
            return JsonRpcException(
                query['id'],
                INVALID_REQUEST,
                f'Некорректный запрос: {query}'
            ).as_dict()

        if 'method' not in query:
            return JsonRpcException(
                query['id'],
                INVALID_REQUEST,
                f'Некорректный запрос: {query}'
            ).as_dict()

    def process(self, query):
        """Выполнение метода
        """
        self.validate(query)
        method = query['method']
        if method not in self.methods:
            return JsonRpcException(
                query['id'],
                METHOD_NOT_FOUND,
                f'Метод не найден: {method}'
            ).as_dict()

        func = self.methods[method]

        if 'params' not in query:
            params = {}
        else:
            params = query['params']
        # for key in params:
        #     print(f'{key}: {type(params[key]).__name__} = {params[key]}')
        logging.debug('params: {}'.format(params))

        try:
            if isinstance(params, (tuple, set, list)):
                response = func(*params)
            elif isinstance(params, dict):
                response = func(**params)
            else:
                return JsonRpcException(
                    query['id'],
                    INVALID_PARAMS,
                    'Invalid params: {0}'.format(params)
                )
        except BaseException as e:
            result = {
                "jsonrpc": "2.0",
                "id": query['id'],
                "error": {
                    "message": str(e)
                }
            }
        else:
            result = {
                "jsonrpc": "2.0",
                "id": query['id'],
                "result": response
            }
        return result


    def __call__(self, queries):
        """Вызов метода
        """
        if isinstance(queries, dict):
            result = self.process(queries)
        elif isinstance(queries, list):
            result = []
            for query in queries:
                result.append(self.process(query))
        return result

    def __getitem__(self, key):
        return self.methods[key]

    def __iter__(self):
        return iter(self.methods)

    def __len__(self):
        return len(self.methods)

    def __setitem__(self, key, value):
        self.methods[key] = value

    def __delitem__(self, key):
        del self.methods[key]

    def __repr__(self):
        return repr(self.methods)