Flask应用工厂函数

什么是应用工厂函数

应用工厂函数是Flask中一种重要的设计模式，它允许你创建一个函数来构建和配置Flask应用实例。这种模式提供了更好的代码组织、配置管理和测试支持。

为什么使用工厂函数

传统方式的局限性

# 传统方式 - 直接创建app
from flask import Flask

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///db.sqlite'

# 注册蓝图
from myapp.views import main
app.register_blueprint(main)

# 初始化扩展
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy(app)

if __name__ == '__main__':
    app.run()

问题：

• 应用对象在模块级别创建，难以配置不同环境
• 扩展在导入时初始化，可能导致循环导入
• 测试时需要创建新的应用实例

工厂函数的优势

# 工厂函数方式
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()  # 先创建扩展实例，但不绑定app

def create_app(config_name='default'):
    """创建Flask应用实例的工厂函数"""
    app = Flask(__name__)
    
    # 加载配置
    if config_name == 'development':
        app.config.from_object('config.DevelopmentConfig')
    elif config_name == 'testing':
        app.config.from_object('config.TestingConfig')
    elif config_name == 'production':
        app.config.from_object('config.ProductionConfig')
    else:
        app.config.from_object('config.DefaultConfig')
    
    # 初始化扩展（延迟绑定）
    db.init_app(app)
    
    # 注册蓝图
    from .views import main
    app.register_blueprint(main)
    
    # 注册错误处理器
    from .errors import not_found_error, internal_error
    app.register_error_handler(404, not_found_error)
    app.register_error_handler(500, internal_error)
    
    return app

优势：

• 支持多环境配置
• 避免循环导入
• 便于测试
• 支持多个应用实例

基本工厂函数实现

最简单的工厂函数

# app/__init__.py
from flask import Flask

def create_app():
    app = Flask(__name__)
    
    # 基本配置
    app.config['SECRET_KEY'] = 'dev-secret-key'
    app.config['DEBUG'] = True
    
    # 简单的路由
    @app.route('/')
    def hello():
        return 'Hello, World!'
    
    return app

使用工厂函数

# run.py
from app import create_app

app = create_app()

if __name__ == '__main__':
    app.run(debug=True)

配置管理

配置类

# config.py
import os

class Config:
    """基础配置类"""
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard-to-guess-string'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    
    @staticmethod
    def init_app(app):
        """初始化应用"""
        pass

class DevelopmentConfig(Config):
    """开发环境配置"""
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
        'sqlite:///dev.db'

class TestingConfig(Config):
    """测试环境配置"""
    TESTING = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or \
        'sqlite:///test.db'
    WTF_CSRF_ENABLED = False  # 测试时禁用CSRF保护

class ProductionConfig(Config):
    """生产环境配置"""
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'sqlite:///prod.db'
    
    @classmethod
    def init_app(cls, app):
        Config.init_app(app)
        
        # 生产环境日志配置
        import logging
        from logging.handlers import RotatingFileHandler
        
        file_handler = RotatingFileHandler(
            'flask.log', maxBytes=10240, backupCount=10
        )
        file_handler.setFormatter(logging.Formatter(
            '%(asctime)s %(levelname)s: %(message)s '
            '[in %(pathname)s:%(lineno)d]'
        ))
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)
        app.logger.setLevel(logging.INFO)
        app.logger.info('Flask application startup')

config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

在工厂函数中使用配置类

def create_app(config_name='default'):
    app = Flask(__name__)
    
    # 加载配置
    app.config.from_object(config[config_name])
    
    # 调用配置类的初始化方法
    config[config_name].init_app(app)
    
    return app

扩展初始化

延迟初始化模式

# extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_mail import Mail
from flask_bootstrap import Bootstrap

db = SQLAlchemy()
migrate = Migrate()
login = LoginManager()
mail = Mail()
bootstrap = Bootstrap()

# 工厂函数中初始化
def create_app(config_name='default'):
    app = Flask(__name__)
    
    # 加载配置
    app.config.from_object(config[config_name])
    
    # 初始化扩展
    db.init_app(app)
    migrate.init_app(app, db)
    login.init_app(app)
    mail.init_app(app)
    bootstrap.init_app(app)
    
    # 配置登录管理器
    login.login_view = 'auth.login'
    login.login_message = '请先登录以访问此页面'
    
    return app

蓝图注册

组织项目结构

myapp/
├── __init__.py      # 工厂函数
├── config.py        # 配置
├── extensions.py    # 扩展
├── models.py        # 数据模型
├── auth/
│   ├── __init__.py
│   ├── forms.py
│   └── views.py
├── main/
│   ├── __init__.py
│   └── views.py
└── api/
    ├── __init__.py
    └── v1/
        ├── __init__.py
        └── views.py

蓝图示例

# auth/__init__.py
from flask import Blueprint

auth = Blueprint('auth', __name__)

from . import views

# auth/views.py
from flask import render_template, redirect, url_for, flash, request
from . import auth
from .forms import LoginForm, RegistrationForm

