Coverage for flask/src/app.py: 82%

1#!/usr/bin/env python3 

2""" 

3水やりを自動化するアプリのサーバーです 

4 

5Usage: 

6 app.py [-c CONFIG] [-p PORT] [-D] [-d] 

7 

8Options: 

9 -c CONFIG : CONFIG を設定ファイルとして読み込んで実行します。[default: config.yaml] 

10 -p PORT : WEB サーバのポートを指定します。[default: 5000] 

11 -d : ダミーモードで実行します。CI テストで利用することを想定しています。 

12 -D : デバッグモードで動作します。 

13""" 

14 

15import atexit 

16import logging 

17import os 

18import signal 

19 

20import flask_cors 

21 

22import flask 

23 

24import my_lib.webapp.base 

25import my_lib.webapp.event 

26import my_lib.webapp.log 

27import my_lib.webapp.util 

28import my_lib.proc_util 

29import rasp_water.control.webapi.schedule 

30import rasp_water.control.webapi.valve 

31import rasp_water.control.webapi.test.time 

32import rasp_water.metrics.webapi.page 

33 

34 

35SCHEMA_CONFIG = "config.schema" 

36 

37def term(): 

38 import rasp_water.control.scheduler 

39 

40 rasp_water.control.scheduler.term() 

41 

42 # 子プロセスを終了 

43 my_lib.proc_util.kill_child() 

44 

45 # プロセス終了 

46 logging.info("Graceful shutdown completed") 

47 os._exit(0) 

48 

49 

50def sig_handler(num, frame): # noqa: ARG001 

51 global should_terminate 

52 

53 logging.warning("receive signal %d", num) 

54 

55 if num == signal.SIGTERM: 

56 term() 

57 

58 

59def create_app(config, dummy_mode=False): 

60 # NOTE: オプションでダミーモードが指定された場合、環境変数もそれに揃えておく 

61 if dummy_mode: 

62 os.environ["DUMMY_MODE"] = "true" 

63 else: # pragma: no cover 

64 os.environ["DUMMY_MODE"] = "false" 

65 

66 # NOTE: テストのため、環境変数 DUMMY_MODE をセットしてからロードしたいのでこの位置 

67 import my_lib.webapp.config 

68 

69 my_lib.webapp.config.URL_PREFIX = "/rasp-water" 

70 my_lib.webapp.config.init(config) 

71 

72 app = flask.Flask("rasp-water") 

73 

74 # NOTE: アクセスログは無効にする 

75 logging.getLogger("werkzeug").setLevel(logging.ERROR) 

76 

77 if os.environ.get("WERKZEUG_RUN_MAIN") == "true": 

78 if dummy_mode: 

79 logging.warning("Set dummy mode") 

80 else: # pragma: no cover 

81 pass 

82 

83 rasp_water.control.webapi.schedule.init(config) 

84 rasp_water.control.webapi.valve.init(config) 

85 my_lib.webapp.log.init(config) 

86 

87 def notify_terminate(): # pragma: no cover 

88 rasp_water.control.valve.set_state(rasp_water.control.valve.VALVE_STATE.CLOSE) 

89 my_lib.webapp.log.info("🏃 アプリを再起動します。") 

90 my_lib.webapp.log.term() 

91 

92 atexit.register(notify_terminate) 

93 else: # pragma: no cover 

94 pass 

95 

96 flask_cors.CORS(app) 

97 

98 app.config["CONFIG"] = config 

99 app.config["DUMMY_MODE"] = dummy_mode 

100 

101 app.json.compat = True 

102 

103 app.register_blueprint(rasp_water.control.webapi.valve.blueprint, url_prefix=my_lib.webapp.config.URL_PREFIX) 

104 app.register_blueprint(rasp_water.control.webapi.schedule.blueprint, url_prefix=my_lib.webapp.config.URL_PREFIX) 

105 app.register_blueprint(rasp_water.metrics.webapi.page.blueprint, url_prefix=my_lib.webapp.config.URL_PREFIX) 

106 

107 if dummy_mode: 107 ↛ 110line 107 didn't jump to line 110 because the condition on line 107 was always true

108 app.register_blueprint(rasp_water.control.webapi.test.time.blueprint, url_prefix=my_lib.webapp.config.URL_PREFIX) 

109 

110 app.register_blueprint(my_lib.webapp.base.blueprint_default) 

111 app.register_blueprint(my_lib.webapp.base.blueprint, url_prefix=my_lib.webapp.config.URL_PREFIX) 

112 app.register_blueprint(my_lib.webapp.event.blueprint, url_prefix=my_lib.webapp.config.URL_PREFIX) 

113 app.register_blueprint(my_lib.webapp.log.blueprint, url_prefix=my_lib.webapp.config.URL_PREFIX) 

114 app.register_blueprint(my_lib.webapp.util.blueprint, url_prefix=my_lib.webapp.config.URL_PREFIX) 

115 

116 my_lib.webapp.config.show_handler_list(app) 

117 

118 return app 

119 

120 

121if __name__ == "__main__": 

122 import pathlib 

123 

124 import docopt 

125 import my_lib.config 

126 import my_lib.logger 

127 

128 args = docopt.docopt(__doc__) 

129 

130 config_file = args["-c"] 

131 port = args["-p"] 

132 dummy_mode = args["-d"] 

133 debug_mode = args["-D"] 

134 

135 my_lib.logger.init("hems.rasp-water", level=logging.DEBUG if debug_mode else logging.INFO) 

136 

137 config = my_lib.config.load(config_file, pathlib.Path(SCHEMA_CONFIG)) 

138 

139 app = create_app(config, dummy_mode) 

140 

141 signal.signal(signal.SIGTERM, sig_handler) 

142 

143 # Flaskアプリケーションを実行 

144 try: 

145 # NOTE: スクリプトの自動リロード停止したい場合は use_reloader=False にする 

146 app.run(host="0.0.0.0", port=port, threaded=True, use_reloader=True, debug=debug_mode) # noqa: S104 

147 except KeyboardInterrupt: 

148 logging.info("Received KeyboardInterrupt, shutting down...") 

149 sig_handler(signal.SIGINT, None) 

150 finally: 

151 term()