Python 如何在 Flask-SQLAlchemy 应用程序中执行原始 SQL
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/17972020/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
How to execute raw SQL in Flask-SQLAlchemy app
提问by starwing123
How do you execute raw SQL in SQLAlchemy?
你如何在 SQLAlchemy 中执行原始 SQL？
I have a python web app that runs on flask and interfaces to the database through SQLAlchemy.
我有一个 python web 应用程序，它运行在烧瓶上，并通过 SQLAlchemy 连接到数据库。
I need a way to run the raw SQL. The query involves multiple table joins along with Inline views.
我需要一种运行原始 SQL 的方法。该查询涉及多个表连接以及内联视图。
I've tried:
我试过了：
connection = db.session.connection()
connection.execute( <sql here> )
But I keep getting gateway errors.
但我不断收到网关错误。
采纳答案by Miguel
Have you tried:
你有没有尝试过：
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
回答by jhnwsk
Have you tried using connection.execute(text( <sql here> ), <bind params here> )and bind parameters as described in the docs? This can help solve many parameter formatting and performance problems. Maybe the gateway error is a timeout? Bind parameters tend to make complex queries execute substantially faster.
您是否尝试过按照文档中的connection.execute(text( <sql here> ), <bind params here> )描述使用和绑定参数？这可以帮助解决许多参数格式和性能问题。也许网关错误是超时？绑定参数往往会使复杂查询的执行速度大大加快。
回答by Jake Berger
docs: SQL Expression Language Tutorial - Using Text
example:
例子：
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')
回答by jpmc26
SQL Alchemy session objects have their own executemethod:
SQL Alchemy 会话对象有自己的execute方法：
result = db.session.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})
Allyour 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 engineor the connectionputs 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.sessionwill ensure this is the case for your application.
您的所有应用程序查询都应该通过会话对象，无论它们是否是原始 SQL。这确保查询由事务正确管理，这允许同一请求中的多个查询作为单个单元提交或回滚。使用引擎或连接离开事务会使您面临更大的微妙风险，可能难以检测到可能导致数据损坏的错误。每个请求应该只与一个事务相关联，使用db.session将确保您的应用程序是这种情况。
Also take note that executeis designed for parameterized queries. Use parameters, like :valin the example, for any inputs to the query to protect yourself from SQL injection attacks. You can provide the value for these parameters by passing a dictas the second argument, where each key is the name of the parameter as it appears in the query. The exact syntax of the parameter itself may be different depending on your database, but all of the major relational databases support them in some form.
另请注意，它execute是为参数化查询而设计的。:val对查询的任何输入使用参数（如示例中所示）以保护自己免受 SQL 注入攻击。您可以通过将 adict作为第二个参数传递来为这些参数提供值，其中每个键是在查询中出现的参数名称。参数本身的确切语法可能因您的数据库而异，但所有主要关系数据库都以某种形式支持它们。
Assuming it's a SELECTquery, this will return an iterableof RowProxyobjects.
假设它是一个SELECT查询，这将返回一个可迭代的RowProxy对象。
You can access individual columns with a variety of techniques:
您可以使用多种技术访问各个列：
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
Personally, I prefer to convert the results into namedtuples:
就个人而言，我更喜欢将结果转换为namedtuples：
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)
If you're not using the Flask-SQLAlchemy extension, you can still easily use a session:
如果您没有使用 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})
回答by TrigonaMinima
You can get the results of SELECT SQL queries using from_statement()and text()as shown here. You don't have to deal with tuples this way. As an example for a class Userhaving the table name usersyou can try,
你可以使用SELECT SQL查询的结果from_statement()，并text()如图所示这里。您不必以这种方式处理元组。作为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
回答by Devi
result = db.engine.execute(text("<sql here>"))
executes the <sql here>but doesn't commit it unless you're on autocommitmode. So, inserts and updates wouldn't reflect in the database.
执行<sql here>但不提交，除非您处于autocommit模式。因此，插入和更新不会反映在数据库中。
To commit after the changes, do
要在更改后提交，请执行
result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))
回答by Luigi Lopez
This is a simplified answer of how to run SQL query from Flask Shell
这是如何从 Flask Shell 运行 SQL 查询的简化答案
First, map your module (if your module/app is manage.py in the principal folder and you are in a UNIX Operating system), run:
首先，映射您的模块（如果您的模块/应用程序是主体文件夹中的 manage.py 并且您在 UNIX 操作系统中），运行：
export FLASK_APP=manage
Run Flask shell
运行烧瓶外壳
flask shell
Import what we need::
导入我们需要的东西::
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy(app)
from sqlalchemy import text
Run your query:
运行您的查询：
result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))
This use the currently database connection which has the application.
这使用具有应用程序的当前数据库连接。
回答by Joe Gasewicz
If you want to avoid tuples, another way is by calling the first, oneor allmethods:
如果您想避免使用元组，另一种方法是调用first,one或all方法：
query = db.engine.execute("SELECT * FROM blogs "
"WHERE id = 1 ")
assert query.first().name == "Welcome to my blog"
