问题描述：

如何在 SQLAlchemy 中执行原始 SQL？

我有一个在 flask 上运行并通过 SQLAlchemy 与数据库交互的 python web 应用程序。

我需要一种方法来运行原始 SQL。查询涉及多个表连接以及内联视图。

我尝试过：

connection = db.session.connection()
connection.execute( <sql here> )

但我一直收到网关错误。

解决方案 1：

SQLAlchemy 2.0：

with engine.connect() as connection:
    result = connection.execute(text('SELECT * FROM your_table'))
    # do something with the result..

SQLAlchemy 1.x：

from sqlalchemy import text

sql = text('select name from penguins')
result = db.engine.execute(sql)
names = [row[0] for row in result]
print names

请注意，这db.engine.execute()是“无连接”的，在 SQLAlchemy 2.0 中已弃用。

解决方案 2：

SQL Alchemy 会话对象有自己的execute方法：

result = db.session.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})

所有应用程序查询都应通过会话对象，无论它们是否是原始 SQL。这可确保查询由事务正确管理，从而允许同一请求中的多个查询作为单个单元提交或回滚。使用引擎或连接超出事务范围会使您面临更大的风险，因为可能难以检测到的细微错误可能会导致数据损坏。每个请求应仅与一个事务相关联，使用db.session将确保您的应用程序符合这一要求。

还请注意，它execute是为参数化查询而设计的。使用参数（如:val示例中所示）作为查询的任何输入，以保护自己免受 SQL 注入攻击。您可以通过传递dict作为第二个参数来提供这些参数的值，其中每个键都是查询中出现的参数的名称。参数本身的确切语法可能因数据库而异，但所有主要关系数据库都以某种形式支持它们。

假设它是一个SELECT查询，它将返回一个可迭代的RowProxy对象。

您可以使用多种技术访问各个列：

for r in result:
    print(r[0]) # Access by positional index
    print(r['my_column']) # Access by column name as a string
    r_dict = dict(r.items()) # convert to dict keyed by column names

就我个人而言，我更喜欢将结果转换为namedtuple：

from collections import namedtuple

Record = namedtuple('Record', result.keys())
records = [Record(*r) for r in result.fetchall()]
for r in records:
    print(r.my_column)
    print(r)

如果您没有使用 Flask-SQLAlchemy 扩展，您仍然可以轻松使用会话：

import sqlalchemy
from sqlalchemy.orm import sessionmaker, scoped_session

engine = sqlalchemy.create_engine('my connection string')
Session = scoped_session(sessionmaker(bind=engine))

s = Session()
result = s.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})

解决方案 3：

from_statement()您可以使用和获得 SELECT SQL 查询的结果，如下text()所示。您不必以这种方式处理元组。例如，对于具有表名的类，您可以尝试，User`users`

from sqlalchemy.sql import text

user = session.query(User).from_statement(
    text("""SELECT * FROM users where name=:name""")
).params(name="ed").all()

return user

解决方案 4：

文档：SQL 表达式语言教程 - 使用文本

例子：

from sqlalchemy.sql import text

connection = engine.connect()

# recommended
cmd = 'select * from Employees where EmployeeGroup = :group'
employeeGroup = 'Staff'
employees = connection.execute(text(cmd), group = employeeGroup)

# or - wee more difficult to interpret the command
employeeGroup = 'Staff'
employees = connection.execute(
                  text('select * from Employees where EmployeeGroup = :group'), 
                  group = employeeGroup)

# or - notice the requirement to quote 'Staff'
employees = connection.execute(
                  text("select * from Employees where EmployeeGroup = 'Staff'"))


for employee in employees: logger.debug(employee)
# output
(0, 'Tim', 'Gurra', 'Staff', '991-509-9284')
(1, 'Jim', 'Carey', 'Staff', '832-252-1910')
(2, 'Lee', 'Asher', 'Staff', '897-747-1564')
(3, 'Ben', 'Hayes', 'Staff', '584-255-2631')

解决方案 5：

对于 SQLAlchemy ≥ 1.4

从 SQLAlchemy 1.4 开始，无连接或隐式执行已被弃用，即

db.engine.execute(...) # DEPRECATED

以及裸字符串作为查询。

新的 API 需要显式连接，例如

from sqlalchemy import text

with db.engine.connect() as connection:
    result = connection.execute(text("SELECT * FROM ..."))
    for row in result:
        # ...

类似地，如果有可用的会话，也鼓励使用现有的会话：

result = session.execute(sqlalchemy.text("SELECT * FROM ..."))

或使用参数：

session.execute(sqlalchemy.text("SELECT * FROM a_table WHERE a_column = :val"),
                {'val': 5})

有关更多详细信息，请参阅文档中的“无连接执行，隐式执行”。

解决方案 6：

result = db.engine.execute(text("<sql here>"))

执行<sql here>但不提交，除非您处于autocommit模式。因此，插入和更新不会反映在数据库中。

要在更改后提交，请执行以下操作

result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))

解决方案 7：

这是关于如何从 Flask Shell 运行 SQL 查询的简化答案

首先，映射您的模块（如果您的模块/应用程序是主文件夹中的 manage.py 并且您在 UNIX 操作系统中），请运行：

export FLASK_APP=manage

运行 Flask shell

flask shell

导入我们需要的内容::

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy(app)
from sqlalchemy import text

运行查询：

result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))

这使用当前具有应用程序的数据库连接。

解决方案 8：

Flask-SQLAlchemy v: 3.0.x / SQLAlchemy v: 1.4

users = db.session.execute(db.select(User).order_by(User.title.desc()).limit(150)).scalars()

因此基本上对于flask-sqlalchemy的最新稳定版本，文档具体建议将该session.execute()方法与 结合使用db.select(Object)。

解决方案 9：

您是否尝试过按照文档中的connection.execute(text( <sql here> ), <bind params here> )说明使用和绑定参数？这可以帮助解决许多参数格式和性能问题。也许网关错误是超时？绑定参数往往会使复杂查询的执行速度大大加快。

解决方案 10：

如果要避免使用元组，另一种方法是调用first、one或all方法：

query = db.engine.execute("SELECT * FROM blogs "
                           "WHERE id = 1 ")

assert query.first().name == "Welcome to my blog"