Coverage for hiperta_stream/utils/logging_utils.py: 82%

22 statements  

« prev     ^ index     » next       coverage.py v7.4.3, created at 2024-07-16 10:16 +0000

1import logging 

2import sys 

3from enum import Enum 

4 

5# Mapping between logging levels exposed to users, and logging levels in software 

6# Users level should eventually follow ACADA AE Logging ICD https://redmine.cta-observatory.org/dmsf/files/14915/view 

7LOGGING_LEVELS_DICT = { 

8 "DEBUG": logging.DEBUG, 

9 "INFO": logging.INFO, 

10 "WARNING": logging.WARNING, 

11 "ERROR": logging.ERROR, 

12 "CRITICAL": logging.CRITICAL, 

13} 

14 

15# Enum class of logging levels, to constraint the variable type in configuration json schema. 

16LOGGING_LEVELS = Enum("LOGGING_LEVELS", {v: v for v in LOGGING_LEVELS_DICT.keys()}) 

17 

18 

19def log_uncaught_exceptions(): 

20 """Makes all uncaught exception to be logged by the default logger. 

21 

22 Keyboard exceptions and children classes are not logged so one can kill the program with ctr+C. 

23 """ 

24 

25 def handle_exception(exc_type, exc_value, exc_traceback): 

26 if not issubclass(exc_type, KeyboardInterrupt): 

27 logging.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback)) 

28 

29 sys.__excepthook__(exc_type, exc_value, exc_traceback) 

30 return 

31 

32 sys.excepthook = handle_exception 

33 

34 

35def init_logging(log_level="DEBUG", log_filename="hiperta_stream_start.log"): 

36 """(Re-)initialize all loggers""" 

37 

38 logging.captureWarnings(True) # log all warnings from the warnings module. 

39 log_uncaught_exceptions() # log all uncaught exceptions as well 

40 

41 logging_format = "%(asctime)s - %(levelname)s - SAG RECO MANAGER - %(filename)s:%(lineno)s - %(message)s" 

42 logging_level = LOGGING_LEVELS(log_level) 

43 # This is deactivated as it makes logging call randomly deadlock in jenkins CI machine 

44 # TODO: issue 

45 # output to stderr 

46 # handlers = [logging.StreamHandler()] # output to stderr 

47 handlers = [] 

48 if log_filename: # and also output to file if asked 

49 handlers.append(logging.FileHandler(log_filename)) 

50 logging.basicConfig( 

51 level=logging_level.name, 

52 format=logging_format, 

53 handlers=handlers, 

54 force=True, 

55 ) 

56 

57 logging.info("Logging configured - start logging")