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

Question

如何在SQLAlchemy中执行原始SQL？

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

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

我试过：

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

但我总是会遇到网关错误。

Answer 1

文档：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')

Answer 2

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

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

所有应用程序查询都应该通过会话对象，无论它们是否是原始SQL。这确保查询由事务正确管理，这允许将同一请求中的多个查询作为一个单元提交或回滚。使用引擎或连接在事务之外会使您面临更大的风险，即难以检测可能会给您留下损坏数据的错误。每个请求应该只与一个事务相关联，使用db.session将确保您的应用程序就是这种情况。

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

假设这是一个SELECT查询，它将返回RowProxy对象的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})

Answer 3

你有没有试过：

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中是不推荐的。