flask 只使用flask_SQLAlchemy连接池功能
时间: 2025-06-19 11:25:57 浏览: 22
### 在 Flask 中仅使用 Flask-SQLAlchemy 的连接池功能
尽管 Flask-SQLAlchemy 主要被设计用来作为 SQLAlchemy ORM 的封装工具,但它也提供了对原生 SQLAlchemy 引擎的支持。这意味着即使不想使用其 ORM 特性,仍然可以利用它内置的数据库连接池机制。
#### 实现步骤
1. **安装必要的包**
需要确保已安装 Flask 和 Flask-SQLAlchemy 。可以通过以下命令完成安装:
```bash
pip install flask
pip install flask_sqlalchemy
```
2. **配置 SQLALCHEMY_ENGINE_OPTIONS**
Flask-SQLAlchemy 支持通过 `SQLALCHEMY_ENGINE_OPTIONS` 来自定义 SQLAlchemy 引擎选项。这使得我们可以精确地调整连接池行为,而无需启用任何 ORM 功能。
3. **禁用模型声明**
为了防止自动加载 ORM 模型类,应避免调用 `db.Model` 或者其他与 ORM 相关的方法。
下面是具体实现代码:
```python
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
# 数据库URI示例 (MySQL 使用 pymysql)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://username:password@localhost/dbname'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 自定义引擎选项以优化连接池表现
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
"pool_size": 10, # 设置连接池大小为 10
"max_overflow": 5, # 超过连接池大小后最多可额外创建的连接数量
"pool_pre_ping": True # 开启预 ping 连接检测,提高稳定性
}
# 初始化 SQLAlchemy 对象
db = SQLAlchemy(app)
# 测试连接池功能
@app.route('/test')
def test_db():
with db.engine.connect() as connection:
result = connection.execute("SELECT NOW()")
current_time = result.fetchone()[0]
return f'Current Time from DB: {current_time}'
if __name__ == '__main__':
app.run(debug=True)
```
在这个例子中,`SQLALCHEMY_ENGINE_OPTIONS` 字典用于传递给 SQLAlchemy 的 `create_engine()` 方法[^5]。这些选项直接影响到连接池的行为模式,例如设定最大活跃连接数目(`pool_size`)、超出限制后的备用容量(`max_overflow`)以及是否开启预先心跳探测(`pool_pre_ping`)等功能。
值得注意的是,这里并没有涉及到任何关于 ORM 表结构定义的部分,而是单纯依靠 `db.engine.connect()` 接口执行原始 SQL 查询语句[^3]。
---
###
阅读全文
相关推荐


















