如何在SQLAlchemy中执行原始SQL ?

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

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

我试过了:

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

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


当前回答

如果你想避免元组,另一种方法是调用第一个,一个或所有方法:

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

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

其他回答

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

可以使用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

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

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

如果你想避免元组,另一种方法是调用第一个,一个或所有方法:

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

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