Gan, Jimmy
6c0d5a4e63
- Flask-based mold cost calculation application - PostgreSQL database with SQLAlchemy ORM - Docker/Podman containerization - Comprehensive documentation in docs/ - Clean, organized codebase without obsolete files - Production-ready deployment configuration - Enhanced pytest configuration with coverage reporting - Master/Development branch structure
124 lines
4.2 KiB
Python
124 lines
4.2 KiB
Python
from flask import Flask
|
|
import os
|
|
import logging
|
|
from datetime import datetime
|
|
from sqlalchemy.exc import SQLAlchemyError, OperationalError
|
|
from sqlalchemy import inspect
|
|
from time import sleep
|
|
from .config import get_config
|
|
from .logging_config import configure_logging
|
|
from dotenv import load_dotenv
|
|
from .extensions import db, login_manager, migrate, csrf, limiter, cache, mail
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from flask_login import LoginManager
|
|
from flask_migrate import Migrate
|
|
from flask_limiter import Limiter
|
|
from flask_limiter.util import get_remote_address
|
|
from flask_caching import Cache
|
|
|
|
# Load environment variables first
|
|
load_dotenv()
|
|
|
|
db = SQLAlchemy()
|
|
migrate = Migrate()
|
|
limiter = Limiter(key_func=get_remote_address)
|
|
cache = Cache()
|
|
|
|
def create_app(config_class=None):
|
|
app = Flask(__name__, static_folder='static')
|
|
|
|
@app.context_processor
|
|
def inject_now():
|
|
return {'now': datetime.utcnow()}
|
|
|
|
# Load configuration
|
|
if config_class is None:
|
|
config_class = get_config()
|
|
app.config.from_object(config_class)
|
|
|
|
# Initialize configuration
|
|
config_class.init_app(app)
|
|
|
|
# Ensure secret key is properly set
|
|
if not app.config.get('SECRET_KEY'):
|
|
raise ValueError("SECRET_KEY must be set in the configuration")
|
|
app.config['SECRET_KEY'] = app.config['SECRET_KEY'].encode('utf-8')
|
|
|
|
# Ensure instance folder exists
|
|
try:
|
|
os.makedirs(app.instance_path)
|
|
except OSError:
|
|
pass
|
|
|
|
# Configure logging
|
|
configure_logging(app)
|
|
|
|
# Initialize extensions with retry logic
|
|
max_retries = 3
|
|
retry_delay = 5
|
|
|
|
# Initialize extensions only if they haven't been initialized
|
|
if not hasattr(app, 'extensions') or 'sqlalchemy' not in app.extensions:
|
|
for attempt in range(max_retries):
|
|
try:
|
|
# Initialize SQLAlchemy only if not already initialized
|
|
if 'sqlalchemy' not in app.extensions:
|
|
db.init_app(app)
|
|
migrate.init_app(app, db)
|
|
|
|
# Initialize other extensions
|
|
if 'csrf' not in app.extensions:
|
|
csrf.init_app(app)
|
|
if 'limiter' not in app.extensions:
|
|
limiter.init_app(app)
|
|
if 'cache' not in app.extensions:
|
|
cache.init_app(app)
|
|
if 'mail' not in app.extensions:
|
|
mail.init_app(app)
|
|
if 'login_manager' not in app.extensions:
|
|
login_manager.init_app(app)
|
|
|
|
# Test database connection
|
|
with app.app_context():
|
|
db.engine.connect()
|
|
break
|
|
except OperationalError as e:
|
|
if attempt == max_retries - 1:
|
|
app.logger.error(f"Failed to connect to database after {max_retries} attempts: {str(e)}")
|
|
raise
|
|
app.logger.warning(f"Database connection attempt {attempt + 1} failed: {str(e)}")
|
|
sleep(retry_delay)
|
|
|
|
# Configure login manager
|
|
login_manager.login_view = 'auth.login'
|
|
login_manager.login_message = 'Please log in to access this page.'
|
|
login_manager.login_message_category = 'warning'
|
|
|
|
@login_manager.user_loader
|
|
def load_user(user_id):
|
|
from .models import User
|
|
return User.query.get(int(user_id))
|
|
|
|
# Register blueprints
|
|
from .routes import main, auth, admin
|
|
app.register_blueprint(main.bp)
|
|
app.register_blueprint(auth.bp)
|
|
app.register_blueprint(admin.bp)
|
|
|
|
# Register error handlers
|
|
from .errors import register_error_handlers
|
|
register_error_handlers(app)
|
|
|
|
# Debug: log all registered endpoints at startup (warning level and flush)
|
|
app.logger.warning("Registered endpoints (debug):")
|
|
for rule in app.url_map.iter_rules():
|
|
app.logger.warning(f"{rule.endpoint} {rule.methods} {rule.rule}")
|
|
for handler in app.logger.handlers:
|
|
handler.flush()
|
|
|
|
# Debug: write all registered endpoints to a file
|
|
with open('logs/endpoints.log', 'w') as f:
|
|
for rule in app.url_map.iter_rules():
|
|
f.write(f"{rule.endpoint} {rule.methods} {rule.rule}\n")
|
|
|
|
return app |