Coverage for yasmon/cli.py: 94%

1from yasmon.tasks import TaskRunner 

2from yasmon.tasks import TaskError 

3from yasmon.processor import YAMLProcessor 

4from yasmon.processor import YAMLSyntaxError 

5from yasmon.callbacks import CallbackSyntaxError 

6from yasmon.loggers import LoggerSyntaxError 

7 

8from loguru import logger 

9import argparse 

10from pathlib import Path 

11from enum import IntEnum 

12import sys 

13 

14logger.remove(0) 

15 

16 

17class ExitCodes(IntEnum): 

18 Success = 0 # noqa 

19 UnexpectedException = 1 # noqa 

20 YAMLSyntaxError = 2 # noqa 

21 FileNotFoundError = 3 # noqa 

22 TaskNotImplementedError = 4 # noqa 

23 CallbackNotImplementedError = 5 # noqa 

24 CallbackSyntaxError = 6 # noqa 

25 TaskError = 7 # noqa 

26 LoggerSyntaxError = 8 # noqa 

27 

28 

29def parse_args(*args): 

30 parser = argparse.ArgumentParser() 

31 parser.add_argument("--config", type=str, 

32 help="yaml config file path", 

33 default=str(Path(Path.home(), 

34 '.config', 

35 'yasmon', 

36 'config.yaml'))) 

37 return parser.parse_args(args) 

38 

39 

40def setup_default_logger(): 

41 journal_logger_format = ( 

42 "<level>{level: <8}</level> | " 

43 "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | " 

44 "<level>{message}</level>" 

45 ) 

46 return logger.add(sys.stderr, format=journal_logger_format, level='DEBUG') 

47 

48 

49def execute(args, default_logger_id): 

50 try: 

51 processor = YAMLProcessor() 

52 processor.load_file(args.config) 

53 

54 loggers = processor.add_loggers() 

55 num_usr_loggers = len(loggers) 

56 if num_usr_loggers > 0: 

57 logger.remove(default_logger_id) 

58 logger.info(f'there are {num_usr_loggers} user loggers defined: ' 

59 'default stderr logger removed') 

60 else: 

61 logger.warning('there are no user loggers defined: ' 

62 'keeping default stderr logger') 

63 

64 callbacks = processor.get_callbacks() 

65 tasks = processor.get_tasks(callbacks) 

66 runner = TaskRunner(tasks) 

67 runner.loop.run_until_complete(runner()) 

68 return ExitCodes.Success 

69 except YAMLSyntaxError as err: 

70 logger.error(f'{err.__class__.__name__}: {err}') 

71 return ExitCodes.YAMLSyntaxError 

72 except FileNotFoundError as err: 

73 logger.error(f'{err.__class__.__name__}: {err}') 

74 return ExitCodes.FileNotFoundError 

75 except CallbackSyntaxError as err: 

76 logger.error(f'{err.__class__.__name__}: {err}') 

77 return ExitCodes.CallbackSyntaxError 

78 except TaskError as err: 

79 logger.error(f'{err.__class__.__name__}: {err}') 

80 return ExitCodes.TaskError 

81 except LoggerSyntaxError as err: 

82 logger.error(f'{err.__class__.__name__}: {err}') 

83 return ExitCodes.LoggerSyntaxError 

84 except Exception as err: 

85 logger.error(f'Unexpected {err.__class__.__name__}: {err}') 

86 return ExitCodes.UnexpectedException 

87 

88 

89def main(): 

90 default_logger_id = setup_default_logger() 

91 args = parse_args(*sys.argv[1:]) 

92 return execute(args, default_logger_id) 

93 

94 

95if __name__ == '__main__': 

96 sys.exit(main())