sneakyklaus/src/app.py

"""Flask application factory for Sneaky Klaus.

This module provides the application factory function that creates
and configures the Flask application instance.
"""

from pathlib import Path

from flask import Flask
from flask_bcrypt import Bcrypt
from flask_session import Session
from flask_sqlalchemy import SQLAlchemy
from flask_wtf.csrf import CSRFProtect

# Initialize Flask extensions (without app instance)
db = SQLAlchemy()
bcrypt = Bcrypt()
csrf = CSRFProtect()
session = Session()


def create_app(config_name: str | None = None) -> Flask:
    """Create and configure the Flask application.

    Args:
        config_name: Configuration environment name (development, production, testing).
                    If None, uses FLASK_ENV environment variable.

    Returns:
        Configured Flask application instance.
    """
    # Create Flask app instance
    app = Flask(__name__)

    # Load configuration
    from src.config import ProductionConfig, get_config

    config_class = get_config(config_name)
    app.config.from_object(config_class)

    # Validate production config if needed
    if config_name == "production":
        ProductionConfig.validate()

    # Ensure data directory exists
    data_dir = Path(app.config["DATA_DIR"])
    data_dir.mkdir(parents=True, exist_ok=True)

    # Initialize extensions
    db.init_app(app)
    bcrypt.init_app(app)
    csrf.init_app(app)

    # Initialize session with filesystem backend
    # Ensure session directory exists
    session_dir = Path(app.config.get("SESSION_FILE_DIR", data_dir / "sessions"))
    session_dir.mkdir(parents=True, exist_ok=True)
    session.init_app(app)

    # Register blueprints
    from src.routes.admin import admin_bp
    from src.routes.participant import participant_bp
    from src.routes.setup import setup_bp

    app.register_blueprint(setup_bp)
    app.register_blueprint(admin_bp)
    app.register_blueprint(participant_bp)

    # Register error handlers
    register_error_handlers(app)

    # First-run setup check
    register_setup_check(app)

    # Import models to ensure they're registered with SQLAlchemy
    with app.app_context():
        from src.models import (  # noqa: F401
            Admin,
            Exchange,
            MagicToken,
            Participant,
            RateLimit,
        )
        # Schema managed by Alembic migrations (applied via entrypoint script)

    return app


def register_error_handlers(app: Flask) -> None:
    """Register custom error handlers.

    Args:
        app: Flask application instance.
    """
    from flask import render_template

    @app.errorhandler(404)
    def not_found_error(_error):
        """Handle 404 Not Found errors."""
        return render_template("errors/404.html"), 404

    @app.errorhandler(403)
    def forbidden_error(_error):
        """Handle 403 Forbidden errors."""
        return render_template("errors/403.html"), 403

    @app.errorhandler(500)
    def internal_error(error):
        """Handle 500 Internal Server Error."""
        db.session.rollback()  # Rollback any failed transactions
        app.logger.error(f"Internal server error: {error}")
        return render_template("errors/500.html"), 500

    @app.errorhandler(429)
    def rate_limit_error(_error):
        """Handle 429 Too Many Requests errors."""
        return render_template("errors/429.html"), 429


def register_setup_check(app: Flask) -> None:
    """Register before_request handler for first-run setup detection.

    Args:
        app: Flask application instance.
    """
    from flask import redirect, request, url_for

    @app.before_request
    def check_setup_required():
        """Redirect to setup page if no admin exists.

        This runs before every request to ensure the application
        has been set up with an admin account.
        """
        # Skip check for certain endpoints
        # Participant routes are public and don't require setup
        if request.endpoint in [
            "setup.setup",
            "static",
            "health",
            "participant.register",
            "participant.register_success",
            "participant.request_access",
            "participant.magic_login",
            "participant.dashboard",
            "participant.logout",
            "participant.profile_edit",
            "participant.update_preferences",
            "participant.withdraw",
        ]:
            return

        # Check if admin exists (always check in testing mode)
        from src.models.admin import Admin

        admin_count = db.session.query(Admin).count()
        requires_setup = admin_count == 0

        # Redirect to setup if needed
        if requires_setup and request.endpoint != "setup.setup":
            return redirect(url_for("setup.setup"))