Daniel Schick
b5dd7422f4
Release pool connection handle und all circumstances especially also when a query fails before the call is finished. This should avoid connection starvation. fix prod. link production fix Fixed application path
104 lines
3.2 KiB
Python
104 lines
3.2 KiB
Python
from flask import Flask
|
|
|
|
import os
|
|
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.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
|
|
|
|
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)
|
|
|
|
logging.basicConfig(filename='brecal.log', level=logging.WARNING, format='%(asctime)s | %(name)s | %(levelname)s | %(message)s')
|
|
local_db.initPool(os.path.dirname(app.instance_path))
|
|
logging.info('App started')
|
|
|
|
# Setup Routine jobs (e.g., reevaluation of shipcalls)
|
|
setup_schedule(update_shipcalls_interval_in_minutes=60)
|
|
run_schedule_permanently_in_background(latency=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",
|
|
]
|
|
|