distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""JSON-RPC 2.0 server primitives for I2PControl."""
2from __future__ import annotations
3
4import json
5from dataclasses import dataclass, field
6from typing import Any
7
8# Standard JSON-RPC 2.0 error codes
9PARSE_ERROR = -32700
10INVALID_REQUEST = -32600
11METHOD_NOT_FOUND = -32601
12INTERNAL_ERROR = -32603
13
14
15@dataclass
16class JSONRPCRequest:
17 """Parsed JSON-RPC 2.0 request."""
18
19 method: str
20 params: dict = field(default_factory=dict)
21 id: int | str | None = None
22 jsonrpc: str = "2.0"
23
24
25@dataclass
26class JSONRPCResponse:
27 """JSON-RPC 2.0 response."""
28
29 result: dict | None = None
30 error: dict | None = None
31 id: int | str | None = None
32 jsonrpc: str = "2.0"
33
34 def to_dict(self) -> dict[str, Any]:
35 d: dict[str, Any] = {"jsonrpc": self.jsonrpc, "id": self.id}
36 if self.error is not None:
37 d["error"] = self.error
38 else:
39 d["result"] = self.result
40 return d
41
42
43def parse_jsonrpc_request(data: str) -> JSONRPCRequest:
44 """Parse a JSON-RPC 2.0 request string.
45
46 Raises ``ValueError`` on invalid JSON, missing ``jsonrpc`` field,
47 or missing ``method`` field.
48 """
49 try:
50 obj = json.loads(data)
51 except (json.JSONDecodeError, TypeError) as exc:
52 raise ValueError("Parse error: invalid JSON") from exc
53
54 if not isinstance(obj, dict):
55 raise ValueError("Parse error: request must be a JSON object")
56
57 if "jsonrpc" not in obj or obj["jsonrpc"] != "2.0":
58 raise ValueError("Invalid request: missing or wrong jsonrpc field")
59
60 if "method" not in obj:
61 raise ValueError("Invalid request: missing method field")
62
63 return JSONRPCRequest(
64 method=obj["method"],
65 params=obj.get("params", {}),
66 id=obj.get("id"),
67 jsonrpc=obj["jsonrpc"],
68 )
69
70
71def build_jsonrpc_response(
72 result: dict | None = None,
73 error: dict | None = None,
74 request_id: int | str | None = None,
75) -> dict[str, Any]:
76 """Build a JSON-RPC 2.0 response dict."""
77 resp: dict[str, Any] = {"jsonrpc": "2.0", "id": request_id}
78 if error is not None:
79 resp["error"] = error
80 else:
81 resp["result"] = result
82 return resp
83
84
85def build_jsonrpc_error(
86 code: int,
87 message: str,
88 request_id: int | str | None = None,
89) -> dict[str, Any]:
90 """Build a JSON-RPC 2.0 error response."""
91 return build_jsonrpc_response(
92 error={"code": code, "message": message},
93 request_id=request_id,
94 )