add src/jsonrpc/rpc.py

src/jsonrpc/rpc.py

@@ -0,0 +1,166 @@
+__author__ = 'RemiZOffAlex'
+__email__ = 'remizoffalex@mail.ru'
+
+import logging
+import traceback
+
+from inspect import signature
+from .exceptions import *
+
+
+log = logging.getLogger(__name__)
+
+
+class RPC:
+    """Основной класс RPC
+    """
+    def __init__(self, debug: bool = False):
+        self.__handlers = {}
+        self.debug = debug
+
+    def register(self, name: str):
+        """Регистратор обработчика
+        """
+        assert len(name) > 0, 'Не указано имя метода'
+
+        def wrap(handler):
+            self.__handlers[name] = handler
+            return handler
+        return wrap
+
+    def description(self, name: str):
+        """Описание процедуры
+        """
+        if name not in self.__handlers:
+            return None
+        handler = self.__handlers[name]
+        sig = signature(method.function)
+        # for key in sig.parameters:
+        #     print(sig.parameters[key].annotation)
+        result = {
+            'name': name,
+            'function': getattr(method, '__name__', None),
+            'summary': getattr(method, '__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.__handlers:
+            return None
+        handler = self.__handlers[name]
+        sig = signature(handler)
+
+        result = {
+            "jsonrpc": "3.0",
+            "method": name,
+            "id": '00000000-0000-0000-0000-000000000000'
+        }
+        if len(sig.parameters) == 0:
+            params = {}
+            for key in sig.parameters:
+                params[key] = ''
+            result["params"] = params
+
+        return result
+
+    def validate(self, query):
+        """Валидация запроса
+        """
+        required = [
+            'id', 'method', 'jsonrpc'
+        ]
+        for field in required:
+            if field not in query:
+                result = InvalidRequestError(
+                    message=f'Некорректный запрос: {query}'
+                )
+                return result
+        if query['jsonrpc'] not in ['2.0', '3.0']:
+            result = InvalidRequestError(
+                message=f'Некорректный запрос: {query}'
+            )
+            return result
+
+    def process(self, query):
+        """Выполнение метода
+        """
+        result = self.validate(query)
+        if result:
+            return result
+        name = query['method']
+        if name not in self.__handlers:
+            result = MethodNotFoundError(
+                message=f'Метод не найден: {name}',
+                id=query['id']
+            )
+            return result
+
+        handler = self.__handlers[name]
+
+        try:
+            response = handler(query)
+        except JSONRPCError as e:
+            log.error(traceback.format_exc())
+            response = InternalError(
+                id=query['id'],
+                message=str(e)
+                # message=traceback.format_exc()
+            )
+        except Exception as e:
+            log.error(traceback.format_exc())
+            response = InternalError(
+                id=query['id'],
+                message=str(e)
+                # message=traceback.format_exc()
+            )
+        else:
+            response = {
+                "jsonrpc": query['jsonrpc'],
+                "id": query['id'],
+                "result": response
+            }
+        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:
+                response = self.process(query)
+                if response:
+                    result.append(response)
+        return result
+
+    def __getattr__(self, key):
+        if key == 'handlers':
+            return self.__handlers
+
+    def __getitem__(self, key):
+        handler = self.__handlers[key]
+        return handler
+
+    def __iter__(self):
+        return iter(self.__handlers)
+
+    def __len__(self):
+        return len(self.__handlers)
+
+    def __setitem__(self, key, handler):
+        self.__handlers[key] = handler
+
+    def __delitem__(self, key):
+        del self.__handlers[key]
+
+    def __repr__(self):
+        return repr(self.__handlers)