Coverage for flogin/default_events.py: 100%

1from __future__ import annotations 

2 

3import logging 

4from typing import TYPE_CHECKING, Any 

5 

6from .jsonrpc import ErrorResponse, QueryResponse 

7from .query import Query, RawQuery 

8from .settings import Settings 

9 

10if TYPE_CHECKING: 

11 from collections.abc import Awaitable, Callable 

12 

13 from .plugin import Plugin 

14 

15 

16log = logging.getLogger(__name__) 

17 

18__all__ = ("on_error",) 

19 

20 

21async def on_error( 

22 event_method: str, error: Exception, *args: Any, **kwargs: Any 

23) -> ErrorResponse: 

24 """gets called when an error occurs in an event""" 

25 log.exception("Ignoring exception in event %r", event_method, exc_info=error) 

26 return ErrorResponse.internal_error(error) 

27 

28 

29def get_default_events(plugin: Plugin[Any]) -> dict[str, Callable[..., Awaitable[Any]]]: 

30 def on_query( 

31 data: RawQuery, raw_settings: dict[str, Any] 

32 ) -> Awaitable[ErrorResponse | QueryResponse]: 

33 query = Query(data, plugin) 

34 plugin._last_query = query 

35 plugin._results.clear() 

36 

37 if plugin._settings_are_populated is False: 

38 log.info("Settings have not been populated yet, creating a new instance") 

39 plugin._settings_are_populated = True 

40 plugin.settings = Settings( 

41 raw_settings, no_update=plugin.options.get("settings_no_update", False) 

42 ) 

43 else: 

44 plugin.settings._update(raw_settings) 

45 return plugin.process_search_handlers(query) 

46 

47 def on_context_menu(data: list[str]) -> Awaitable[ErrorResponse | QueryResponse]: 

48 return plugin.process_context_menus(data) 

49 

50 return { 

51 event.__name__: event 

52 for event in ( 

53 on_error, 

54 on_query, 

55 on_context_menu, 

56 plugin._initialize_wrapper, 

57 ) 

58 }