16075520752

Committed 04 Jul 2025 02:03PM UTC coverage: 94.178% (-0.6%) from 94.807%

Build # 16075520752

Build Type

push

github

Committed by

alorence

Commit Message

Try to simplify SonarQube upload

Run Details

87 of 94 branches covered (92.55%)

1197 of 1271 relevant lines covered (94.18%)

0.94 hits per line

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

94.34

modernrpc/jsonrpc/backends/marshalling.py

# PEP 585: use of list[Any] instead of List[Any] is available since Python 3.9, enable it for older versions
# PEP 604: use of typeA | typeB is available since Python 3.10, enable it for older versions
from __future__ import annotations

from typing import TYPE_CHECKING, Iterable, overload

from modernrpc.exceptions import RPCInvalidRequest
from modernrpc.jsonrpc.handler import JsonRpcRequest
from modernrpc.types import RpcErrorResult

# NoneType is available in types base module only from Python 3.10
try:
    from types import NoneType
except ImportError:
    NoneType = type(None)  # type: ignore[misc]

if TYPE_CHECKING:
    from typing import Any

    from modernrpc.jsonrpc.handler import JsonRpcResult


class Unmarshaller:
    @staticmethod
    def validate_dict_request(request_data: dict[str, Any]) -> None:
        if "method" not in request_data:
            raise RPCInvalidRequest("Unable to find method name", data=request_data)

        if "jsonrpc" not in request_data:
            raise RPCInvalidRequest("jsonrpc required")

        if request_data["jsonrpc"] != "2.0":
            raise RPCInvalidRequest(
                f'Parameter "jsonrpc" has an unsupported value "{request_data["jsonrpc"]}". It must be set to "2.0"'
            )

        # Notification request won't have "id" field
        # None is also an allowed value. Both case are valid
        request_id = request_data.get("id")
        if type(request_id) not in (NoneType, int, float, str):
            raise RPCInvalidRequest(
                'Parameter "id" has an unsupported value. According to JSON-RPC 2.0 standard, it must '
                f"be a String, a Number or a Null value. Found: {type(request_id)}"
            )

    @overload
    def dict_to_request(self, structured_data: dict[str, Any]) -> JsonRpcRequest: ...

    @overload
    def dict_to_request(self, structured_data: list[dict[str, Any]]) -> list[JsonRpcRequest]: ...

    def dict_to_request(
        self, structured_data: dict[str, Any] | list[dict[str, Any]]
    ) -> JsonRpcRequest | list[JsonRpcRequest]:
        if isinstance(structured_data, list):
            return [self.dict_to_request(data) for data in structured_data]

        self.validate_dict_request(structured_data)

        method_name = structured_data["method"]
        params = structured_data.get("params")
        args = params if isinstance(params, (list, tuple)) else []
        kwargs = params if isinstance(params, dict) else {}

        # Standard request (with "id" field)
        if "id" in structured_data:
            return JsonRpcRequest(
                request_id=structured_data["id"],
                method_name=method_name,
                args=args,
                kwargs=kwargs,
            )

        # Notification request
        return JsonRpcRequest(
            method_name=method_name,
            args=args,
            kwargs=kwargs,
        )


class Marshaller:
    @overload
    def result_to_dict(self, result: JsonRpcResult) -> dict[str, Any] | None: ...

    @overload
    def result_to_dict(self, result: Iterable[JsonRpcResult]) -> list[dict[str, Any] | None]: ...

    def result_to_dict(
        self, result: JsonRpcResult | Iterable[JsonRpcResult]
    ) -> dict[str, Any] | None | list[dict[str, Any] | None]:
        if isinstance(result, Iterable):
            return [self.result_to_dict(r) for r in result]

        if result.request.is_notification:
            return None

        base_result = {
            "id": result.request.request_id,
            "jsonrpc": result.request.jsonrpc,
        }

        if isinstance(result, RpcErrorResult):
            response_payload: dict[str, Any] = {
                **base_result,
                "error": {
                    "code": result.code,
                    "message": result.message,
                },
            }
            if result.data:
                response_payload["error"]["data"] = result.data
            return response_payload

        return {
            **base_result,
            "result": result.data,
        }

