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

Question

如何在SQLAlchemy中执行原始SQL？

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

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

我试过：

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

但我一直有网关错误。

Answer 1

SQL炼金术会话对象有自己的execute方法：

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

所有应用程序查询都应通过会话对象，无论它们是否是原始SQL。这可以确保查询正确地managed by a transaction，从而允许同一请求中的多个查询作为单个单元提交或回滚。使用engine或connection离开事务会使您面临更大的风险，即可能很难检测到可能会导致数据损坏的错误。每个请求应该只与一个事务相关联，使用db.session将确保您的应用程序是这样的。

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

假设它是一个SELECT查询，这将返回an iterable个^{}对象。

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

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

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

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