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

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

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

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

SQL Alchemy会话对象有自己的执行方法:

result = db.session.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})

All your application queries should be going through a session object, whether they're raw SQL or not. This ensures that the queries are properly managed by a transaction, which allows multiple queries in the same request to be committed or rolled back as a single unit. Going outside the transaction using the engine or the connection puts you at much greater risk of subtle, possibly hard to detect bugs that can leave you with corrupted data. Each request should be associated with only one transaction, and using db.session will ensure this is the case for your application.

还要注意，execute是为参数化查询设计的。使用参数，如示例中的:val，用于查询的任何输入，以保护自己免受SQL注入攻击。您可以通过传递dict作为第二个参数来提供这些参数的值，其中每个键都是查询中出现的参数的名称。参数本身的确切语法可能因数据库而异，但所有主要的关系数据库都以某种形式支持它们。

假设它是一个SELECT查询，这将返回一个RowProxy对象的可迭代对象。

您可以使用各种技术访问单个列:

for r in result:
    print(r[0]) # Access by positional index
    print(r['my_column']) # Access by column name as a string
    r_dict = dict(r.items()) # convert to dict keyed by column names

就我个人而言，我更喜欢将结果转换为命名元组:

from collections import namedtuple

Record = namedtuple('Record', result.keys())
records = [Record(*r) for r in result.fetchall()]
for r in records:
    print(r.my_column)
    print(r)

如果您没有使用Flask-SQLAlchemy扩展，您仍然可以轻松地使用会话:

import sqlalchemy
from sqlalchemy.orm import sessionmaker, scoped_session

engine = sqlalchemy.create_engine('my connection string')
Session = scoped_session(sessionmaker(bind=engine))

s = Session()
result = s.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})

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

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