@auth.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()
    if form.validate_on_submit():
        # 处理登录逻辑
        flash('登录成功！', 'success')
        return redirect(url_for('main.index'))
    return render_template('auth/login.html', form=form)

@auth.route('/register', methods=['GET', 'POST'])
def register():
    form = RegistrationForm()
    if form.validate_on_submit():
        # 处理注册逻辑
        flash('注册成功！请登录。', 'success')
        return redirect(url_for('auth.login'))
    return render_template('auth/register.html', form=form)

在工厂函数中注册蓝图

def create_app(config_name='default'):
    app = Flask(__name__)
    
    # 加载配置
    app.config.from_object(config[config_name])
    
    # 初始化扩展
    from .extensions import db, login, mail
    db.init_app(app)
    login.init_app(app)
    mail.init_app(app)
    
    # 注册蓝图
    from .auth import auth as auth_blueprint
    app.register_blueprint(auth_blueprint, url_prefix='/auth')
    
    from .main import main as main_blueprint
    app.register_blueprint(main_blueprint)
    
    from .api.v1 import api as api_blueprint
    app.register_blueprint(api_blueprint, url_prefix='/api/v1')
    
    return app

上下文处理器

添加全局变量

def create_app(config_name='default'):
    app = Flask(__name__)
    
    # ... 其他初始化代码 ...
    
    @app.context_processor
    def inject_global_vars():
        """注入全局变量到所有模板"""
        from .models import Category
        
        return {
            'site_name': '我的Flask应用',
            'current_year': datetime.now().year,
            'categories': Category.query.all()
        }
    
    @app.context_processor
    def utility_processor():
        """注入工具函数到所有模板"""
        def format_price(amount):
            return f"¥{amount:,.2f}"
        
        return dict(format_price=format_price)
    
    return app

命令行接口

使用Flask CLI

def create_app(config_name='default'):
    app = Flask(__name__)
    
    # ... 其他初始化代码 ...
    
    # 注册CLI命令
    @app.cli.command('init-db')
    def init_db_command():
        """初始化数据库"""
        from .extensions import db
        db.create_all()
        print('数据库已初始化。')
    
    @app.cli.command('create-admin')
    @click.argument('username')
    @click.argument('email')
    @click.password_option()
    def create_admin_command(username, email, password):
        """创建管理员用户"""
        from .models import User
        from .extensions import db
        
        admin = User(
            username=username,
            email=email,
            is_admin=True
        )
        admin.set_password(password)
        
        db.session.add(admin)
        db.session.commit()
        print(f'管理员用户 {username} 已创建。')
    
    return app

测试支持

测试工厂函数

# tests/conftest.py
import pytest
from myapp import create_app, db as _db

@pytest.fixture(scope='session')
def app():
    """创建测试应用"""
    app = create_app('testing')
    return app

@pytest.fixture(scope='session')
def db(app):
    """创建测试数据库"""
    with app.app_context():
        _db.create_all()
        yield _db
        _db.drop_all()

@pytest.fixture(scope='function')
def client(app):
    """测试客户端"""
    return app.test_client()

@pytest.fixture(scope='function')
def session(db):
    """数据库会话"""
    connection = db.engine.connect()
    transaction = connection.begin()
    
    options = dict(bind=connection, binds={})
    session = db.create_scoped_session(options=options)
    
    db.session = session
    
    yield session
    
    transaction.rollback()
    connection.close()
    session.remove()

测试示例

# tests/test_auth.py
def test_login(client):
    """测试登录功能"""
    response = client.post('/auth/login', data={
        'username': 'testuser',
        'password': 'testpass'
    })
    assert response.status_code == 200
    assert b'登录成功' in response.data

def test_registration(client, session):
    """测试注册功能"""
    response = client.post('/auth/register', data={
        'username': 'newuser',
        'email': 'new@example.com',
        'password': 'newpass',
        'confirm_password': 'newpass'
    })
    assert response.status_code == 302  # 重定向
    assert response.location.endswith('/auth/login')

实际项目示例

完整的工厂函数

# myapp/__init__.py
import os
from flask import Flask
from flask.logging import default_handler
import logging
from logging.handlers import RotatingFileHandler

def create_app(config_name=None):
    """应用工厂函数"""
    if config_name is None:
        config_name = os.getenv('FLASK_CONFIG', 'development')
    
    app = Flask(__name__)
    
    # 加载配置
    from .config import config
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)
    
    # 配置日志
    if not app.debug and not app.testing:
        if not os.path.exists('logs'):
            os.mkdir('logs')
        
        file_handler = RotatingFileHandler(
            'logs/myapp.log',
            maxBytes=10240,
            backupCount=10
        )
        file_handler.setFormatter(logging.Formatter(
            '%(asctime)s %(levelname)s: %(message)s '
            '[in %(pathname)s:%(lineno)d]'
        ))
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)
        app.logger.removeHandler(default_handler)
        app.logger.setLevel(logging.INFO)
        app.logger.info('MyApp startup')
    
    # 初始化扩展
    from .extensions import db, migrate, login, csrf, mail
    db.init_app(app)
    migrate.init_app(app, db)
    login.init_app(app)
    csrf.init_app(app)
    mail.init_app(app)
    
    # 注册蓝图
    from .blueprints.auth import auth_bp
    from .blueprints.main import main_bp
    from .blueprints.api import api_bp
    from .blueprints.admin import admin_bp
    
    app.register_blueprint(auth_bp, url_prefix='/auth')
    app.register_blueprint(main_bp)
    app.register_blueprint(api_bp, url_prefix='/api/v1')
    app.register_blueprint(admin_bp, url_prefix='/admin')
    
    # 注册错误处理器
    from .errors import register_error_handlers
    register_error_handlers(app)
    
    # 注册上下文处理器
    from .context import register_context_processors
    register_context_processors(app)
    
    # 注册CLI命令
    from .commands import register_commands
    register_commands(app)
    
    # 注册shell上下文
    @app.shell_context_processor
    def make_shell_context():
        from .models import User, Post, Comment
        return {
            'db': db,
            'User': User,
            'Post': Post,
            'Comment': Comment
        }
    
    return app

最佳实践

1. 环境变量配置

# .env 文件
FLASK_APP=myapp
FLASK_ENV=development
SECRET_KEY=your-secret-key
DATABASE_URL=postgresql://user:pass@localhost/dbname

2. 使用python-dotenv

from dotenv import load_dotenv
load_dotenv()  # 加载.env文件

3. 避免循环导入

• 在工厂函数内部导入蓝图
• 使用延迟初始化扩展
• 将模型定义放在单独的文件中

4. 配置管理

• 使用配置类
• 支持多环境
• 敏感信息使用环境变量

5. 错误处理

• 统一错误页面
• 日志记录
• 生产环境友好错误信息

应用部署

使用工厂函数模式后，应用的部署变得更加灵活。以下是几种常见的部署方式：

1. 开发服务器部署

# run.py
from myapp import create_app

app = create_app('development')

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

2. Gunicorn部署

Gunicorn是一个Python WSGI HTTP服务器，适合生产环境部署。

安装Gunicorn

pip install gunicorn

基本使用

# 使用工厂函数
gunicorn "myapp:create_app()" -w 4 -b 0.0.0.0:8000

# 指定配置文件
gunicorn "myapp:create_app('production')" -w 4 -b 0.0.0.0:8000

Gunicorn配置文件

# gunicorn_config.py
import multiprocessing

# 服务器配置
bind = "0.0.0.0:8000"
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "sync"
threads = 2

# 日志配置
accesslog = "/var/log/gunicorn/access.log"
errorlog = "/var/log/gunicorn/error.log"
loglevel = "info"

# 进程管理
max_requests = 1000
max_requests_jitter = 50
timeout = 120
keepalive = 5

# 应用工厂函数
def app(environ, start_response):
    """WSGI应用入口"""
    from myapp import create_app
    flask_app = create_app('production')
    return flask_app(environ, start_response)

使用配置文件启动

gunicorn -c gunicorn_config.py "myapp:create_app('production')"

系统服务配置（Systemd）

# /etc/systemd/system/myapp.service
[Unit]
Description=My Flask Application
After=network.target

[Service]
User=www-data
Group=www-data
WorkingDirectory=/path/to/myapp
Environment="PATH=/path/to/venv/bin"
Environment="FLASK_CONFIG=production"
ExecStart=/path/to/venv/bin/gunicorn \
          --workers 4 \
          --bind unix:/tmp/myapp.sock \
          --access-logfile /var/log/myapp/access.log \
          --error-logfile /var/log/myapp/error.log \
          "myapp:create_app()"
Restart=always

[Install]
WantedBy=multi-user.target

3. uWSGI部署

uWSGI是另一个流行的WSGI服务器，支持更多高级特性。

uWSGI配置文件

# myapp_uwsgi.ini
[uwsgi]
# 应用配置
module = myapp:create_app()
callable = app

# 服务器配置
http = :8000
master = true
processes = 4
threads = 2

# 进程管理
harakiri = 30
max-requests = 1000
reload-on-rss = 200

# 日志配置
logto = /var/log/uwsgi/myapp.log
log-format = %(addr) - %(user) [%(ltime)] "%(method) %(uri) %(proto)" %(status) %(size) "%(referer)" "%(uagent)"

启动命令

uwsgi --ini myapp_uwsgi.ini

总结

Flask应用工厂函数模式提供了：

• 灵活性：支持多环境配置
• 可测试性：便于单元测试和集成测试
• 可维护性：代码组织清晰
• 可扩展性：易于添加新功能
• 部署友好：支持多种部署方式

对于中小型项目，简单的工厂函数就足够了。对于大型项目，建议采用更结构化的工厂函数，包含配置管理、扩展初始化、蓝图注册、错误处理等完整功能。