Coverage for flogin/jsonrpc/responses.py: 75%

1from __future__ import annotations 

2 

3import json 

4from typing import TYPE_CHECKING, Any, Generic, TypeVar, cast 

5 

6from ..utils import MISSING 

7from .base_object import ToMessageBase 

8from .enums import ErrorCode 

9 

10if TYPE_CHECKING: 

11 from .._types.jsonrpc.responses import ( 

12 ErrorPayload, 

13 ) 

14 from .._types.jsonrpc.responses import ( 

15 RawErrorResponse as ErrorResponsePayload, 

16 ) 

17 from .._types.jsonrpc.responses import ( 

18 RawExecuteResponse as ExecuteResponsePayload, 

19 ) 

20 from .._types.jsonrpc.responses import ( 

21 RawQueryResponse as QueryResponsePayload, 

22 ) 

23 from .results import Result 

24 

25T = TypeVar("T") 

26 

27__all__ = ( 

28 "ErrorResponse", 

29 "ExecuteResponse", 

30 "QueryResponse", 

31) 

32 

33 

34class BaseResponse(ToMessageBase[T], Generic[T]): 

35 r"""This represents a response to flow. 

36 

37 .. WARNING:: 

38 This class is NOT to be used as is. Use one of it's subclasses instead. 

39 """ 

40 

41 def to_message(self, id: int) -> bytes: 

42 return ( 

43 json.dumps( 

44 { 

45 "jsonrpc": "2.0", 

46 "id": id, 

47 } 

48 | cast("dict[Any, Any]", self.to_dict()) 

49 ) 

50 + "\r\n" 

51 ).encode() 

52 

53 

54class ErrorResponse(BaseResponse["ErrorResponsePayload"]): 

55 r"""This represents an error sent to or from flow. 

56 

57 Attributes 

58 -------- 

59 code: :class:`int` 

60 The error code for the error 

61 message: :class:`str` 

62 The error's message 

63 data: Optional[Any] 

64 Any extra data 

65 """ 

66 

67 __slots__ = "code", "data", "message" 

68 

69 def __init__(self, code: int, message: str, data: Any | None = None) -> None: 

70 self.code = code 

71 self.message = message 

72 self.data = data 

73 

74 def to_dict(self) -> ErrorResponsePayload: 

75 data = self.data 

76 if isinstance(data, Exception): 

77 data = f"{data}" 

78 return {"error": {"code": self.code, "message": self.message, "data": data}} 

79 

80 @classmethod 

81 def from_dict( 

82 cls: type[ErrorResponse], data: ErrorPayload | ErrorResponsePayload 

83 ) -> ErrorResponse: 

84 if "error" in data: 

85 data = data["error"] 

86 return cls(code=data["code"], message=data["message"], data=data.get("data")) 

87 

88 @classmethod 

89 def internal_error(cls: type[ErrorResponse], data: Any = None) -> ErrorResponse: 

90 return cls( 

91 code=ErrorCode.server_error_start.value, message="Internal error", data=data 

92 ) 

93 

94 

95class QueryResponse(BaseResponse["QueryResponsePayload"]): 

96 r"""This response represents the response from search handler's callbacks and context menus. See the :ref:`search handler section <search_handlers>` for more information about using search handlers. 

97 

98 Attributes 

99 -------- 

100 results: list[:class:`~flogin.jsonrpc.results.Result`] 

101 The results to be sent as the result of the query 

102 settings_changes: dict[:class:`str`, Any] 

103 Any changes to be made to the plugin's settings. 

104 debug_message: :class:`str` 

105 A debug message if you want 

106 """ 

107 

108 __slots__ = "debug_message", "results", "settings_changes" 

109 

110 def __init__( 

111 self, 

112 results: list[Result], 

113 settings_changes: dict[str, Any] | None = None, 

114 debug_message: str = MISSING, 

115 ) -> None: 

116 self.results = results 

117 self.settings_changes = settings_changes or {} 

118 self.debug_message = debug_message or "" 

119 

120 def to_dict(self) -> QueryResponsePayload: 

121 return { 

122 "result": { 

123 "settingsChange": self.settings_changes, 

124 "debugMessage": self.debug_message or "", 

125 "result": [res.to_dict() for res in self.results], 

126 } 

127 } 

128 

129 

130class ExecuteResponse(BaseResponse["ExecuteResponsePayload"]): 

131 r"""This response is a generic response for jsonrpc requests, most notably result callbacks. 

132 

133 Attributes 

134 -------- 

135 hide: :class:`bool` 

136 Whether to hide the flow menu after execution or not 

137 """ 

138 

139 __slots__ = ("hide",) 

140 

141 def __init__(self, hide: bool = True) -> None: 

142 self.hide = hide 

143 

144 def to_dict(self) -> ExecuteResponsePayload: 

145 return {"result": {"hide": self.hide}}