18452590329

pending completion

Build # 18452590329

Build Type

Pull #121

github

Committed by

web-flow

Commit Message

Merge 7fa0c1b6f into a7cf0b823

Pull Request Pull Request #121: Bump ruff from 0.13.3 to 0.14.0

Run Details

166 of 166 branches covered (100.0%)

1383 of 1397 relevant lines covered (99.0%)

11.6 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	13✔
4
5	from typing import TYPE_CHECKING, Iterable, overload	13✔
6
7	from modernrpc.constants import NOT_SET	13✔
8	from modernrpc.exceptions import RPCInvalidRequest	13✔
9	from modernrpc.jsonrpc.handler import JsonRpcRequest	13✔
10	from modernrpc.types import DictStrAny, RpcErrorResult	13✔
11
12	try:	13✔
13	# types.NoneType is available only with Python 3.10+
14	from types import NoneType	13✔
15	except ImportError:	1✔
16	NoneType = type(None) # type: ignore[misc]	1✔
17
18	if TYPE_CHECKING:
19	from modernrpc.jsonrpc.handler import JsonRpcResult
20
21
22	class Unmarshaller:	13✔
23	def __init__(self, validate_version: bool = True):	13✔
24	self.validate_version = validate_version	13✔
25
26	def validate_dict_request(self, request_data: DictStrAny) -> None:	13✔
27	if "method" not in request_data:	13✔
28	raise RPCInvalidRequest("Unable to find method name", data=request_data)	13✔
29
30	if "jsonrpc" not in request_data:	13✔
31	raise RPCInvalidRequest("jsonrpc required")	13✔
32
33	if self.validate_version and request_data["jsonrpc"] != "2.0":	13✔
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")	13✔
41	if type(request_id) not in (NoneType, int, float, str):	13✔
42	raise RPCInvalidRequest(	13✔
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	1✔
48	def dict_to_request(self, structured_data: DictStrAny) -> JsonRpcRequest: ...	1✔
49
50	@overload	1✔
51	def dict_to_request(self, structured_data: list[DictStrAny]) -> list[JsonRpcRequest]: ...	1✔
52
53	def dict_to_request(self, structured_data: DictStrAny \| list[DictStrAny]) -> JsonRpcRequest \| list[JsonRpcRequest]:	13✔
54	if isinstance(structured_data, list):	13✔
55	return [self.dict_to_request(data) for data in structured_data]	13✔
56
57	self.validate_dict_request(structured_data)	13✔
58
59	method_name = structured_data["method"]	13✔
60	params = structured_data.get("params")	13✔
61	args = params if isinstance(params, (list, tuple)) else []	13✔
62	kwargs = params if isinstance(params, dict) else {}	13✔
63
64	# Build request object. If "id" not in data, it's a notification request: request_id == NOT_SET
65	return JsonRpcRequest(	13✔
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:	13✔
74	@overload	1✔
75	def result_to_dict(self, result: JsonRpcResult) -> DictStrAny \| None: ...	1✔
76
77	@overload	1✔
78	def result_to_dict(self, result: Iterable[JsonRpcResult]) -> list[DictStrAny \| None]: ...	1✔
79
80	def result_to_dict(	13✔
81	self, result: JsonRpcResult \| Iterable[JsonRpcResult]
82	) -> DictStrAny \| None \| list[DictStrAny \| None]:
83	if isinstance(result, Iterable):	13✔
84	return [self.result_to_dict(r) for r in result]	13✔
85
86	if result.request.is_notification:	13✔
87	return None	13✔
88
89	base_result = {	13✔
90	"id": result.request.request_id,
91	"jsonrpc": result.request.jsonrpc,
92	}
93
94	if isinstance(result, RpcErrorResult):	13✔
95	response_payload: DictStrAny = {	13✔
96	**base_result,
97	"error": {
98	"code": result.code,
99	"message": result.message,
100	},
101	}
102	if result.data:	13✔
103	response_payload["error"]["data"] = result.data	13✔
104	return response_payload	13✔
105
106	return {	13✔
107	**base_result,
108	"result": result.data,
109	}

alorence / django-modern-rpc / 18452590329

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