如何在SQLAlchemy中执行原始SQL ?

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

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

我试过了:

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

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


当前回答

对于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})

更多细节请参见文档中的“无连接执行,隐式执行”。

其他回答

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

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

这是如何从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))

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

你有没有试过:

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

or:

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

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