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

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

result = db.session.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})
全部你的应用程序查询应该通过一个会话对象，无论它们是否是原始SQL。这可以确保查询是正确的由交易管理的它允许同一请求中的多个查询作为一个单元被提交或回滚。在事务之外使用engine or the connection让你面临更大的风险，可能难以发现的错误会让你的数据被破坏。每个请求应该只与一个事务相关联，使用db.session将确保你的应用程序是这样的。
还需要注意的是，execute是针对参数化查询.使用参数，如例子中的:val，用于查询的任何输入，以保护自己免受SQL注入攻击。你可以通过传递dict作为第二个参数来提供这些参数的值，其中每个键是参数的名称，因为它出现在查询中。参数本身的确切语法可能因你的数据库而异，但所有主要的关系型数据库都以某种形式支持它们。
假设它是一个SELECT的查询，这将返回一个可迭代的对象 of RowProxy objects.
你可以用各种技术访问各个栏目。
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)
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})