135 lines
4.5 KiB
Python
135 lines
4.5 KiB
Python
from flask import Flask
|
|
|
|
import os
|
|
import sys
|
|
import logging
|
|
from . import local_db
|
|
|
|
from .api import shipcalls
|
|
from .api import participant
|
|
from .api import times
|
|
from .api import notifications
|
|
from .api import berths
|
|
from .api import ships
|
|
from .api import login
|
|
from .api import user
|
|
from .api import history
|
|
from .api import ports
|
|
|
|
from BreCal.schemas import defs
|
|
from BreCal.brecal_utils.file_handling import get_project_root, ensure_path
|
|
from BreCal.brecal_utils.test_handling import execute_test_with_pytest, execute_coverage_test
|
|
from BreCal.brecal_utils.time_handling import difference_to_then
|
|
|
|
from BreCal.validators.time_logic import TimeLogic
|
|
from BreCal.validators.validation_rules import ValidationRules
|
|
from BreCal.validators.schema_validation import validation_state_and_validation_name
|
|
|
|
from BreCal.stubs.times_agency import get_times_agency
|
|
from BreCal.stubs.times_bsmd import get_times_bsmd
|
|
from BreCal.stubs.times_mooring import get_times_mooring
|
|
from BreCal.stubs.times_pilot import get_times_pilot
|
|
from BreCal.stubs.times_portauthority import get_times_port_authority
|
|
from BreCal.stubs.times_terminal import get_times_terminal
|
|
from BreCal.stubs.times_tug import get_times_tug
|
|
from BreCal.stubs.times_full import get_times_full_simple
|
|
from BreCal.stubs.df_times import get_df_times
|
|
|
|
from BreCal.services.schedule_routines import setup_schedule, run_schedule_permanently_in_background
|
|
|
|
def create_app(test_config=None, instance_path=None):
|
|
app = Flask(__name__, instance_relative_config=True)
|
|
app.config.from_mapping(
|
|
SECRET_KEY='dev'
|
|
)
|
|
if test_config is None:
|
|
app.config.from_pyfile('config.py', silent=True)
|
|
else:
|
|
app.config.from_mapping(test_config)
|
|
|
|
if instance_path is not None:
|
|
app.instance_path = instance_path
|
|
elif app.config.get("INSTANCE_PATH"):
|
|
app.instance_path = app.config["INSTANCE_PATH"]
|
|
|
|
try:
|
|
import os
|
|
print(f'Instance path = {app.instance_path}')
|
|
if not os.path.exists(app.instance_path):
|
|
os.makedirs(app.instance_path)
|
|
except OSError:
|
|
pass
|
|
|
|
# Add blueprints
|
|
app.register_blueprint(shipcalls.bp)
|
|
app.register_blueprint(participant.bp)
|
|
app.register_blueprint(times.bp)
|
|
app.register_blueprint(notifications.bp)
|
|
app.register_blueprint(berths.bp)
|
|
app.register_blueprint(ships.bp)
|
|
app.register_blueprint(login.bp)
|
|
app.register_blueprint(user.bp)
|
|
app.register_blueprint(history.bp)
|
|
app.register_blueprint(ports.bp)
|
|
|
|
log_level = getattr(logging, app.config.get("LOG_LEVEL", "DEBUG"))
|
|
log_format = "%(asctime)s [%(levelname)s] %(message)s"
|
|
root_logger = logging.getLogger()
|
|
for handler in list(root_logger.handlers):
|
|
root_logger.removeHandler(handler)
|
|
|
|
if app.config.get("LOG_TO_STDERR"):
|
|
handler = logging.StreamHandler(sys.stderr)
|
|
else:
|
|
log_file = app.config.get("LOG_FILE", "brecaltest.log")
|
|
if not os.path.isabs(log_file):
|
|
log_file = os.path.join(app.instance_path, log_file)
|
|
log_dir = os.path.dirname(log_file)
|
|
if log_dir and not os.path.exists(log_dir):
|
|
os.makedirs(log_dir, exist_ok=True)
|
|
handler = logging.FileHandler(log_file)
|
|
|
|
handler.setFormatter(logging.Formatter(log_format, datefmt="%d.%m.%Y %H:%M:%S"))
|
|
root_logger.addHandler(handler)
|
|
root_logger.setLevel(log_level)
|
|
|
|
if app.config.get("SECRET_KEY"):
|
|
os.environ["SECRET_KEY"] = app.config["SECRET_KEY"]
|
|
|
|
defs.SMTP_DEBUG_LEVEL = app.config.get("SMTP_DEBUG_LEVEL", defs.SMTP_DEBUG_LEVEL)
|
|
|
|
local_db.initPool(os.path.dirname(app.instance_path), config=app.config)
|
|
logging.info('App started')
|
|
|
|
# Setup Routine jobs (e.g., reevaluation of shipcalls)
|
|
setup_schedule(
|
|
update_shipcalls_interval_in_minutes=app.config.get("SCHEDULE_UPDATE_SHIPCALLS_MINUTES", 60),
|
|
notification_cooldown_mins=app.config.get("NOTIFICATION_COOLDOWN_MINS", defs.NOTIFICATION_COOLDOWN_MINS),
|
|
)
|
|
run_schedule_permanently_in_background(latency=app.config.get("SCHEDULE_BACKGROUND_LATENCY_SECONDS", 30))
|
|
logging.info('Routine Jobs are defined.')
|
|
|
|
return app
|
|
|
|
__all__ = [
|
|
"get_project_root",
|
|
"ensure_path",
|
|
"execute_test_with_pytest",
|
|
"execute_coverage_test",
|
|
"difference_to_then",
|
|
"TimeLogic",
|
|
"ValidationRules",
|
|
"validation_state_and_validation_name",
|
|
|
|
"get_times_agency",
|
|
"get_times_bsmd",
|
|
"get_times_mooring",
|
|
"get_times_pilot",
|
|
"get_times_port_authority",
|
|
"get_times_terminal",
|
|
"get_times_tug",
|
|
"get_times_full_simple",
|
|
"get_df_times",
|
|
]
|
|
|