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

Question

你如何在 SQLAlchemy 中执行原始 SQL？

我有一个 python web 应用程序，它在 flask 上运行并通过 SQLAlchemy 连接到数据库。

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

我试过了：

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

但我不断收到网关错误。

原文由 starwing123 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

你有没有尝试过：

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

要么：

 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 中已弃用。

原文由 Miguel Grinberg 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 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 s：

 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})

原文由 jpmc26 发布，翻译遵循 CC BY-SA 4.0 许可协议