16540272538

Committed 26 Jul 2025 01:22PM UTC coverage: 99.007% (+4.3%) from 94.679%

Build # 16540272538

Build Type

push

github

Committed by

alorence

Commit Message

Fix python version used when running check

Run Details

161 of 161 branches covered (100.0%)

1296 of 1309 relevant lines covered (99.01%)

25.52 hits per line

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

98.78

modernrpc/server.py

from __future__ import annotations

import functools
import logging
from typing import TYPE_CHECKING, Callable, Union

from django.utils.module_loading import import_string
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from modernrpc import Protocol, RpcRequestContext
from modernrpc.compat import async_csrf_exempt, async_require_post
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
from modernrpc.views import handle_rpc_request, handle_rpc_request_async

if TYPE_CHECKING:
    from typing import Any

    from django.http import HttpRequest

    from modernrpc.handler import RpcHandler


logger = logging.getLogger(__name__)

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


class RegistryMixin:
    def __init__(self, auth: Any = 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: Any = 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."
                )

            wrapper = ProcedureWrapper(
                func,
                name,
                protocol=protocol,
                auth=self.auth if auth == NOT_SET else auth,
                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}")
            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: Any = NOT_SET,
        error_handler: RpcErrorHandler | None = None,
    ) -> None:
        super().__init__(auth)
        self.request_handlers_classes = 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

    def register_namespace(self, namespace: RpcNamespace, name: str | None = None) -> None:
        prefix = f"{name}." if name else ""
        logger.debug(f"About to register {len(namespace.procedures)} procedures into namespace '{prefix}'")

        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:
        klass = first_true(self.request_handlers_classes, pred=lambda cls: cls.can_handle(request))
        try:
            return klass()
        except TypeError:
            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))

    @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(handle_rpc_request, server=self)
        return csrf_exempt(require_POST(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(handle_rpc_request_async, server=self)
        return async_csrf_exempt(async_require_post(view_func))

1	from __future__ import annotations	26✔
2
3	import functools	26✔
4	import logging	26✔
5	from typing import TYPE_CHECKING, Callable, Union	26✔
6
7	from django.utils.module_loading import import_string	26✔
8	from django.views.decorators.csrf import csrf_exempt	26✔
9	from django.views.decorators.http import require_POST	26✔
10
11	from modernrpc import Protocol, RpcRequestContext	26✔
12	from modernrpc.compat import async_csrf_exempt, async_require_post	26✔
13	from modernrpc.config import settings	26✔
14	from modernrpc.constants import NOT_SET, SYSTEM_NAMESPACE_DOTTED_PATH	26✔
15	from modernrpc.core import ProcedureWrapper	26✔
16	from modernrpc.exceptions import RPCException, RPCInternalError, RPCMethodNotFound	26✔
17	from modernrpc.helpers import check_flags_compatibility, first_true	26✔
18	from modernrpc.views import handle_rpc_request, handle_rpc_request_async	26✔
19
20	if TYPE_CHECKING:
21	from typing import Any
22
23	from django.http import HttpRequest
24
25	from modernrpc.handler import RpcHandler
26
27
28	logger = logging.getLogger(__name__)	26✔
29
30	RpcErrorHandler = Callable[[BaseException, RpcRequestContext], Union[RPCException, None]]	26✔
31
32
33	class RegistryMixin:	26✔
34	def __init__(self, auth: Any = NOT_SET) -> None:	26✔
35	self._registry: dict[str, ProcedureWrapper] = {}	26✔
36	self.auth = auth	26✔
37
38	def register_procedure(	26✔
39	self,
40	procedure: Callable \| None = None,
41	name: str \| None = None,
42	protocol: Protocol = Protocol.ALL,
43	auth: Any = NOT_SET,
44	context_target: str \| None = None,
45	) -> Callable:
46	"""
47	Registers a procedure for handling RPC (Remote Procedure Call) requests. This function can be used as a
48	decorator to register the given callable function as an RPC procedure.
49
50	:param procedure: A callable function to be registered as an RPC procedure
51	:param name: A custom name for the procedure. If not provided, the callable function's name is used by default.
52	:param protocol: Specifies the protocol (e.g., JSON-RPC, XML-RPC) under which the procedure can be executed.
53	Defaults: Protocol.ALL
54	:param auth: Defines user authentication settings or access rules for the procedure. Defaults to NOT_SET, in
55	which case the server's authentication settings will be used.
56	:param context_target: Specify the procedure argument name for accessing the RPC request context.
57
58	:raises ValueError: If a procedure can't be registered
59	"""
60
61	def decorated(func: Callable) -> Callable:	26✔
62	if name and name.startswith("rpc."):	26✔
63	raise ValueError(	26✔
64	'According to JSON-RPC specs, method names starting with "rpc." are reserved for system extensions '
65	"and must not be used. See https://www.jsonrpc.org/specification#extensions for more information."
66	)
67
68	wrapper = ProcedureWrapper(	26✔
69	func,
70	name,
71	protocol=protocol,
72	auth=self.auth if auth == NOT_SET else auth,
73	context_target=context_target,
74	)
75
76	if wrapper.name in self._registry and wrapper != self._registry[wrapper.name]:	26✔
77	raise ValueError(f"Procedure {wrapper.name} is already registered")	×
78
79	self._registry[wrapper.name] = wrapper	26✔
80	logger.debug(f"Registered procedure {wrapper.name}")	26✔
81	return func	26✔
82
83	# If @server.register_procedure() is used with parenthesis (with or without argument)
84	if procedure is None:	26✔
85	return decorated	26✔
86
87	# If @server.register_procedure is used without a parenthesis
88	return decorated(procedure)	26✔
89
90	@property	26✔
91	def procedures(self) -> dict[str, ProcedureWrapper]:	26✔
92	return self._registry	26✔
93
94
95	class RpcNamespace(RegistryMixin): ...	26✔
96
97
98	class RpcServer(RegistryMixin):	26✔
99	def __init__(	26✔
100	self,
101	register_system_procedures: bool = True,
102	supported_protocol: Protocol = Protocol.ALL,
103	auth: Any = NOT_SET,
104	error_handler: RpcErrorHandler \| None = None,
105	) -> None:
106	super().__init__(auth)	26✔
107	self.request_handlers_classes = list(	26✔
108	filter(
109	lambda cls: check_flags_compatibility(cls.protocol, supported_protocol),
110	(import_string(klass) for klass in settings.MODERNRPC_HANDLERS),
111	)
112	)
113
114	if register_system_procedures:	26✔
115	system_namespace = import_string(SYSTEM_NAMESPACE_DOTTED_PATH)	26✔
116	self.register_namespace(system_namespace, "system")	26✔
117
118	self.error_handler = error_handler	26✔
119
120	def register_namespace(self, namespace: RpcNamespace, name: str \| None = None) -> None:	26✔
121	prefix = f"{name}." if name else ""	26✔
122	logger.debug(f"About to register {len(namespace.procedures)} procedures into namespace '{prefix}'")	26✔
123
124	for procedure_name, wrapper in namespace.procedures.items():	26✔
125	self.register_procedure(	26✔
126	wrapper.func_or_coro,
127	name=f"{prefix}{procedure_name}",
128	protocol=wrapper.protocol,
129	auth=wrapper.auth,
130	context_target=wrapper.context_target,
131	)
132
133	def get_procedure_wrapper(self, name: str, protocol: Protocol) -> ProcedureWrapper:	26✔
134	"""Return the procedure wrapper with given name compatible with given protocol, or raise RPCMethodNotFound"""
135	try:	26✔
136	wrapper = self.procedures[name]	26✔
137	except KeyError:	26✔
138	raise RPCMethodNotFound(name) from None	26✔
139
140	if check_flags_compatibility(wrapper.protocol, protocol):	26✔
141	return wrapper	26✔
142
143	raise RPCMethodNotFound(name) from None	26✔
144
145	def get_request_handler(self, request: HttpRequest) -> RpcHandler \| None:	26✔
146	klass = first_true(self.request_handlers_classes, pred=lambda cls: cls.can_handle(request))	26✔
147	try:	26✔
148	return klass()	26✔
149	except TypeError:	26✔
150	return None	26✔
151
152	def on_error(self, exception: BaseException, context: RpcRequestContext) -> RPCException:	26✔
153	"""
154	If an error handler is defined, call it to run arbitrary code and return its not-None result.
155	Else, check the given exception type and return it if it is a subclass of RPCException.
156	Convert any other exception into RPCInternalError.
157	"""
158	if self.error_handler:	26✔
159	try:	26✔
160	self.error_handler(exception, context)	26✔
161	except Exception as exc:	26✔
162	exception = exc	26✔
163
164	if isinstance(exception, RPCException):	26✔
165	return exception	26✔
166	return RPCInternalError(message=str(exception))	26✔
167
168	@property	26✔
169	def view(self) -> Callable:	26✔
170	"""
171	Returns a synchronous view function that can be used in Django URL patterns.
172	The view is decorated with csrf_exempt and require_POST.
173
174	:return: A callable view function
175	"""
176	view_func = functools.partial(handle_rpc_request, server=self)	26✔
177	return csrf_exempt(require_POST(view_func))	26✔
178
179	@property	26✔
180	def async_view(self) -> Callable:	26✔
181	"""
182	Returns an asynchronous view function that can be used in Django URL patterns.
183	The view is decorated with csrf_exempt and require_POST.
184
185	:return: An awaitable async view function
186	"""
187	view_func = functools.partial(handle_rpc_request_async, server=self)	26✔
188	return async_csrf_exempt(async_require_post(view_func))	26✔

alorence / django-modern-rpc / 16540272538

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