Flask 如何在Flask-SQLAlchemy应用程序中执行原生SQL

阅读更多： Flask 教程

连接数据库

在开始执行原生SQL之前，我们首先需要连接到数据库。使用Flask-SQLAlchemy，我们可以在应用程序的配置中指定数据库的连接字符串。下面是一个示例配置：

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://username:password@localhost/db_name'
上面的代码将配置应用程序使用MySQL数据库，并指定连接的用户名、密码和数据库名。你可以根据自己的需要修改连接字符串。
执行原生SQL语句
在Flask-SQLAlchemy中执行原生SQL语句非常简单。我们可以使用db.engine.execute()方法来执行SQL语句，并获取执行结果。下面是执行原生SQL语句的示例代码：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://username:password@localhost/db_name'
db = SQLAlchemy(app)
@app.route('/')
def execute_sql():
    # 执行原生SQL语句
    sql = "SELECT * FROM users"
    result = db.engine.execute(sql)
    # 处理执行结果
    rows = []
    for row in result:
        rows.append(row)
    return str(rows)
上面的代码中，我们定义了一个路由处理函数execute_sql()。在函数中，我们通过执行SELECT * FROM users语句从数据库中选取了所有用户的信息，并将结果以字符串形式返回。你可以根据自己的需要修改SQL语句和处理结果的方式。
使用SQLAlchemy Session对象执行原生SQL
除了使用db.engine.execute()方法执行原生SQL语句外，我们还可以使用SQLAlchemy的Session对象执行原生SQL。Session对象提供了更多的操作方法和功能。下面是使用Session对象执行原生SQL的示例代码：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import sessionmaker
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://username:password@localhost/db_name'
db = SQLAlchemy(app)
Session = sessionmaker(bind=db.engine)
@app.route('/')
def execute_sql():
    # 创建Session对象
    session = Session()
    # 执行原生SQL语句
    sql = "SELECT * FROM users"
    result = session.execute(sql)
    # 处理执行结果
    rows = []
    for row in result:
        rows.append(row)
    session.close()
    return str(rows)
上面的代码中，我们首先创建了一个Session对象，然后使用该对象执行原生SQL语句。执行结果和处理方式与之前的示例相同。记得在处理完结果后关闭Session。
使用参数化查询
在执行原生SQL语句时，我们经常需要向SQL语句中传递参数。为了避免SQL注入攻击，我们应该使用参数化查询来处理这些参数。在Flask-SQLAlchemy中，我们可以使用sqlalchemy.sql.expression.text()方法创建参数化查询的查询字符串。下面是一个使用参数化查询的示例：
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.sql import text
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://username:password@localhost/db_name'
db = SQLAlchemy(app)
@app.route('/')
def execute_sql():
    # 创建参数化查询的查询字符串
    sql = text("SELECT * FROM users WHERE age > :age")
    # 执行参数化查询
    result = db.engine.execute(sql, age=18)
    # 处理执行结果
    rows = []
    for row in result:
        rows.append(row)
    return str(rows)
上述示例中，我们使用sqlalchemy.sql.expression.text()方法创建了一个参数化查询的查询字符串。在查询字符串中使用:age作为参数占位符。然后，我们使用db.engine.execute()方法执行参数化查询，并指定参数的值。在这个例子中，我们查询了年龄大于18岁的用户信息。
在本文中，我们介绍了如何在Flask-SQLAlchemy应用程序中执行原生SQL语句。我们首先连接到数据库，然后使用db.engine.execute()方法或Session对象执行原生SQL。我们还讨论了如何使用参数化查询来处理传递给SQL语句的参数。通过掌握这些技巧，你可以更灵活地操作数据库，并满足特定的需求。希望本文对你在Flask-SQLAlchemy应用程序中执行原生SQL有所帮助！
            
