Source code for modernrpc.views

import logging
from typing import Generator, Sequence, TYPE_CHECKING, Type

from django.core.exceptions import ImproperlyConfigured
from django.http.response import HttpResponse
from django.utils.decorators import method_decorator
from django.utils.functional import cached_property
from django.utils.module_loading import import_string
from django.views.decorators.csrf import csrf_exempt
from django.views.generic.base import TemplateView, View

from modernrpc.conf import settings
from modernrpc.core import Protocol, RPCRequestContext, registry
from modernrpc.helpers import ensure_sequence


if TYPE_CHECKING:
    from modernrpc.handlers.base import RPCHandler

logger = logging.getLogger(__name__)


# Disables CSRF validation for POST requests
[docs]@method_decorator(csrf_exempt, name="dispatch") class RPCEntryPoint(TemplateView): """ This is the main entry point class. It inherits standard Django View class. """ template_name = "modernrpc/default/index.html" entry_point: str = settings.MODERNRPC_DEFAULT_ENTRYPOINT_NAME protocol: Protocol = Protocol.ALL enable_doc: bool = False enable_rpc: bool = True default_encoding = "utf-8" def __init__(self, **kwargs): super().__init__(**kwargs) if not self.handler_classes: raise ImproperlyConfigured("At least 1 handler must be instantiated.") # Copy static list http_method_names locally (in instance), so we can dynamically customize it self.http_method_names = list(View.http_method_names) # Customize allowed HTTP methods name to forbid access to GET when this EntryPoint # must not display documentation... if not self.enable_doc: self.http_method_names.remove("get") # ... and also forbid access to POST when this EntryPoint must not support RPC request (docs only view) if not self.enable_rpc: self.http_method_names.remove("post") logger.debug('RPC entry point "%s" initialized', self.entry_point) @cached_property def handler_classes(self) -> Sequence[Type["RPCHandler"]]: """Return the list of handlers to use when receiving RPC requests.""" handler_classes = [import_string(handler_cls) for handler_cls in settings.MODERNRPC_HANDLERS] if self.protocol == Protocol.ALL: return handler_classes return [cls for cls in handler_classes if cls.protocol in ensure_sequence(self.protocol)] @cached_property def handlers(self) -> Generator["RPCHandler", None, None]: for cls in self.handler_classes: yield cls(entry_point=self.entry_point)
[docs] def post(self, request, *args, **kwargs): """ Handle an XML-RPC or JSON-RPC request. :param request: Incoming request :param args: Unused :param kwargs: Unused :return: A HttpResponse containing XML-RPC or JSON-RPC response with the result of procedure call """ logger.debug( "RPC request received... Content-Type=%s", request.content_type or "<unspecified>", ) if not request.content_type: return HttpResponse( "Unable to handle your request, the Content-Type header is mandatory to allow server " "to determine which handler can interpret your request..", content_type="text/plain", ) # Retrieve the first RPC handler able to parse our request # This weird next(filter(predicate, iterable), default) structure is basically the more_itertools.first_true() # utility. This is written here to avoid dependency to more-itertools package handler = next(filter(lambda candidate: candidate.can_handle(request), self.handlers), None) if not handler: return HttpResponse( "Unable to handle your request. Please ensure you called " "the right entry point. If not, this could be a server error.", content_type="text/plain", ) context = RPCRequestContext(request, handler, handler.protocol, handler.entry_point) request_body = request.body.decode(request.encoding or self.default_encoding) result_data = handler.process_request(request_body, context) return HttpResponse(result_data, content_type=handler.response_content_type())
[docs] def get_context_data(self, **kwargs): """Update context data with list of RPC methods of the current entry point. Will be used to display methods documentation page""" kwargs.setdefault("methods", registry.get_all_methods(self.entry_point, sort_methods=True)) return super().get_context_data(**kwargs)