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

如何在SQLAlchemy中执行原始SQL ?

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

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

我试过了:

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

但是我总是得到网关错误。

373880 次浏览
小开
最佳答案

你有没有试过:

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

小开

你试过使用connection.execute(text( <sql here> ), <bind params here> )和绑定参数描述在文档中吗?这可以帮助解决许多参数格式化和性能问题。也许网关错误是超时?绑定参数往往使复杂的查询执行得更快。

小开

文档: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')
小开

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对象的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

就我个人而言，我更喜欢将结果转换为__abc:

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

你可以使用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
小开 
result = db.engine.execute(text("<sql here>"))

执行<sql here>但不提交它，除非你处于autocommit模式。因此，插入和更新不会反映在数据库中。

要在更改后提交，请执行

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

这是如何从Flask Shell运行SQL查询的简化答案

首先，映射您的模块(如果您的模块/应用程序在principal文件夹中是manage.py，并且您是在UNIX操作系统中)，运行:

export FLASK_APP=manage

运行烧瓶外壳

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

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

小开

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

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


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

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

类似地，如果已有会话可用，则鼓励使用它:

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

或者使用参数:

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

看到“无连接执行，隐式执行"有关详细信息，请参阅文档。

小开

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


提交答案