240 lines
7.3 KiB
Python
240 lines
7.3 KiB
Python
# ==============================================================================
|
|
# Formeleon - Multi-tenant Form Generator
|
|
# Author: Björn Runden <br@runden.net>
|
|
# Module: app.py
|
|
# Description: Main Flask application initialization and Blueprint registration.
|
|
# Sets up database, configuration, and registers all route blueprints.
|
|
# ==============================================================================
|
|
|
|
"""
|
|
Main Flask application module for Formeleon.
|
|
|
|
This module:
|
|
- Loads application configuration
|
|
- Initializes Flask extensions (SQLAlchemy, Migrate, CORS, Session)
|
|
- Registers all route blueprints (API, Admin, Auth)
|
|
- Sets up error handlers
|
|
- Provides utility functions for request context
|
|
|
|
The application uses Flask Blueprints for modular route organization:
|
|
- /api/v1/* - Public Form API for form submissions
|
|
- /admin/* - Admin Dashboard for form management
|
|
- /auth/* - Authentication routes (OIDC callbacks, logout)
|
|
"""
|
|
|
|
import os
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from flask import Flask, request, session, g
|
|
from flask_cors import CORS
|
|
from flask_session import Session
|
|
from flask_migrate import Migrate
|
|
import uuid
|
|
|
|
# corrected for py3.13 from config import Config, config_map
|
|
from config import Config, config_map
|
|
from models import db, Tenant, User, UserRole, Submission, FormConfig, AuditLog, SubmissionStatus
|
|
from utils.cli import register_cli_commands
|
|
from utils.middleware import setup_dev_mode_middleware, get_current_user, get_current_tenant
|
|
from utils.helpers import get_request_context
|
|
|
|
from werkzeug.middleware.proxy_fix import ProxyFix
|
|
|
|
# Initialize extensions globally (will be bound to app in create_app)
|
|
migrate = Migrate()
|
|
cors = CORS()
|
|
|
|
|
|
def get_config_class() -> type:
|
|
"""
|
|
Get the appropriate configuration class based on FLASK_ENV.
|
|
|
|
:return: Configuration class to use
|
|
"""
|
|
env = os.environ.get('FLASK_ENV', 'development')
|
|
return config_map.get(env, Config)
|
|
|
|
|
|
def create_app(config_class: Optional[type] = None) -> Flask:
|
|
"""
|
|
Create and configure the Flask application instance.
|
|
|
|
:param config_class: Configuration class to use (defaults to environment-based)
|
|
:return: Configured Flask application instance
|
|
"""
|
|
# Initialize Flask app
|
|
app = Flask(__name__)
|
|
app.config.from_object(config_class or get_config_class())
|
|
|
|
|
|
# Vertraue 1 Reverse Proxy (NPM) für X-Forwarded-For, Proto, Host, Port
|
|
app.wsgi_app = ProxyFix(
|
|
app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_port=1, x_prefix=1
|
|
)
|
|
|
|
# Set up logging
|
|
setup_logging(app)
|
|
|
|
# Initialize extensions with app
|
|
initialize_extensions(app)
|
|
|
|
# Register blueprints
|
|
register_blueprints(app)
|
|
|
|
# Register error handlers
|
|
register_error_handlers(app)
|
|
|
|
# Register CLI commands
|
|
register_cli_commands(app)
|
|
|
|
# Set up dev mode middleware if enabled
|
|
if app.config['DEV_MODE']:
|
|
setup_dev_mode_middleware(app)
|
|
|
|
return app
|
|
|
|
|
|
def setup_logging(app: Flask) -> None:
|
|
"""
|
|
Configure logging for the application.
|
|
|
|
:param app: Flask application instance
|
|
"""
|
|
log_level = logging.DEBUG if app.config['DEBUG'] else logging.INFO
|
|
logging.basicConfig(
|
|
level=log_level,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
app.logger.setLevel(log_level)
|
|
|
|
# Suppress Flask-SQLAlchemy logs unless in debug mode
|
|
if not app.config['DEBUG']:
|
|
logging.getLogger('flask_sqlalchemy').setLevel(logging.WARNING)
|
|
|
|
|
|
def initialize_extensions(app: Flask) -> None:
|
|
"""
|
|
Initialize Flask extensions with the application.
|
|
|
|
:param app: Flask application instance
|
|
"""
|
|
# Database
|
|
db.init_app(app)
|
|
|
|
# Migration
|
|
migrate.init_app(app, db)
|
|
|
|
# CORS
|
|
if app.config['ENABLE_CORS']:
|
|
cors.init_app(app, resources={
|
|
r'/api/*': {
|
|
'origins': app.config.get('ALLOWED_ORIGINS', '*'),
|
|
'methods': ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS'],
|
|
'allow_headers': ['Content-Type', 'Authorization', 'X-Request-ID']
|
|
}
|
|
})
|
|
|
|
# Session
|
|
app.config['SESSION_TYPE'] = 'sqlalchemy'
|
|
app.config['SESSION_SQLALCHEMY'] = db
|
|
Session(app)
|
|
|
|
# Ensure secret key is set
|
|
if not app.config.get('SECRET_KEY'):
|
|
app.config['SECRET_KEY'] = str(uuid.uuid4())
|
|
app.logger.warning('Generated random SECRET_KEY - set explicitly in production!')
|
|
|
|
|
|
def register_blueprints(app: Flask) -> None:
|
|
"""
|
|
Register all route blueprints with the application.
|
|
|
|
:param app: Flask application instance
|
|
"""
|
|
# Import blueprints here to avoid circular imports
|
|
from routes.api import api_bp
|
|
from routes.admin import admin_bp
|
|
from routes.auth import auth_bp
|
|
|
|
# Register blueprints with URL prefixes
|
|
app.register_blueprint(api_bp, url_prefix='/api/v1')
|
|
app.register_blueprint(admin_bp, url_prefix='/admin')
|
|
app.register_blueprint(auth_bp, url_prefix='/auth')
|
|
|
|
# Register root route for health check
|
|
@app.route('/')
|
|
def health_check():
|
|
"""Health check endpoint."""
|
|
return {
|
|
'status': 'ok',
|
|
'version': '1.0.0',
|
|
'dev_mode': app.config['DEV_MODE']
|
|
}, 200
|
|
|
|
|
|
def register_error_handlers(app: Flask) -> None:
|
|
"""
|
|
Register custom error handlers for the application.
|
|
|
|
:param app: Flask application instance
|
|
"""
|
|
@app.errorhandler(400)
|
|
def bad_request_error(error):
|
|
"""Handle 400 Bad Request errors."""
|
|
return {
|
|
'error': 'Bad Request',
|
|
'message': str(error.description) if error.description else 'Invalid request data'
|
|
}, 400
|
|
|
|
@app.errorhandler(401)
|
|
def unauthorized_error(error):
|
|
"""Handle 401 Unauthorized errors."""
|
|
return {
|
|
'error': 'Unauthorized',
|
|
'message': str(error.description) if error.description else 'Authentication required'
|
|
}, 401
|
|
|
|
@app.errorhandler(403)
|
|
def forbidden_error(error):
|
|
"""Handle 403 Forbidden errors."""
|
|
return {
|
|
'error': 'Forbidden',
|
|
'message': str(error.description) if error.description else 'Access denied'
|
|
}, 403
|
|
|
|
@app.errorhandler(404)
|
|
def not_found_error(error):
|
|
"""Handle 404 Not Found errors."""
|
|
return {
|
|
'error': 'Not Found',
|
|
'message': str(error.description) if error.description else 'Resource not found'
|
|
}, 404
|
|
|
|
@app.errorhandler(429)
|
|
def too_many_requests_error(error):
|
|
"""Handle 429 Too Many Requests errors."""
|
|
return {
|
|
'error': 'Too Many Requests',
|
|
'message': str(error.description) if error.description else 'Rate limit exceeded'
|
|
}, 429
|
|
|
|
@app.errorhandler(500)
|
|
def internal_server_error(error):
|
|
"""Handle 500 Internal Server Error."""
|
|
app.logger.error(f'Internal Server Error: {error}')
|
|
return {
|
|
'error': 'Internal Server Error',
|
|
'message': 'An unexpected error occurred'
|
|
}, 500
|
|
|
|
|
|
# Create application instance
|
|
app = create_app()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# This allows running the app directly with python app.py
|
|
# In production, use gunicorn or other WSGI servers
|
|
app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 5000)), debug=app.config['DEBUG'])
|