如何在SQLAlchemy中执行原始SQL ?
我有一个python web应用程序,运行在flask和接口到数据库通过SQLAlchemy。
我需要一种方法来运行原始SQL。该查询涉及多个表连接和内联视图。
我试过了:
connection = db.session.connection()
connection.execute( <sql here> )
但是我总是得到网关错误。
如何在SQLAlchemy中执行原始SQL ?
我有一个python web应用程序,运行在flask和接口到数据库通过SQLAlchemy。
我需要一种方法来运行原始SQL。该查询涉及多个表连接和内联视图。
我试过了:
connection = db.session.connection()
connection.execute( <sql here> )
但是我总是得到网关错误。
当前回答
可以使用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表达式语言教程-使用文本
例子:
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')
这是如何从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))
这使用当前有应用程序的数据库连接。
对于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,则鼓励使用现有的Session:
result = session.execute(sqlalchemy.text("SELECT * FROM ..."))
或者使用参数:
session.execute(sqlalchemy.text("SELECT * FROM a_table WHERE a_column = :val"),
{'val': 5})
更多细节请参见文档中的“无连接执行,隐式执行”。
如果你想避免元组,另一种方法是调用第一个,一个或所有方法:
query = db.engine.execute("SELECT * FROM blogs "
"WHERE id = 1 ")
assert query.first().name == "Welcome to my blog"
你试过使用连接吗?执行(文本(<sql here>), <绑定params here>)和绑定参数描述在文档?这可以帮助解决许多参数格式化和性能问题。也许网关错误是超时?绑定参数往往使复杂的查询执行得更快。