18986664249

Committed 31 Oct 2025 10:25PM UTC coverage: 98.356% (-0.7%) from 99.016%

Build # 18986664249

Build Type

push

github

Committed by

alorence

Commit Message

Drop views module

Views are now completely defined in server class

Run Details

84 of 84 branches covered (100.0%)

1376 of 1399 relevant lines covered (98.36%)

21.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.45

modernrpc/server.py

from __future__ import annotations

import functools
import logging
from http import HTTPStatus
from typing import TYPE_CHECKING, Any, Callable

from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseNotAllowed
from django.shortcuts import redirect
from django.utils.log import log_response
from django.utils.module_loading import import_string
from django.views.decorators.csrf import csrf_exempt

from modernrpc import Protocol, RpcRequestContext
from modernrpc.compat import async_csrf_exempt
from modernrpc.config import settings
from modernrpc.constants import NOT_SET, SYSTEM_NAMESPACE_DOTTED_PATH
from modernrpc.core import ProcedureWrapper
from modernrpc.exceptions import RPCException, RPCInternalError, RPCMethodNotFound
from modernrpc.helpers import check_flags_compatibility, first_true

if TYPE_CHECKING:
    from django.http import HttpRequest
    from django.shortcuts import SupportsGetAbsoluteUrl

    from modernrpc.handler import RpcHandler
    from modernrpc.types import AuthPredicateType


logger = logging.getLogger(__name__)

RpcErrorHandler = Callable[[BaseException, RpcRequestContext], None]


class RegistryMixin:
    def __init__(self, auth: AuthPredicateType = NOT_SET) -> None:
        self._registry: dict[str, ProcedureWrapper] = {}
        self.auth = auth

    def register_procedure(
        self,
        procedure: Callable | None = None,
        name: str | None = None,
        protocol: Protocol = Protocol.ALL,
        auth: AuthPredicateType = NOT_SET,
        context_target: str | None = None,
    ) -> Callable:
        """
        Registers a procedure for handling RPC (Remote Procedure Call) requests. This function can be used as a
        decorator to register the given callable function as an RPC procedure.

        :param procedure: A callable function to be registered as an RPC procedure
        :param name: A custom name for the procedure. If not provided, the callable function's name is used by default.
        :param protocol: Specifies the protocol (e.g., JSON-RPC, XML-RPC) under which the procedure can be executed.
                         Defaults: Protocol.ALL
        :param auth: Defines user authentication settings or access rules for the procedure. Defaults to NOT_SET, in
                     which case the server's authentication settings will be used.
        :param context_target: Specify the procedure argument name for accessing the RPC request context.

        :raises ValueError: If a procedure can't be registered
        """

        def decorated(func: Callable) -> Callable:
            if name and name.startswith("rpc."):
                raise ValueError(
                    'According to JSON-RPC specs, method names starting with "rpc." are reserved for system extensions '
                    "and must not be used. See https://www.jsonrpc.org/specification#extensions for more information."
                )

            auth_predicate = self.auth if auth == NOT_SET else auth
            wrapper = ProcedureWrapper(
                func, name, protocol=protocol, auth=auth_predicate, context_target=context_target
            )

            if wrapper.name in self._registry and wrapper != self._registry[wrapper.name]:
                raise ValueError(f"Procedure {wrapper.name} is already registered")

            self._registry[wrapper.name] = wrapper
            logger.debug(f"Registered procedure {wrapper.name} into {self.__class__.__name__} instance")
            return func

        # If @server.register_procedure() is used with parenthesis (with or without argument)
        if procedure is None:
            return decorated

        # If @server.register_procedure is used without a parenthesis
        return decorated(procedure)

    @property
    def procedures(self) -> dict[str, ProcedureWrapper]:
        return self._registry


class RpcNamespace(RegistryMixin): ...


