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

Question

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

pythonsqlsqlalchemyflaskflask-sqlalchemy

335

如何在SQLAlchemy中执行原始SQL语句？

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

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

我尝试过：

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

但我一直收到网关错误。

- starwing123

7

我之前看过这个，但是找不到有关运行更新操作的教程。我也不太想学习语法并转换一个相当长（约20行）的SQL查询语句。 - starwing123

144

我以前也这样认为，但现在我强烈不同意。原始的、正确参数化的 SQL 通常比一堆函数调用和生成它的对象更易于阅读和维护。它还为您提供了完整的数据库功能，而无需跳过各种障碍来使 ORM 生成正确的语法（如果可能的话），并防止 ORM 做出意外的事情。你可能会问，“那么为什么还要使用 SQLAlchemy？”，我唯一的答案是，“现有应用程序使用它，改变一切太昂贵了。” - jpmc26

7

作为一个热爱SQL的人，我很难接受将“数据库的钥匙”交给不负责任的炼金术士的想法，并倾向于认为ORM是一种反模式。然而，我很想加速某些组件，比如用户注册/管理，还有生成带有按钮序列的表格，以便我可以编写操作和SQL语句。你在Python框架中是否遇到过适合ORM怀疑论者使用的工具呢？ - zx81

1

@jpmc26 你在Python框架中使用什么来仅使用SQL或类似于C# Dapper的东西？我在Python web框架中看到的所有内容都想让我使用SQLAlchemy，而我不喜欢ORM，如果我使用ORM，它也是非常简单的。 - johnny

@johnny 我自己还没有尝试过，但是原始数据库连接库可能已经足够了。例如，psycopg2具有直接返回namedtuple和dict的游标：http://initd.org/psycopg/docs/extras.html。 - jpmc26

2

已接受的答案已过时。请参见下面 @Demitri 的答案。 - Deepam Gupta

10个回答

282

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

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

所有与应用程序相关的查询都应该通过会话对象进行，无论它们是否是原始 SQL 查询。这样可以确保查询被正确地由事务管理，从而允许在同一请求中提交或回滚多个查询作为单个单元。如果使用engine或connection超出事务范围，则会面临更大的风险，可能会导致难以检测的错误，使您的数据损坏。每个请求应该仅关联一个事务，使用db.session将确保此事务适用于您的应用程序。请注意，execute是为参数化查询设计的。对于查询中的任何输入，请使用参数（例如示例中的:val）来保护自己免受SQL注入攻击。您可以通过将字典作为第二个参数传递来为这些参数提供值，其中每个键都是参数在查询中出现的名称。参数本身的确切语法可能因您的数据库而异，但所有主要关系型数据库都以某种形式支持它们。假设这是一个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})

- jpmc26

1

一个Select语句会返回一个ResultProxy对象。 - Alan B

@AlanB 是的。当我称它为序列并暗示它实现了序列协议时，我的措辞不够准确。我已经进行了更正和澄清。谢谢。 - jpmc26

2

@jpmc26在执行查询后应该关闭会话，例如db.session.close()吗？这样做还会有连接池的好处吗？ - ravi malhotra

字典(r.items())的功能正常。请确保这两个包的版本正确：SQLAlchemy=<1.3.5，Flask-SQLAlchemy=<2.4.1。 - Ramesh Ponnusamy

我总是遇到列表和“TypeError: list indices”错误，所以最终我选择了results = db_session.query("my_table")........ .all()，然后使用for row in results:，接着row.my_column就可以工作了。如果你有一个变量存储列名，那么我建议使用getattr(row, column_name_str_variable)。 - jave.web

71

你可以使用 from_statement() 和 text() 来获取 SELECT SQL 查询的结果，如此示例所示此处。这样你就不必处理元组。例如对于一个名为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

- TrigonaMinima

你让我的一天变得美好。这正是我正在寻找的。你真是我的救星！ - Vijay Yarramsetty

69

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

- Jake Berger

1

SQLAlchemy文档的链接似乎已经过期了。这是更新版本：http://docs.sqlalchemy.org/en/latest/core/tutorial.html#using-textual-sql - Carl

1

我可以问一下为什么我们要使用 == 吗？ - Nam G VU

1

@Jake Berger，非常感谢您。我浪费了将近一天的时间寻找这个答案。我直接执行SQL而没有将其转换为文本。每当我们在where子句中有%students%时，它就会抛出错误。对于您的答案，我给予热烈的掌声。 - Suresh Kumar

1

@NamGVU 因为像大多数编程语言一样，= 通常用于 _赋值_；而 == 则用于比较值。 - Jake Berger

2

@JakeBerger，你有相关链接吗？SQL并不是这样的一种语言，而根据SQLAlchemy文档来看也不是这样的。 - johndodo

显示剩余2条评论

52

对于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:
        # ...

同样地，如果有现成的Session可用，鼓励使用：

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

或使用参数：

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

请参见文档中的"Connectionless Execution, Implicit Execution"了解更多详情。

- Demitri

2

现在，这应该是被接受的答案。 - Deepam Gupta

14

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

执行<sql here>语句，但如果不处于autocommit模式，则不会提交。因此，插入和更新操作不会反映在数据库中。

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

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

- Devi

2

这是一个简化的答案，说明如何从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))

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

- Luigi Lopez

1

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)结合使用。

- Panagiss

0

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

- jhnwsk

2

根据文档，应该是connection.execute(text(<sql here>), <bind params>)。bind params不应该在text()中。将绑定参数提供给execute()方法即可。 - Jake Berger

Jake的链接已经失效了。我认为这个URL现在是相关的：https://docs.sqlalchemy.org/en/latest/core/sqlelement.html?highlight=text#sqlalchemy.sql.expression.text - code_dredd

-1

如果你想避免元组，另一种方法是通过调用first、one或all方法：

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

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

- Joe Gasewicz

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，

- Miguel Grinberg · Accepted Answer

419

你尝试过吗：

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

10

如果您执行插入或更新操作，您如何提交事务？ - David S

17

如果您使用原始SQL，则需要自行控制事务，因此必须自行发出“BEGIN”和“COMMIT”语句。 - Miguel Grinberg

35

db.engine.execute(text("<sql here>")).execution_options(autocommit=True)) 执行并提交它。 - Devi

12

“如果你在使用原始SQL，那么你需要控制事务，因此你必须自己发出BEGIN和COMMIT语句。”这并不完全正确。你可以使用session对象来处理原始的SQL语句。我注意到了这条评论，但你可以看看我的答案，了解如何在使用原始SQL时使用session对象。 - jpmc26

8

为什么我们需要在查询语句中使用text(...)进行包装？ - Nam G VU

显示剩余9条评论