/
app.py
131 lines (99 loc) · 4.25 KB
/
app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
# flake8: noqa: E402
import os
import time
from typing import Optional
from flask import Flask, g, request
from flask.cli import load_dotenv
from flask_mail import Mail
from flask_sslify import SSLify
from flask_json import FlaskJSON
from flask_cors import CORS
from redis import Redis
from rq import Queue
def create(env: Optional[str] = None, path_to_config: Optional[str] = None) -> Flask:
"""
Create a Flask app and configure it.
Set the environment by setting FLASK_ENV as environment variable (also possible in .env).
Or, overwrite any FLASK_ENV setting by passing an env in directly (useful for testing for instance).
A path to a config file can be passed in (otherwise a config file will be searched in the home or instance directories)
"""
from flexmeasures.utils import config_defaults
from flexmeasures.utils.config_utils import read_config, configure_logging
from flexmeasures.utils.app_utils import set_secret_key
from flexmeasures.utils.error_utils import add_basic_error_handlers
# Create app
configure_logging() # do this first, see http://flask.pocoo.org/docs/dev/logging/
# we're loading dotenv files manually & early (can do Flask.run(load_dotenv=False)),
# as we need to know the ENV now (for it to be recognised by Flask()).
load_dotenv()
app = Flask("flexmeasures")
if env is not None: # overwrite
app.env = env
if env == "testing":
app.testing = True
if env == "development":
app.debug = config_defaults.DevelopmentConfig.DEBUG
# App configuration
read_config(app, path_to_config=path_to_config)
add_basic_error_handlers(app)
app.mail = Mail(app)
FlaskJSON(app)
cors = CORS(app)
# configure Redis (for redis queue)
if app.testing:
from fakeredis import FakeStrictRedis
app.queues = dict(
forecasting=Queue(connection=FakeStrictRedis(), name="forecasting"),
scheduling=Queue(connection=FakeStrictRedis(), name="scheduling"),
)
else:
redis_conn = Redis(
app.config["FLEXMEASURES_REDIS_URL"],
port=app.config["FLEXMEASURES_REDIS_PORT"],
db=app.config["FLEXMEASURES_REDIS_DB_NR"],
password=app.config["FLEXMEASURES_REDIS_PASSWORD"],
)
""" FWIW, you could use redislite like this (not on non-recent os.name=="nt" systems or PA, sadly):
from redislite import Redis
redis_conn = Redis("MY-DB-NAME", unix_socket_path="/tmp/my-redis.socket",
)
"""
app.queues = dict(
forecasting=Queue(connection=redis_conn, name="forecasting"),
scheduling=Queue(connection=redis_conn, name="scheduling"),
)
# Some basic security measures
if not app.env == "documentation":
set_secret_key(app)
if app.config.get("SECURITY_PASSWORD_SALT", None) is None:
app.config["SECURITY_PASSWORD_SALT"] = app.config["SECRET_KEY"]
if not app.env in ("documentation", "development"):
SSLify(app)
# Register database and models, including user auth security handlers
from flexmeasures.data import register_at as register_db_at
register_db_at(app)
# Register the API
from flexmeasures.api import register_at as register_api_at
register_api_at(app)
# Register plugins
from flexmeasures.utils.app_utils import register_plugins
register_plugins(app)
# Register the UI
# If plugins registered routes already (e.g. "/"),
# they have precedence (first registration wins).
from flexmeasures.ui import register_at as register_ui_at
register_ui_at(app)
# Profile endpoints (if needed, e.g. during development)
@app.before_request
def before_request():
if app.config.get("FLEXMEASURES_PROFILE_REQUESTS", False):
g.start = time.time()
@app.teardown_request
def teardown_request(exception=None):
if app.config.get("FLEXMEASURES_PROFILE_REQUESTS", False):
diff = time.time() - g.start
if all([kw not in request.url for kw in ["/static", "favicon.ico"]]):
app.logger.info(
f"[PROFILE] {str(round(diff, 2)).rjust(6)} seconds to serve {request.url}."
)
return app