class RpcServer(RegistryMixin):
    def __init__(
        self,
        register_system_procedures: bool = True,
        supported_protocol: Protocol = Protocol.ALL,
        auth: AuthPredicateType = NOT_SET,
        error_handler: RpcErrorHandler | None = None,
        allow_get_method: str | Callable[..., Any] | SupportsGetAbsoluteUrl | None = None,
        default_encoding: str = settings.MODERNRPC_DEFAULT_ENCODING,
    ) -> None:
        super().__init__(auth)
        self.handler_klasses = list(
            filter(
                lambda cls: check_flags_compatibility(cls.protocol, supported_protocol),
                (import_string(klass) for klass in settings.MODERNRPC_HANDLERS),
            )
        )

        if register_system_procedures:
            system_namespace = import_string(SYSTEM_NAMESPACE_DOTTED_PATH)
            self.register_namespace(system_namespace, "system")

        self.error_handler = error_handler
        self.redirect_to = allow_get_method
        self.default_encoding = default_encoding

    def register_namespace(self, namespace: RpcNamespace, name: str | None = None) -> None:
        if name:
            prefix = name + "."
            logger.debug(
                f"About to register {len(namespace.procedures)} procedure(s) "
                f"from namespace '{name}' into the top-level server"
            )
        else:
            prefix = ""
            logger.debug(
                f"About to register {len(namespace.procedures)} procedure(s) "
                f"from unnamed namespace into the top-level server"
            )

        for procedure_name, wrapper in namespace.procedures.items():
            self.register_procedure(
                wrapper.func_or_coro,
                name=f"{prefix}{procedure_name}",
                protocol=wrapper.protocol,
                auth=wrapper.auth,
                context_target=wrapper.context_target,
            )

    def get_procedure_wrapper(self, name: str, protocol: Protocol) -> ProcedureWrapper:
        """Return the procedure wrapper with given name compatible with given protocol, or raise RPCMethodNotFound"""
        try:
            wrapper = self.procedures[name]
        except KeyError:
            raise RPCMethodNotFound(name) from None

        if check_flags_compatibility(wrapper.protocol, protocol):
            return wrapper

        raise RPCMethodNotFound(name) from None

    def get_request_handler(self, request: HttpRequest) -> RpcHandler | None:
        handler_klass = first_true(self.handler_klasses, pred=lambda cls: cls.can_handle(request))
        try:
            return handler_klass()
        except TypeError:
            # first_true() returned None -> TypeError: 'NoneType' object is not callable
            return None

    def on_error(self, exception: BaseException, context: RpcRequestContext) -> RPCException:
        """
        If an error handler is defined, call it to run arbitrary code and return its not-None result.
        Else, check the given exception type and return it if it is a subclass of RPCException.
        Convert any other exception into RPCInternalError.
        """
        if self.error_handler:
            try:
                self.error_handler(exception, context)
            except Exception as exc:
                exception = exc

        if isinstance(exception, RPCException):
            return exception
        return RPCInternalError(message=str(exception))

    def build_method_not_allowed_reponse(self, request: HttpRequest) -> HttpResponseNotAllowed:
        """
        Build an HttpResponseNotAllowed instance with the correct list of allowed methods.
        """
        allowed_methods = ["GET", "POST"] if self.redirect_to else ["POST"]
        response = HttpResponseNotAllowed(allowed_methods)
        log_response(f"Method Not Allowed ({request.method}): {request.path}", response=response, request=request)
        return response

    def check_request(self, request: HttpRequest) -> HttpResponse | None:
        """
        Check incoming request for common issues. When everything is fine, return None. Else, return the appropriate
        HttpResponse instance.

        :param request: Request instance as received by Django
        :return: A response instance (HttpResponsePermanentRedirect, HttpResponseNotAllowed, HttpResponse) or None
        """
        if request.method == "GET":
            if self.redirect_to:
                return redirect(to=self.redirect_to, permanent=True)

            return self.build_method_not_allowed_reponse(request)

        if request.method != "POST":
            return self.build_method_not_allowed_reponse(request)

        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.",
                status=HTTPStatus.BAD_REQUEST,
                content_type="text/plain",
            )
        return None

    @staticmethod
    def build_response(handler: RpcHandler, result_data: str | tuple[int, str]) -> HttpResponse:
        if isinstance(result_data, tuple) and len(result_data) == 2:
            status, result_data = result_data
        else:
            status = HTTPStatus.OK

        return HttpResponse(result_data, status=status, content_type=handler.response_content_type)

    def handle_rpc_request(self, request: HttpRequest) -> HttpResponse:
        """
        Synchronous view function to handle RPC requests.

        :param request: The HTTP request object
        :return: An HTTP response object
        """
        response = self.check_request(request)
        if response:
            return response

        handler = self.get_request_handler(request)
        if not handler:
            return HttpResponseBadRequest(
                f"Unable to handle your request, unsupported Content-Type {request.content_type}.",
                content_type=request.content_type,
            )

        result_data = handler.process_request(
            request.body.decode(request.encoding or self.default_encoding),
            RpcRequestContext(request, self, handler, handler.protocol),
        )

        return self.build_response(handler, result_data)

    async def async_handle_rpc_request(self, request: HttpRequest) -> HttpResponse:
        """
        Asynchronous view function to handle RPC requests.

        :param request: The HTTP request object
        :return: An HTTP response object
        """
        response = self.check_request(request)
        if response:
            return response

        handler = self.get_request_handler(request)
        if not handler:
            return HttpResponseBadRequest(
                f"Unable to handle your request, unsupported Content-Type {request.content_type}.",
                content_type=request.content_type,
            )

        result_data = await handler.aprocess_request(
            request.body.decode(request.encoding or self.default_encoding),
            RpcRequestContext(request, self, handler, handler.protocol),
        )

        return self.build_response(handler, result_data)

    @property
    def view(self) -> Callable:
        """
        Returns a synchronous view function that can be used in Django URL patterns.
        The view is decorated with csrf_exempt and require_POST.

        :return: A callable view function
        """
        view_func = functools.partial(self.handle_rpc_request)
        return csrf_exempt(view_func)

    @property
    def async_view(self) -> Callable:
        """
        Returns an asynchronous view function that can be used in Django URL patterns.
        The view is decorated with csrf_exempt and require_POST.

        :return: An awaitable async view function
        """
        view_func = functools.partial(self.async_handle_rpc_request)
        return async_csrf_exempt(view_func)

