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

你有没有试过:

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

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

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

要在更改后提交，请执行

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

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

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

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