19383791482

Committed 10 Nov 2025 12:28PM UTC coverage: 99.018% (+0.1%) from 98.878%

Build # 19383791482

Build Type

push

github

Committed by

alorence

Commit Message

typo

Run Details

178 of 179 branches covered (99.44%)

1412 of 1426 relevant lines covered (99.02%)

29.03 hits per line

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

98.11

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.constants import NOT_SET
from modernrpc.exceptions import RPCInvalidRequest
from modernrpc.jsonrpc.handler import JsonRpcRequest
from modernrpc.types import DictStrAny, RpcErrorResult

try:
    # types.NoneType is available only with Python 3.10+
    from types import NoneType
except ImportError:
    NoneType = type(None)  # type: ignore[misc]

if TYPE_CHECKING:
    from modernrpc.jsonrpc.handler import JsonRpcResult


class Unmarshaller:
    def __init__(self, validate_version: bool = True):
        self.validate_version = validate_version

    def validate_dict_request(self, request_data: DictStrAny) -> 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 self.validate_version and 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 cases 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: DictStrAny) -> JsonRpcRequest: ...

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

    def dict_to_request(self, structured_data: DictStrAny | list[DictStrAny]) -> 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 {}

        # Build request object. If "id" not in data, it's a notification request: request_id == NOT_SET
        return JsonRpcRequest(
            request_id=structured_data.get("id", NOT_SET),
            method_name=method_name,
            args=args,
            kwargs=kwargs,
        )


class Marshaller:
    @overload
    def result_to_dict(self, result: JsonRpcResult) -> DictStrAny | None: ...

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

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

alorence / django-modern-rpc / 19383791482

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