Loading clams/__init__.py 0 → 100644 +104 −0 Original line number Diff line number Diff line import json import secrets import base64 from pathlib import Path from flask import Flask from flask_apscheduler import APScheduler from flask_sqlalchemy import SQLAlchemy from .db_config import get_database_uri, get_uri_for_sqlite db = SQLAlchemy() scheduler = APScheduler() from .model import * def create_app(test_db_path: Path = None) -> Flask: """ Create the Flask application. This will also attempt to connect to a database, either a remote database or a local SQLite database. When a test database path is provided, the application will automatically switch into testing mode. This means: - The scheduler will not be initialized. No jobs will be run. - The test database will be used, no matter what deviating configuration may be specified. :param test_db_path: A path to the database used for unit testing purposes :return: the Flask application object """ app = Flask(__name__) app.logger.info("App instance was successfully created.") instance_path = Path(app.instance_path) instance_path.mkdir(exist_ok=True) sqlalchemy_database_uri = ( get_uri_for_sqlite(test_db_path) if test_db_path else get_database_uri(instance_path) ) _connect_db(app, sqlalchemy_database_uri) _load_flask_config(app) _register_blueprints(app) if test_db_path is None: # Only start scheduler when we are not in testing mode _init_scheduler(app) return app def _load_flask_config(app: Flask) -> None: """ Load the flask specific configuration from the database. This will also ensure a SECRET_KEY is set, creating one randomly if necessary. """ with app.app_context(): for entry in FlaskConfigEntry.query.all(): entry.apply(app) if not app.config["SECRET_KEY"]: secret_key_entry = FlaskConfigEntry( key="SECRET_KEY", value=secrets.token_hex() ) secret_key_entry.apply(app) db.session.add(secret_key_entry) db.session.commit() if not app.config["SECRET_KEY"]: raise Exception( "app.config['SECRET_KEY'] should have been set, but wasn't. Aborting!" ) def _init_scheduler(app: Flask) -> None: """ Initialize the scheduler (which will add all jobs defined in jobs.py) """ from . import jobs scheduler.init_app(app) scheduler.start() def _register_blueprints(app: Flask) -> None: """ Register all blueprints """ from . import main, api app.register_blueprint(main.bp) app.register_blueprint(api.bp) def _connect_db(app: Flask, sqlalchemy_database_uri: str) -> None: """ Connect to the database using the given URI and create all database tables if they don't yet exist. """ app.config["SQLALCHEMY_DATABASE_URI"] = sqlalchemy_database_uri db.init_app(app) Loading
clams/__init__.py 0 → 100644 +104 −0 Original line number Diff line number Diff line import json import secrets import base64 from pathlib import Path from flask import Flask from flask_apscheduler import APScheduler from flask_sqlalchemy import SQLAlchemy from .db_config import get_database_uri, get_uri_for_sqlite db = SQLAlchemy() scheduler = APScheduler() from .model import * def create_app(test_db_path: Path = None) -> Flask: """ Create the Flask application. This will also attempt to connect to a database, either a remote database or a local SQLite database. When a test database path is provided, the application will automatically switch into testing mode. This means: - The scheduler will not be initialized. No jobs will be run. - The test database will be used, no matter what deviating configuration may be specified. :param test_db_path: A path to the database used for unit testing purposes :return: the Flask application object """ app = Flask(__name__) app.logger.info("App instance was successfully created.") instance_path = Path(app.instance_path) instance_path.mkdir(exist_ok=True) sqlalchemy_database_uri = ( get_uri_for_sqlite(test_db_path) if test_db_path else get_database_uri(instance_path) ) _connect_db(app, sqlalchemy_database_uri) _load_flask_config(app) _register_blueprints(app) if test_db_path is None: # Only start scheduler when we are not in testing mode _init_scheduler(app) return app def _load_flask_config(app: Flask) -> None: """ Load the flask specific configuration from the database. This will also ensure a SECRET_KEY is set, creating one randomly if necessary. """ with app.app_context(): for entry in FlaskConfigEntry.query.all(): entry.apply(app) if not app.config["SECRET_KEY"]: secret_key_entry = FlaskConfigEntry( key="SECRET_KEY", value=secrets.token_hex() ) secret_key_entry.apply(app) db.session.add(secret_key_entry) db.session.commit() if not app.config["SECRET_KEY"]: raise Exception( "app.config['SECRET_KEY'] should have been set, but wasn't. Aborting!" ) def _init_scheduler(app: Flask) -> None: """ Initialize the scheduler (which will add all jobs defined in jobs.py) """ from . import jobs scheduler.init_app(app) scheduler.start() def _register_blueprints(app: Flask) -> None: """ Register all blueprints """ from . import main, api app.register_blueprint(main.bp) app.register_blueprint(api.bp) def _connect_db(app: Flask, sqlalchemy_database_uri: str) -> None: """ Connect to the database using the given URI and create all database tables if they don't yet exist. """ app.config["SQLALCHEMY_DATABASE_URI"] = sqlalchemy_database_uri db.init_app(app)