Flask 问答
Flask 使用marshmallow从SQLAlchemy更新行Flask 强制应用程序/json MIME类型在视图中（Flask）Flask 使用gunicorn附加命令行参数Flask: 无法将jinja2变量传递到JavaScript片段Flask 在 Flask 中使用 SwaggerFlask Jinja Templates - 将浮点数格式化为逗号分隔的货币Flask 两个常用方法 json.dumps 和 flask.jsonify之间的区别和用法Flask CLI 在通过 docker-compose 运行时会抛出 'OSError: Exec format error' 错误Flask Flask应用程序没有异常会引发500错误Flask 在独立线程中启动Flask应用Flask Flask在Flask中执行两次Flask 禁用Flask中的缓存Flask 常见问题：ValueError: signal only works in main threadFlask 错误：在路径中找不到可执行文件“python”Flask 如何在测试中使用SQLAlchemy设置Flask应用程序Flask 将变量传递给 Flask 的 render_templateFlask如何查询Many-To-Many关系的SQLAlchemyFlask 实现强制页面刷新的最佳方法是什么Flask sqlalchemy postgresql enum 在数据库迁移中没有创建类型Flask 不是用于动态选择字段的有效选择Flask 如何在Flask-sqlalchemy中使用count()函数Flask 在Gunicorn中共享内存Flask 如何使用SQLAlchemy选择一个列Flask 如何在Python中后台运行长时间任务Flask 如何在多线程模式下使用Gunicorn运行FlaskFlask 获取 Flask 应用的根路径Flask 如何在 Flask 中在 HTML 页面上显示图片Flask Import 无法在VS Code上的Python 3.9.2中解析或从Pylance源中解析Flask 如何将Flask运行在80端口Flask 在多个地方重复使用Jinja2中的代码块Flask Python中utf-8编码下的Jsonify字典转换问题Flask 如何在本地无需使用https测试flask-oauthlibFlask Docker环境中的Python变量Flask：将Python字典转化为HTML表格Flask Gunicorn与Flask使用错误的Python版本问题Flask 如何使用 virtualenv 运行 uwsgiFlask 如何在Flask模板中显示当前年份Flask 中如何在路由函数外获取路径参数Flask 使用Flask_login和jinja2模板Flask 使用 Python 和 Flask 返回 API 错误消息Flask 从同一文件夹中导入函数Flask flask-sqlalchemy - PostgreSQL - 为表定义特定的模式Flask .ini 文件加载环境变量Flask Alembic：alembic revision导致导入错误Flask AttributeError: 'Flask'对象没有属性'user_options'Flask 动态更新数据，无需刷新页面Flask Flask-principal教程（认证+授权）Flask TypeError: list object is not an iterator错误Flask：从Flask返回一个requests.Response对象Flask 在Heroku上部署Flask应用Flask 如何在Flask-SQLAlchemy应用程序中执行原生SQLFlask 文件 "/usr/bin/pip", 第9行, 在  中, from pip import main ImportError: 无法导入名称 mainFlask Flask 日志根本不起作用Flask 调试器PIN是什么Flask 如何在Flask中从配置文件中导入Flask 使用Flask获取select标签中的值Flask 使用基本的Flask和Flask-RESTful进行API开发Flask 从Flask传递一个JSON对象给JavaScriptFlask 如何在Flask和Flask-login中传递“next”URLFlask 在Flask中测试文件上传Flask 错误："方法不允许。该方法不允许请求的URL"Flask 使用Flask上传和下载文件Flask 如何使Flask-SQLAlchemy在引发异常时自动回滚会话Flask Flask App：在函数运行时更新进度条Flask 在jinja2模板中使用CSS样式表Flask 使用函数在Flask-SQLAlchemy模型上查询时出现“BaseQuery对象不可调用”的错误Flask Apache SetEnv 使用mod_wsgi时的功能不符合预期Flask 有一种公认的在Flask中使用API密钥进行身份验证的方式吗Flask：Python Flask导入错误Flask 使用Flask获取访问者的IP地址Flask 如何在Flask中使用ajax调用上传文件Flask：Flask同时验证多个字段的装饰器Flask Flask-SQLAlchemy - 大于或等于Flask 通过Curl发送JSON请求到FlaskFlask 运行问题和错误请求Flask 中类似于 Jinja2/Flask 中 'startswith' 方法的方法Flask Pip 默认行为与 virtualenv 冲突Flask 在Windows上使用waitress来提供Flask应用程序Flask Flask-Migrate没有创建表格Flask uWSGI, Flask, sqlalchemy和postgres: SSL错误：解密失败或错误的记录macFlask 如何在Windows上运行AirflowFlask 是 'from flask import request' 和 'import requests' 相同吗Flask flask_uploads: ImportError: 无法导入名为 'secure_filename' 的模块Flask 注册蓝图时出现“function”对象没有属性“name”Flask 如何在 Celery 任务中使用 Flask-SQLAlchemyFlask Flask-Login 检查用户是否被验证，不使用装饰器Flask 如何使用Flask-Cache和Redis缓存SQL Alchemy调用Flask Flask-RESTful - 上传图片Flask 如何在SQLAlchemy查询中使用avg和sum函数Flask Flask RESTful 跨域问题与 Angular: PUT, OPTIONS方法Flask 改变python/flask中使用Twitter Bootstrap CSS的链接的活动类Flask 如何在Python中使用Flask压缩/最小化JSON/Jsonify的大小Flask 中的 Tensor 不是这个图的元素；部署 Keras 模型Flask：使用render_template方法从变量中渲染文本Flask 我能在Flask和SQLAlchemy中避免循环引用吗Flask Flask和Flask-WTForms中有没有一种方法可以在循环遍历表单时隐藏csrf标签Flask Flask中可以进行POST请求吗Flask 在wtforms中给字段添加css类Flask 如何在Flask中获取HTTP头Flask 获取Jinja2中的请求参数Flask Python Flask关闭事件处理器