1	# PEP 585: use of list[Any] instead of List[Any] is available since Python 3.9, enable it for older versions
2	# PEP 604: use of typeA \| typeB is available since Python 3.10, enable it for older versions
3	from __future__ import annotations	1✔
4
5	from typing import TYPE_CHECKING, Iterable, overload	1✔
6
7	from modernrpc.exceptions import RPCInvalidRequest	1✔
8	from modernrpc.jsonrpc.handler import JsonRpcRequest	1✔
9	from modernrpc.types import RpcErrorResult	1✔
10
11	# NoneType is available in types base module only from Python 3.10
12	try:	1✔
13	from types import NoneType	1✔
14	except ImportError:	×
15	NoneType = type(None) # type: ignore[misc]	×
16
17	if TYPE_CHECKING:
18	from typing import Any
19
20	from modernrpc.jsonrpc.handler import JsonRpcResult
21
22
23	class Unmarshaller:	1✔
24	@staticmethod	1✔
25	def validate_dict_request(request_data: dict[str, Any]) -> None:	1✔
26	if "method" not in request_data:	1✔
27	raise RPCInvalidRequest("Unable to find method name", data=request_data)	1✔
28
29	if "jsonrpc" not in request_data:	1✔
30	raise RPCInvalidRequest("jsonrpc required")	1✔
31
32	if request_data["jsonrpc"] != "2.0":	1✔
33	raise RPCInvalidRequest(	×
34	f'Parameter "jsonrpc" has an unsupported value "{request_data["jsonrpc"]}". It must be set to "2.0"'
35	)
36
37	# Notification request won't have "id" field
38	# None is also an allowed value. Both case are valid
39	request_id = request_data.get("id")	1✔
40	if type(request_id) not in (NoneType, int, float, str):	1✔
41	raise RPCInvalidRequest(	1✔
42	'Parameter "id" has an unsupported value. According to JSON-RPC 2.0 standard, it must '
43	f"be a String, a Number or a Null value. Found: {type(request_id)}"
44	)
45
46	@overload	1✔
47	def dict_to_request(self, structured_data: dict[str, Any]) -> JsonRpcRequest: ...	1✔
48
49	@overload	1✔
50	def dict_to_request(self, structured_data: list[dict[str, Any]]) -> list[JsonRpcRequest]: ...	1✔
51
52	def dict_to_request(	1✔
53	self, structured_data: dict[str, Any] \| list[dict[str, Any]]
54	) -> JsonRpcRequest \| list[JsonRpcRequest]:
55	if isinstance(structured_data, list):	1✔
56	return [self.dict_to_request(data) for data in structured_data]	1✔
57
58	self.validate_dict_request(structured_data)	1✔
59
60	method_name = structured_data["method"]	1✔
61	params = structured_data.get("params")	1✔
62	args = params if isinstance(params, (list, tuple)) else []	1✔
63	kwargs = params if isinstance(params, dict) else {}	1✔
64
65	# Standard request (with "id" field)
66	if "id" in structured_data:	1✔
67	return JsonRpcRequest(	1✔
68	request_id=structured_data["id"],
69	method_name=method_name,
70	args=args,
71	kwargs=kwargs,
72	)
73
74	# Notification request
75	return JsonRpcRequest(	1✔
76	method_name=method_name,
77	args=args,
78	kwargs=kwargs,
79	)
80
81
82	class Marshaller:	1✔
83	@overload	1✔
84	def result_to_dict(self, result: JsonRpcResult) -> dict[str, Any] \| None: ...	1✔
85
86	@overload	1✔
87	def result_to_dict(self, result: Iterable[JsonRpcResult]) -> list[dict[str, Any] \| None]: ...	1✔
88
89	def result_to_dict(	1✔
90	self, result: JsonRpcResult \| Iterable[JsonRpcResult]
91	) -> dict[str, Any] \| None \| list[dict[str, Any] \| None]:
92	if isinstance(result, Iterable):	1✔
93	return [self.result_to_dict(r) for r in result]	1✔
94
95	if result.request.is_notification:	1✔
96	return None	1✔
97
98	base_result = {	1✔
99	"id": result.request.request_id,
100	"jsonrpc": result.request.jsonrpc,
101	}
102
103	if isinstance(result, RpcErrorResult):	1✔
104	response_payload: dict[str, Any] = {	1✔
105	**base_result,
106	"error": {
107	"code": result.code,
108	"message": result.message,
109	},
110	}
111	if result.data:	1✔
112	response_payload["error"]["data"] = result.data	1✔
113	return response_payload	1✔
114
115	return {	1✔
116	**base_result,
117	"result": result.data,
118	}

alorence / django-modern-rpc / 16075520752

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