Flask 教程

Flask 数据库集成

lichongyang 约 4,102 字 6 分钟阅读 #004

在之前的“Flask 数据库操作”中，我们学习了如何使用 Flask-SQLAlchemy 进行基础的 CRUD（增删改查）。

今天，我们将视角拔高，探讨 “数据库集成 (Database Integration)”。这不仅仅是写几行查询代码，而是关乎如何在企业级项目架构中优雅、高效、安全地将数据库与 Flask 应用结合起来。

一、 数据库选型指南

在集成之前，首先要根据业务场景选择合适的数据库：

数据库类型代表产品适用场景Flask 集成方案
关系型 (RDBMS)PostgreSQL (首选), MySQL核心业务数据、需要强一致性、复杂关联查询（如订单、用户、财务）。Flask-SQLAlchemy
键值/缓存 (NoSQL)Redis高频读取、会话(Session)存储、分布式锁、排行榜、消息队列。Flask-Redis 或原生 redis-py
文档型 (NoSQL)MongoDB数据结构不固定、海量日志、内容管理系统、物联网数据。Flask-PyMongo

二、 核心难点：在“应用工厂”中正确集成

在大型项目中，我们使用应用工厂模式 (create_app)。新手最容易在这里遇到 “循环导入 (Circular Import)” 和 “缺少应用上下文 (Working outside of application context)” 的错误。

以下是标准且防坑的集成姿势：

1. 抽离扩展实例 (app/extensions.py)

绝对不要在 __init__.py 中直接 db = SQLAlchemy(app)，这会导致循环导入。

纯文本
plaintext
# app/extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

# 1. 只实例化，不绑定 app
db = SQLAlchemy()
migrate = Migrate()

2. 在工厂中初始化 (app/__init__.py)

纯文本
plaintext
# app/__init__.py
from flask import Flask
from config import Config
from .extensions import db, migrate

def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)

    # 2. 绑定扩展到 app
    db.init_app(app)
    migrate.init_app(app, db)

    # 3. 注册蓝图
    from app.main import main_bp
    app.register_blueprint(main_bp)

    # 4. 创建数据库表 (仅限开发环境，生产环境请用 flask db upgrade)
    with app.app_context():
        # 导入所有模型，确保 SQLAlchemy 知道它们的存在
        from app import models 
        db.create_all()

    return app

3. 定义模型 (app/models.py)

纯文本
plaintext
# app/models.py
from .extensions import db

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)

三、 生产环境必备：连接池与引擎配置

在开发环境（SQLite）中，你不需要关心连接池。但在生产环境（MySQL/PostgreSQL）中，数据库连接是宝贵的资源，必须配置连接池 (Connection Pool)。

在 config.py 中添加以下配置：

纯文本
plaintext
import os

class Config:
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')

    # 核心：SQLAlchemy 引擎配置 (针对生产环境优化)
    SQLALCHEMY_ENGINE_OPTIONS = {
        # 连接池大小：保持 10 个常驻连接
        'pool_size': 10,
        # 最大溢出连接数：高峰期允许额外创建 20 个连接 (总共最多 30 个)
        'max_overflow': 20,
        # 连接回收时间：MySQL 默认 8 小时断开空闲连接，设置 3600 秒提前回收，防止使用死连接
        'pool_recycle': 3600,
        # 预检 ping：每次从连接池获取连接前，先 ping 一下数据库，确保连接可用 (极力推荐)
        'pool_pre_ping': True,
        # 获取连接超时时间
        'pool_timeout': 30
    }

四、 进阶：ORM 与原生 SQL 混用

虽然 ORM 很方便，但在执行复杂统计、批量更新、多表复杂 JOIN 时，ORM 生成的 SQL 可能效率低下。此时可以直接执行原生 SQL。

在 SQLAlchemy 2.0+ 和 Flask-SQLAlchemy 3.0+ 中，推荐使用 text() 和 db.session.execute()：

纯文本
plaintext
from sqlalchemy import text
from app.extensions import db

def get_user_statistics():
    # 1. 编写原生 SQL (使用 :param 作为占位符，防止 SQL 注入)
    sql = text("""
        SELECT u.department, COUNT(u.id) as user_count 
        FROM users u
        WHERE u.status = :status
        GROUP BY u.department
    """)

    # 2. 执行并传入参数
    result = db.session.execute(sql, {'status': 'active'})

    # 3. 处理结果
    stats = []
    for row in result:
        stats.append({'department': row.department, 'count': row.user_count})

    return stats

五、 集成 NoSQL：以 Redis 为例

现代 Web 应用几乎离不开 Redis（用于缓存、Session、限流等）。

1. 安装

纯文本
plaintext
pip install redis

2. 在工厂中集成

我们不需要特定的 Flask 扩展，直接使用 redis-py 结合 Flask 的 g 对象或应用上下文即可。

纯文本
plaintext
# app/extensions.py
import redis
from flask import current_app

# 创建一个获取 Redis 客户端的辅助函数
def get_redis_client():
    # 使用 current_app 确保在应用上下文中
    redis_url = current_app.config.get('REDIS_URL', 'redis://localhost:6379/0')
    # 也可以考虑将 client 缓存在 app.extensions 中，避免每次请求都重新连接
    if 'redis_client' not in current_app.extensions:
        current_app.extensions['redis_client'] = redis.from_url(redis_url, decode_responses=True)
    return current_app.extensions['redis_client']

3. 在视图中使用

纯文本
plaintext
# app/main/routes.py
from flask import Blueprint, jsonify
from app.extensions import get_redis_client

main_bp = Blueprint('main', __name__)

@main_bp.route('/page-views')
def page_views():
    r = get_redis_client()

    # 增加页面访问量
    views = r.incr('page:home:views')

    # 设置过期时间 (可选)
    r.expire('page:home:views', 86400) # 1天

    return jsonify({"views": views})

六、 数据库集成的“避坑”指南

  1. 上下文问题 (RuntimeError: Working outside of application context)：
  • 原因：在 Flask 应用上下文之外（如普通的 Python 脚本、Celery 任务、后台线程）访问了 db 或 current_app。
  • 解决：手动推入上下文。 from app import create_app from app.extensions import db app = create_app() with app.app_context(): # 在这里执行数据库操作 db.session.execute(...)
  1. N+1 查询问题：
  • 现象：查询 100 篇文章，然后循环获取每篇文章的作者，导致执行了 101 次 SQL。
  • 解决：使用 joinedload 或 selectinload 进行预加载。
    python from sqlalchemy.orm import selectinload posts = Post.query.options(selectinload(Post.author)).all()
  1. 事务未提交/未回滚：
  • Flask-SQLAlchemy 会在请求结束时自动提交（如果没有异常）。但如果你在后台任务或自定义线程中操作，必须手动 db.session.commit()，并在异常时 db.session.rollback()。

📝 总结

Flask 数据库集成的核心在于：

  1. 架构清晰：使用 extensions.py 解耦，使用工厂模式初始化。
  2. 生产就绪：配置合理的连接池参数 (pool_pre_ping 是救命稻草)。
  3. 灵活变通：ORM 处理常规业务，原生 SQL 处理复杂统计，Redis 处理高频缓存。

至此，你已经掌握了 Flask 从基础路由到生产级数据库集成的全套技能。如果你想继续深入，我们可以探讨：

  1. Flask 异步编程 (Async/Await)：如何让 Flask 支持高并发 I/O？
  2. Celery 异步任务队列：如何处理发邮件、生成报表等耗时任务？
  3. 编写单元测试：如何测试你的数据库操作和 API？

$相关文章

延伸阅读

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注