Coverage for flogin/search_handler.py: 82%

1from __future__ import annotations 

2 

3import logging 

4import re 

5from typing import TYPE_CHECKING, Any, Generic, TypeVar, overload 

6 

7from ._types.search_handlers import ( 

8 PluginT, 

9 SearchHandlerCallbackReturns, 

10 SearchHandlerCondition, 

11) 

12from .conditions import KeywordCondition, PlainTextCondition, RegexCondition 

13from .jsonrpc import ErrorResponse 

14from .utils import MISSING, copy_doc, decorator 

15 

16if TYPE_CHECKING: 

17 from collections.abc import Callable, Iterable 

18 

19 from .query import Query 

20 

21 ErrorHandlerT = TypeVar( 

22 "ErrorHandlerT", 

23 bound=Callable[[Query, Exception], SearchHandlerCallbackReturns], 

24 ) 

25 

26log = logging.getLogger(__name__) 

27 

28__all__ = ("SearchHandler",) 

29 

30 

31class SearchHandler(Generic[PluginT]): 

32 r"""This represents a search handler. 

33 

34 When creating this on your own, the :func:`~flogin.plugin.Plugin.register_search_handler` method can be used to register it. 

35 

36 See the :ref:`search handler section <search_handlers>` for more information about using search handlers. 

37 

38 There is a provided decorator to easily create search handlers: :func:`~flogin.plugin.Plugin.search` 

39 

40 The keywords in the constructor can also be passed into the subclassed init, like so: :: 

41 

42 class MyHandler(SearchHandler, text="text"): 

43 ... 

44 

45 # is equal to 

46 

47 class MyHandler(SearchHandler): 

48 def __init__(self): 

49 super().__init__(text="text") 

50 

51 Parameters 

52 ---------- 

53 condition: Optional[:ref:`condition <condition_example>`] 

54 The condition to determine which queries this handler should run on. If given, this should be the only argument given. 

55 text: Optional[:class:`str`] 

56 A kwarg to quickly add a :class:`~flogin.conditions.PlainTextCondition`. If given, this should be the only argument given. 

57 pattern: Optional[:class:`re.Pattern` | :class:`str`] 

58 A kwarg to quickly add a :class:`~flogin.conditions.RegexCondition`. If given, this should be the only argument given. 

59 keyword: Optional[:class:`str`] 

60 A kwarg to quickly set the condition to a :class:`~flogin.conditions.KeywordCondition` condition with the ``keyword`` kwarg being the only allowed keyword. 

61 allowed_keywords: Optional[Iterable[:class:`str`]] 

62 A kwarg to quickly set the condition to a :class:`~flogin.conditions.KeywordCondition` condition with the kwarg being the list of allowed keywords. 

63 disallowed_keywords: Optional[Iterable[:class:`str`]] 

64 A kwarg to quickly set the condition to a :class:`~flogin.conditions.KeywordCondition` condition with the kwarg being the list of disallowed keywords. 

65 

66 Attributes 

67 ------------ 

68 plugin: :class:`~flogin.plugin.Plugin` | None 

69 Your plugin instance. This is filled before :func:`~flogin.search_handler.SearchHandler.callback` is triggered. 

70 """ 

71 

72 @overload 

73 def __init__(self, condition: SearchHandlerCondition) -> None: ... 

74 

75 @overload 

76 def __init__(self, *, text: str) -> None: ... 

77 

78 @overload 

79 def __init__( 

80 self, 

81 *, 

82 pattern: re.Pattern[str] | str = MISSING, 

83 ) -> None: ... 

84 

85 @overload 

86 def __init__( 

87 self, 

88 *, 

89 keyword: str = MISSING, 

90 ) -> None: ... 

91 

92 @overload 

93 def __init__( 

94 self, 

95 *, 

96 allowed_keywords: Iterable[str] = MISSING, 

97 ) -> None: ... 

98 

99 @overload 

100 def __init__( 

101 self, 

102 *, 

103 disallowed_keywords: Iterable[str] = MISSING, 

104 ) -> None: ... 

105 

106 def __init__( 

107 self, 

108 condition: SearchHandlerCondition | None = None, 

109 *, 

110 text: str = MISSING, 

111 pattern: re.Pattern[str] | str = MISSING, 

112 keyword: str = MISSING, 

113 allowed_keywords: Iterable[str] = MISSING, 

114 disallowed_keywords: Iterable[str] = MISSING, 

115 ) -> None: 

116 if condition is None: 

117 condition = self._builtin_condition_kwarg_to_obj( 

118 text=text, 

119 pattern=pattern, 

120 keyword=keyword, 

121 allowed_keywords=allowed_keywords, 

122 disallowed_keywords=disallowed_keywords, 

123 ) 

124 if condition: 

125 setattr(self, "condition", condition) 

126 

127 self.plugin: PluginT | None = None 

128 

129 @overload 

130 def __init_subclass__(cls: type[SearchHandler], *, text: str) -> None: ... 

131 

132 @overload 

133 def __init_subclass__( 

134 cls: type[SearchHandler], 

135 *, 

136 pattern: re.Pattern[str] | str = MISSING, 

137 ) -> None: ... 

138 

139 @overload 

140 def __init_subclass__( 

141 cls: type[SearchHandler], 

142 *, 

143 keyword: str = MISSING, 

144 ) -> None: ... 

145 

146 @overload 

147 def __init_subclass__( 

148 cls: type[SearchHandler], 

149 *, 

150 allowed_keywords: Iterable[str] = MISSING, 

151 ) -> None: ... 

152 

153 @overload 