1	from __future__ import annotations	22✔
2
3	import functools	22✔
4	import logging	22✔
5	from http import HTTPStatus	22✔
6	from typing import TYPE_CHECKING, Any, Callable	22✔
7
8	from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseNotAllowed	22✔
9	from django.shortcuts import redirect	22✔
10	from django.utils.log import log_response	22✔
11	from django.utils.module_loading import import_string	22✔
12	from django.views.decorators.csrf import csrf_exempt	22✔
13
14	from modernrpc import Protocol, RpcRequestContext	22✔
15	from modernrpc.compat import async_csrf_exempt	22✔
16	from modernrpc.config import settings	22✔
17	from modernrpc.constants import NOT_SET, SYSTEM_NAMESPACE_DOTTED_PATH	22✔
18	from modernrpc.core import ProcedureWrapper	22✔
19	from modernrpc.exceptions import RPCException, RPCInternalError, RPCMethodNotFound	22✔
20	from modernrpc.helpers import check_flags_compatibility, first_true	22✔
21
22	if TYPE_CHECKING:
23	from django.http import HttpRequest
24	from django.shortcuts import SupportsGetAbsoluteUrl
25
26	from modernrpc.handler import RpcHandler
27	from modernrpc.types import AuthPredicateType
28
29
30	logger = logging.getLogger(__name__)	22✔
31
32	RpcErrorHandler = Callable[[BaseException, RpcRequestContext], None]	22✔
33
34
35	class RegistryMixin:	22✔
36	def __init__(self, auth: AuthPredicateType = NOT_SET) -> None:	22✔
37	self._registry: dict[str, ProcedureWrapper] = {}	22✔
38	self.auth = auth	22✔
39
40	def register_procedure(	22✔
41	self,
42	procedure: Callable \| None = None,
43	name: str \| None = None,
44	protocol: Protocol = Protocol.ALL,
45	auth: AuthPredicateType = NOT_SET,
46	context_target: str \| None = None,
47	) -> Callable:
48	"""
49	Registers a procedure for handling RPC (Remote Procedure Call) requests. This function can be used as a
50	decorator to register the given callable function as an RPC procedure.
51
52	:param procedure: A callable function to be registered as an RPC procedure
53	:param name: A custom name for the procedure. If not provided, the callable function's name is used by default.
54	:param protocol: Specifies the protocol (e.g., JSON-RPC, XML-RPC) under which the procedure can be executed.
55	Defaults: Protocol.ALL
56	:param auth: Defines user authentication settings or access rules for the procedure. Defaults to NOT_SET, in
57	which case the server's authentication settings will be used.
58	:param context_target: Specify the procedure argument name for accessing the RPC request context.
59
60	:raises ValueError: If a procedure can't be registered
61	"""
62
63	def decorated(func: Callable) -> Callable:	22✔
64	if name and name.startswith("rpc."):	22✔
65	raise ValueError(	22✔
66	'According to JSON-RPC specs, method names starting with "rpc." are reserved for system extensions '
67	"and must not be used. See https://www.jsonrpc.org/specification#extensions for more information."
68	)
69
70	auth_predicate = self.auth if auth == NOT_SET else auth	22✔
71	wrapper = ProcedureWrapper(	22✔
72	func, name, protocol=protocol, auth=auth_predicate, context_target=context_target
73	)
74
75	if wrapper.name in self._registry and wrapper != self._registry[wrapper.name]:	22✔
76	raise ValueError(f"Procedure {wrapper.name} is already registered")	×
77
78	self._registry[wrapper.name] = wrapper	22✔
79	logger.debug(f"Registered procedure {wrapper.name} into {self.__class__.__name__} instance")	22✔
80	return func	22✔
81
82	# If @server.register_procedure() is used with parenthesis (with or without argument)
83	if procedure is None:	22✔
84	return decorated	22✔
85
86	# If @server.register_procedure is used without a parenthesis
87	return decorated(procedure)	22✔
88
89	@property	22✔
90	def procedures(self) -> dict[str, ProcedureWrapper]:	22✔
91	return self._registry	22✔
92
93
94	class RpcNamespace(RegistryMixin): ...	22✔
95
96
97	class RpcServer(RegistryMixin):	22✔
98	def __init__(	22✔
99	self,
100	register_system_procedures: bool = True,
101	supported_protocol: Protocol = Protocol.ALL,
102	auth: AuthPredicateType = NOT_SET,
103	error_handler: RpcErrorHandler \| None = None,
104	allow_get_method: str \| Callable[..., Any] \| SupportsGetAbsoluteUrl \| None = None,
105	default_encoding: str = settings.MODERNRPC_DEFAULT_ENCODING,
106	) -> None:
107	super().__init__(auth)	22✔
108	self.handler_klasses = list(	22✔
109	filter(
110	lambda cls: check_flags_compatibility(cls.protocol, supported_protocol),
111	(import_string(klass) for klass in settings.MODERNRPC_HANDLERS),
112	)
113	)
114
115	if register_system_procedures:	22✔
116	system_namespace = import_string(SYSTEM_NAMESPACE_DOTTED_PATH)	22✔
117	self.register_namespace(system_namespace, "system")	22✔
118
119	self.error_handler = error_handler	22✔
120	self.redirect_to = allow_get_method	22✔
121	self.default_encoding = default_encoding	22✔
122
123	def register_namespace(self, namespace: RpcNamespace, name: str \| None = None) -> None:	22✔
124	if name:	22✔
125	prefix = name + "."	22✔
126	logger.debug(	22✔
127	f"About to register {len(namespace.procedures)} procedure(s) "
128	f"from namespace '{name}' into the top-level server"
129	)
130	else:
131	prefix = ""	22✔
132	logger.debug(	22✔
133	f"About to register {len(namespace.procedures)} procedure(s) "
134	f"from unnamed namespace into the top-level server"
135	)
136
137	for procedure_name, wrapper in namespace.procedures.items():	22✔
138	self.register_procedure(	22✔
139	wrapper.func_or_coro,
140	name=f"{prefix}{procedure_name}",
141	protocol=wrapper.protocol,
142	auth=wrapper.auth,
143	context_target=wrapper.context_target,
144	)
145
146	def get_procedure_wrapper(self, name: str, protocol: Protocol) -> ProcedureWrapper:	22✔
147	"""Return the procedure wrapper with given name compatible with given protocol, or raise RPCMethodNotFound"""
148	try:	22✔
149	wrapper = self.procedures[name]	22✔
150	except KeyError:	22✔
151	raise RPCMethodNotFound(name) from None	22✔
152
153	if check_flags_compatibility(wrapper.protocol, protocol):	22✔
154	return wrapper	22✔
155
156	raise RPCMethodNotFound(name) from None	22✔
157
158	def get_request_handler(self, request: HttpRequest) -> RpcHandler \| None:	22✔
159	handler_klass = first_true(self.handler_klasses, pred=lambda cls: cls.can_handle(request))	22✔
160	try:	22✔
161	return handler_klass()	22✔
162	except TypeError:	22✔
163	# first_true() returned None -> TypeError: 'NoneType' object is not callable
164	return None	22✔
165
166	def on_error(self, exception: BaseException, context: RpcRequestContext) -> RPCException:	22✔
167	"""
168	If an error handler is defined, call it to run arbitrary code and return its not-None result.
169	Else, check the given exception type and return it if it is a subclass of RPCException.
170	Convert any other exception into RPCInternalError.
171	"""
172	if self.error_handler:	22✔
173	try:	22✔
174	self.error_handler(exception, context)	22✔
175	except Exception as exc:	22✔
176	exception = exc	22✔
177
178	if isinstance(exception, RPCException):	22✔
179	return exception	22✔
180	return RPCInternalError(message=str(exception))	22✔
181
182	def build_method_not_allowed_reponse(self, request: HttpRequest) -> HttpResponseNotAllowed:	22✔
183	"""
184	Build an HttpResponseNotAllowed instance with the correct list of allowed methods.
185	"""
186	allowed_methods = ["GET", "POST"] if self.redirect_to else ["POST"]	22✔
187	response = HttpResponseNotAllowed(allowed_methods)	22✔
188	log_response(f"Method Not Allowed ({request.method}): {request.path}", response=response, request=request)	22✔
189	return response	22✔
190
191	def check_request(self, request: HttpRequest) -> HttpResponse \| None:	22✔
192	"""
193	Check incoming request for common issues. When everything is fine, return None. Else, return the appropriate
194	HttpResponse instance.
195
196	:param request: Request instance as received by Django
197	:return: A response instance (HttpResponsePermanentRedirect, HttpResponseNotAllowed, HttpResponse) or None
198	"""
199	if request.method == "GET":	22✔
200	if self.redirect_to:	22✔
201	return redirect(to=self.redirect_to, permanent=True)	×
202
203	return self.build_method_not_allowed_reponse(request)	22✔
204
205	if request.method != "POST":	22✔
206	return self.build_method_not_allowed_reponse(request)	22✔
207
208	if not request.content_type:	22✔
209	return HttpResponse(	22✔
210	"Unable to handle your request, the Content-Type header is mandatory to allow server "
211	"to determine which handler can interpret your request.",
212	status=HTTPStatus.BAD_REQUEST,
213	content_type="text/plain",
214	)
215	return None	22✔
216
217	@staticmethod	22✔
218	def build_response(handler: RpcHandler, result_data: str \| tuple[int, str]) -> HttpResponse:	22✔
219	if isinstance(result_data, tuple) and len(result_data) == 2:	22✔
220	status, result_data = result_data	22✔
221	else:
222	status = HTTPStatus.OK	22✔
223
224	return HttpResponse(result_data, status=status, content_type=handler.response_content_type)	22✔
225
226	def handle_rpc_request(self, request: HttpRequest) -> HttpResponse:	22✔
227	"""
228	Synchronous view function to handle RPC requests.
229
230	:param request: The HTTP request object
231	:return: An HTTP response object
232	"""
233	response = self.check_request(request)	22✔
234	if response:	22✔
235	return response	22✔
236
237	handler = self.get_request_handler(request)	22✔
238	if not handler:	22✔
239	return HttpResponseBadRequest(	22✔
240	f"Unable to handle your request, unsupported Content-Type {request.content_type}.",
241	content_type=request.content_type,
242	)
243
244	result_data = handler.process_request(	22✔
245	request.body.decode(request.encoding or self.default_encoding),
246	RpcRequestContext(request, self, handler, handler.protocol),
247	)
248
249	return self.build_response(handler, result_data)	22✔
250
251	async def async_handle_rpc_request(self, request: HttpRequest) -> HttpResponse:	22✔
252	"""
253	Asynchronous view function to handle RPC requests.
254
255	:param request: The HTTP request object
256	:return: An HTTP response object
257	"""
258	response = self.check_request(request)	22✔
259	if response:	22✔
260	return response	22✔
261
262	handler = self.get_request_handler(request)	22✔
263	if not handler:	22✔
264	return HttpResponseBadRequest(	22✔
265	f"Unable to handle your request, unsupported Content-Type {request.content_type}.",
266	content_type=request.content_type,
267	)
268
269	result_data = await handler.aprocess_request(	22✔
270	request.body.decode(request.encoding or self.default_encoding),
271	RpcRequestContext(request, self, handler, handler.protocol),
272	)
273
274	return self.build_response(handler, result_data)	22✔
275
276	@property	22✔
277	def view(self) -> Callable:	22✔
278	"""
279	Returns a synchronous view function that can be used in Django URL patterns.
280	The view is decorated with csrf_exempt and require_POST.
281
282	:return: A callable view function
283	"""
284	view_func = functools.partial(self.handle_rpc_request)	22✔
285	return csrf_exempt(view_func)	22✔
286
287	@property	22✔
288	def async_view(self) -> Callable:	22✔
289	"""
290	Returns an asynchronous view function that can be used in Django URL patterns.
291	The view is decorated with csrf_exempt and require_POST.
292
293	:return: An awaitable async view function
294	"""
295	view_func = functools.partial(self.async_handle_rpc_request)	22✔
296	return async_csrf_exempt(view_func)	22✔

alorence / django-modern-rpc / 18986664249

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous