1. Flask应用基础搭建与数据库连接实战

刚接触Flask时最让人头疼的就是如何规范地组织项目结构，特别是需要连接数据库的时候。我在三个不同企业的后台系统开发中都使用过Flask+SQLAlchemy的方案，这里分享一个经过实战检验的标准做法。

1.1 最小化Flask应用结构

先来看最基本的项目目录结构：

/project-root ├── app.py # 应用入口文件 ├── config.py # 配置文件 ├── requirements.txt # 依赖清单 └── /instance # 实例文件夹（存放数据库文件）

关键提示：永远不要把数据库配置直接写在app.py里，采用工厂模式创建app是更专业的做法

新建app.py时建议使用以下模板：

from flask import Flask def create_app(config_filename): app = Flask(__name__) app.config.from_pyfile(config_filename) # 初始化扩展 from extensions import db db.init_app(app) # 注册蓝图 from views import main_blueprint app.register_blueprint(main_blueprint) return app

1.2 配置管理的最佳实践

在config.py中建议采用类继承的方式管理配置：

class Config: SECRET_KEY = os.getenv('SECRET_KEY', 'dev_key') SQLALCHEMY_TRACK_MODIFICATIONS = False class DevelopmentConfig(Config): SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join( os.path.dirname(__file__), '../instance/dev.db' ) class ProductionConfig(Config): SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')

2. SQLAlchemy深度集成方案

2.1 数据库连接的核心配置

在extensions.py中集中管理扩展：

from flask_sqlalchemy import SQLAlchemy from flask_migrate import Migrate db = SQLAlchemy() migrate = Migrate()

连接MySQL时的特殊配置项：

app.config['SQLALCHEMY_ENGINE_OPTIONS'] = { 'pool_size': 10, 'pool_recycle': 300, 'pool_pre_ping': True }

2.2 模型定义规范

推荐使用以下模型基类：

class BaseModel(db.Model): __abstract__ = True id = db.Column(db.Integer, primary_key=True) created_at = db.Column(db.DateTime, default=db.func.now()) updated_at = db.Column(db.DateTime, default=db.func.now(), onupdate=db.func.now()) class User(BaseModel): username = db.Column(db.String(80), unique=True) email = db.Column(db.String(120), unique=True)

踩坑记录：SQLite的DateTime字段在服务重启后时区会重置，生产环境务必使用PostgreSQL或MySQL

3. 完整工作流实现

3.1 数据库迁移配置

1. 安装迁移工具：

pip install flask-migrate

2. 初始化迁移仓库：

from flask_migrate import Migrate migrate = Migrate(app, db)

3. 常用命令：

flask db init # 初始化 flask db migrate # 生成迁移脚本 flask db upgrade # 执行升级

3.2 连接池优化技巧

对于高并发场景需要调整连接池参数：

app.config['SQLALCHEMY_ENGINE_OPTIONS'] = { 'pool_size': 20, 'max_overflow': 10, 'pool_timeout': 30, 'pool_recycle': 3600 }

4. 生产环境问题排查

4.1 常见连接错误处理

1. MySQL连接超时：

app.config['SQLALCHEMY_ENGINE_OPTIONS'] = { 'pool_pre_ping': True, 'pool_recycle': 3600 }

2. 连接泄露检测：

from sqlalchemy import event from sqlalchemy.pool import Pool @event.listens_for(Pool, 'checkout') def on_checkout(dbapi_conn, connection_record, connection_proxy): print(f'Connection checked out: {dbapi_conn}')

4.2 性能监控方案

集成SQLAlchemy的事件监听：

from sqlalchemy import event from sqlalchemy.engine import Engine import time @event.listens_for(Engine, "before_cursor_execute") def before_cursor_execute(conn, cursor, statement, parameters, context, executemany): context._query_start_time = time.time() @event.listens_for(Engine, "after_cursor_execute") def after_cursor_execute(conn, cursor, statement, parameters, context, executemany): duration = time.time() - context._query_start_time if duration > 0.5: # 记录慢查询 app.logger.warning(f"Slow query: {statement} took {duration:.2f}s")

5. 进阶架构设计

5.1 多数据库连接方案

在大型项目中可能需要连接多个数据库：

class MultiDatabaseSQLAlchemy: def __init__(self): self._engines = {} def init_app(self, app): for name, uri in app.config['DATABASES'].items(): self._engines[name] = create_engine(uri) def get_engine(self, name='default'): return self._engines.get(name)

5.2 读写分离实现

通过路由策略实现读写分离：

class RoutingSession(SignallingSession): def get_bind(self, mapper=None, clause=None): if self._flushing: # 写操作使用主库 return engines['master'] return engines['slave']

在项目中使用异步SQLAlchemy时，需要特别注意连接生命周期管理。我通常在请求开始时获取连接，在请求结束时显式关闭：

@app.teardown_request def teardown_request(exception): if hasattr(g, 'db_session'): g.db_session.remove()