154 def __init_subclass__( 

155 cls: type[SearchHandler], 

156 *, 

157 disallowed_keywords: Iterable[str] = MISSING, 

158 ) -> None: ... 

159 

160 def __init_subclass__( 

161 cls: type[SearchHandler], 

162 *, 

163 text: str = MISSING, 

164 pattern: re.Pattern[str] | str = MISSING, 

165 keyword: str = MISSING, 

166 allowed_keywords: Iterable[str] = MISSING, 

167 disallowed_keywords: Iterable[str] = MISSING, 

168 ) -> None: 

169 con = cls._builtin_condition_kwarg_to_obj( 

170 text=text, 

171 pattern=pattern, 

172 keyword=keyword, 

173 allowed_keywords=allowed_keywords, 

174 disallowed_keywords=disallowed_keywords, 

175 ) 

176 if con is not None: 

177 setattr(cls, "condition", con) 

178 

179 @classmethod 

180 def _builtin_condition_kwarg_to_obj( 

181 cls: type[SearchHandler], 

182 *, 

183 text: str = MISSING, 

184 pattern: re.Pattern[str] | str = MISSING, 

185 keyword: str = MISSING, 

186 allowed_keywords: Iterable[str] = MISSING, 

187 disallowed_keywords: Iterable[str] = MISSING, 

188 ) -> SearchHandlerCondition | None: 

189 if text is not MISSING: 

190 return PlainTextCondition(text) 

191 if pattern is not MISSING: 

192 if isinstance(pattern, str): 

193 pattern = re.compile(pattern) 

194 return RegexCondition(pattern) 

195 if keyword is not MISSING: 

196 return KeywordCondition(allowed_keywords=[keyword]) 

197 if allowed_keywords is not MISSING: 

198 return KeywordCondition(allowed_keywords=allowed_keywords) 

199 if disallowed_keywords is not MISSING: 

200 return KeywordCondition(disallowed_keywords=disallowed_keywords) 

201 

202 def condition(self, query: Query) -> bool: 

203 r"""A function that determines whether or not to fire off this search handler for a given query 

204 

205 Parameters 

206 ---------- 

207 query: :class:`~flogin.query.Query` 

208 The query object for the query request 

209 

210 Returns 

211 -------- 

212 :class:`bool` 

213 Whether or not to fire off this handler for the given query. 

214 """ 

215 

216 return True 

217 

218 def callback(self, query: Query) -> SearchHandlerCallbackReturns: 

219 r"""|coro| 

220 

221 Override this function to add the search handler behavior you want for the set condition. 

222 

223 This method can return/yield almost anything, and flogin will convert it into a list of :class:`~flogin.jsonrpc.results.Result` objects before sending it to flow. 

224 

225 Returns 

226 ------- 

227 list[:class:`~flogin.jsonrpc.results.Result`] | :class:`~flogin.jsonrpc.results.Result` | str | Any 

228 A list of results, an results, or something that can be converted into a list of results. 

229 

230 Yields 

231 ------ 

232 :class:`~flogin.jsonrpc.results.Result` | str | Any 

233 A result object or something that can be converted into a result object. 

234 """ 

235 ... 

236 

237 def on_error(self, query: Query, error: Exception) -> SearchHandlerCallbackReturns: 

238 r"""|coro| 

239 

240 Override this function to add an error response behavior to this handler's callback. 

241 

242 If the error was handled: 

243 You can return/yield almost anything, and flogin will convert it into a list of :class:`~flogin.jsonrpc.results.Result` objects before sending it to flow. 

244 

245 If the error was not handled: 

246 Return a :class:`~flogin.jsonrpc.responses.ErrorResponse` object 

247 

248 Parameters 

249 ---------- 

250 query: :class:`~flogin.query.Query` 

251 The query that was being handled when the error occured. 

252 error: :class:`Exception` 

253 The error that occured 

254 

255 Returns 

256 ------- 

257 :class:`~flogin.jsonrpc.responses.ErrorResponse` | list[:class:`~flogin.jsonrpc.results.Result`] | :class:`~flogin.jsonrpc.results.Result` | str | Any 

258 A list of results, an results, or something that can be converted into a list of results. 

259 

260 Yields 

261 ------ 

262 :class:`~flogin.jsonrpc.results.Result` | str | Any 

263 A result object or something that can be converted into a result object. 

264 """ 

265 ... 

266 

267 if not TYPE_CHECKING: 

268 

269 @copy_doc(callback) 

270 async def callback(self, query: Query) -> Any: 

271 raise RuntimeError("Callback was not overriden") 

272 

273 @copy_doc(on_error) 

274 async def on_error(self, query: Query, error: Exception) -> Any: 

275 log.exception( 

276 "Ignoring exception in search handler callback (%r)", 

277 self, 

278 exc_info=error, 

279 ) 

280 return ErrorResponse.internal_error(error) 

281 

282 @property 

283 def name(self) -> str: 

284 """:class:`str`: The name of the search handler's callback""" 

285 return self.callback.__name__ 

286 

287 @decorator 

288 def error(self, func: ErrorHandlerT) -> ErrorHandlerT: 

289 """A decorator that registers a error handler for this search handler. 

290 

291 For more information see :class:`~flogin.search_handler.SearchHandler.on_error` 

292 

293 Example 

294 --------- 

295 

296 .. code-block:: python3 

297 

298 @plugin.search() 

299 async def my_hander(query): 

300 .. 

301 

302 @my_handler.error 

303 async def my_error_handler(query, error): 

304 ... 

305 

306 """ 

307 

308 setattr(self, "on_error", func) 

309 return func