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

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

要在更改后提交，请执行

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

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

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

要在更改后提交，请执行

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

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

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

类似地，如果有可用的Session，则鼓励使用现有的Session:

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

或者使用参数:

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

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

你有没有试过:

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

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

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