Coverage for src/local_deep_research/notifications/apprise_log_utils.py: 100%

1import logging 

2from collections.abc import Callable 

3from typing import Final 

4 

5 

6APPRISE_DIAGNOSTIC: Final = ( 

7 "Apprise emitted a diagnostic; details suppressed to protect " 

8 "notification credentials." 

9) 

10 

11 

12class _AppriseSanitizingRecordFactory[**P]: 

13 def __init__(self, delegate: Callable[P, logging.LogRecord]) -> None: 

14 self._delegate = delegate 

15 

16 def __call__(self, *args: P.args, **kwargs: P.kwargs) -> logging.LogRecord: 

17 record = self._delegate(*args, **kwargs) 

18 if record.name == "apprise" or record.name.startswith("apprise."): 

19 record.msg = APPRISE_DIAGNOSTIC 

20 record.args = () 

21 record.exc_info = None 

22 record.exc_text = None 

23 record.stack_info = None 

24 return record 

25 

26 

27def install_apprise_log_record_factory() -> None: 

28 current_factory = logging.getLogRecordFactory() 

29 if isinstance(current_factory, _AppriseSanitizingRecordFactory): 

30 return 

31 logging.setLogRecordFactory( 

32 _AppriseSanitizingRecordFactory(current_factory) 

33 )