backupmanager_ynh/sources/app.py

import json
import logging
import os
from datetime import datetime

from flask import Flask
from werkzeug.middleware.proxy_fix import ProxyFix

# --- Configuration -----------------------------------------------------------

_config_path = os.environ.get(
    "BACKUPMANAGER_CONFIG",
    os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.py"),
)

app = Flask(__name__)
app.config.from_pyfile(_config_path)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///" + app.config["DB_PATH"]
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)
app.jinja_env.filters["fromjson"] = json.loads

os.makedirs(os.path.dirname(app.config["LOG_PATH"]), exist_ok=True)
logging.basicConfig(
    filename=app.config["LOG_PATH"],
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

# --- Extensions --------------------------------------------------------------

from db import db, Job

db.init_app(app)

from scheduler import init_scheduler, schedule_job

# --- Blueprints --------------------------------------------------------------

from blueprints.jobs import bp as bp_jobs
from blueprints.destinations import bp as bp_dest
from blueprints.network import bp as bp_network
from blueprints.settings import bp as bp_cfg
from blueprints.api import bp as bp_api

app.register_blueprint(bp_jobs)
app.register_blueprint(bp_dest)
app.register_blueprint(bp_network)
app.register_blueprint(bp_cfg)
app.register_blueprint(bp_api)

# --- Context processor -------------------------------------------------------

@app.context_processor
def _inject_globals():
    return {
        "instance_name": app.config.get("INSTANCE_NAME", ""),
        "instance_url": app.config.get("INSTANCE_URL", ""),
        "api_token": app.config.get("API_TOKEN", ""),
        "now": datetime.utcnow(),
    }

# --- Démarrage ---------------------------------------------------------------

with app.app_context():
    from sqlalchemy import inspect, text
    _insp = inspect(db.engine)
    if "jobs" in _insp.get_table_names():
        _cols = [c["name"] for c in _insp.get_columns("jobs")]
        if "retention_gfs_config" not in _cols:
            with db.engine.connect() as _conn:
                _conn.execute(text("ALTER TABLE jobs ADD COLUMN retention_gfs_config TEXT"))
                _conn.commit()
    db.create_all()
    init_scheduler(app)
    for _job in Job.query.filter_by(enabled=True).all():
        schedule_job(_job)