Coverage for postrfp / shared / init / sysconfig.py: 80%

1import os 

2import time 

3import logging 

4import logging.config 

5 

6from postrfp import conf 

7from postrfp.conf.settings import AppSettings 

8from postrfp.shared.constants import RunMode 

9 

10 

11config_log = logging.getLogger(__name__) 

12 

13 

14def configure_postrfp(env_file=None) -> "AppSettings": 

15 from postrfp.jobs.executor import init_jobs_executor 

16 from postrfp.jobs.events.listeners import init_event_listeners 

17 from postrfp.model.audit import visible 

18 

19 if env_file: 

20 # This is valid for Pydantic 2. MyPy error is presumably a bug 

21 conf.CONF = AppSettings(_env_file=env_file) # type: ignore 

22 else: 

23 conf.CONF = AppSettings() 

24 

25 os.environ["TZ"] = "UTC" 

26 time.tzset() 

27 

28 init_jobs_executor() 

29 

30 # Deploy process_event DAG if using Dagu executor 

31 if conf.CONF.task_executor == "dagu": 

32 deploy_process_event_dag_safely() 

33 

34 configure_logging() 

35 init_event_listeners() 

36 log_setup_details(env_file) 

37 visible.check_all_events_assigned() 

38 return conf.CONF 

39 

40 

41def deploy_process_event_dag_safely() -> None: 

42 """ 

43 Deploy the process_event DAG to Dagu, with error handling. 

44 

45 Failures are logged but don't prevent startup, since the DAG 

46 might already exist or Dagu might not be available yet. 

47 """ 

48 try: 

49 from postrfp.jobs.dagu import deploy_process_event_dag 

50 

51 deploy_process_event_dag() 

52 except Exception as e: 

53 config_log.warning( 

54 "Could not deploy process_event DAG (Dagu may not be ready): %s", e 

55 ) 

56 

57 

58def log_setup_details(env_file): 

59 config_log.info("\n postrfp configured: ") 

60 if env_file is None: 

61 config_log.info("-- No environment file provided, using default config") 

62 else: 

63 config_log.info("-- using settings env file: %s", env_file) 

64 config_log.info("-- run_mode: %s", conf.CONF.run_mode) 

65 config_log.info("-- db_url: %s", conf.CONF.conn_string()) 

66 config_log.info("-- task_executor: %s", conf.CONF.task_executor) 

67 config_log.info("-- app_base_url: %s", conf.CONF.app_base_url) 

68 if conf.CONF.task_executor == "dagu": 

69 config_log.info("-- dagu_base_url: %s", conf.CONF.dagu_base_url) 

70 config_log.info("-- mailer: %s" % conf.CONF.mailer) 

71 config_log.info("-- email_to_override: %s" % conf.CONF.email_to_override) 

72 if conf.CONF.postmark_api_key: 

73 config_log.info("-- postmark-api-key set") 

74 

75 

76def configure_logging(log_config_dict=None) -> None: # pragma: no cover 

77 if log_config_dict is not None: 

78 logging.config.dictConfig(log_config_dict) 

79 return 

80 from ...conf import logconf 

81 

82 # fmt: off 

83 match conf.CONF.run_mode: 

84 

85 case RunMode.production: 

86 email = "errors@supplierselect.com" 

87 logconf.PRODUCTION["handlers"]["email"]["fromaddr"] = email 

88 subject = "POSTRFP Error on %s" % conf.CONF.webapp_hostname 

89 logconf.PRODUCTION["handlers"]["email"]["subject"] = subject 

90 logging.config.dictConfig(logconf.PRODUCTION) 

91 

92 case RunMode.development: 

93 logging.config.dictConfig(logconf.DEVELOPMENT) 

94 

95 case RunMode.test: 

96 logging.config.dictConfig(logconf.TEST) 

97 

98 case _: 

99 raise ValueError("Cannot configure logging without valid RunMode")