如何在Flask-SQLAlchemy应用程序中执行原始SQL

Question

如何在SQLAlchemy中执行原始SQL？

我有一个python web应用程序，它运行在烧瓶上，并通过SQLAlchemy与数据库连接。

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

我试过了：

connection = db.session.connection()

connection.execute( <sql here> )

但我不断收到网关错误。

Answer 1

你有没有尝试过：result = db.engine.execute("<sql here>")要么：from sqlalchemy import textsql = text('select name from penguins')result = db.engine.execute(sql)names = [row[0] for row in result]print names

Answer 2

SQL Alchemy会话对象有自己的execute方法：result = db.session.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})您的所有查询都应该通过会话对象，无论它们是否是原始SQL。这可确保查询由事务正确管理，从而允许将同一请求中的多个查询作为单个单元提交或回滚。使用引擎或连接走出事务会使您面临更大的微妙风险，可能很难检测到可能导致数据损坏的错误。每个请求应仅与一个事务相关联，并且使用db.session将确保您的应用程序的情况。假设它是一个SELECT查询，这将返回一个可迭代的RowProxy对象。您可以使用各种技术访问各个列：for r in result:&nbsp; &nbsp; print(r[0]) # Access by positional index&nbsp; &nbsp; print(r['my_column']) # Access by column name as a string&nbsp; &nbsp; r_dict = dict(r.items()) # convert to dict keyed by column names就个人而言，我更喜欢将结果转换为namedtuples：from collections import namedtupleRecord = namedtuple('Record', result.keys())records = [Record(*r) for r in result.fetchall()]for r in records:&nbsp; &nbsp; print(r)如果您没有使用Flask-SQLAlchemy扩展，您仍然可以轻松使用会话：import sqlalchemyfrom sqlalchemy.orm import sessionmaker, scoped_sessionengine = 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})

Answer 3

docs：SQL表达式语言教程 - 使用文本例：from sqlalchemy.sql import textconnection = engine.connect()# recommendedcmd = 'select * from Employees where EmployeeGroup == :group'employeeGroup = 'Staff'employees = connection.execute(text(cmd), group = employeeGroup)# or - wee more difficult to interpret the commandemployeeGroup = 'Staff'employees = connection.execute(&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; text('select * from Employees where EmployeeGroup == :group'),&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; group = employeeGroup)# or - notice the requirement to quote "Staff"employees = connection.execute(&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; text('select * from Employees where EmployeeGroup == "Staff"'))for employee in employees: logger.debug(employee)# output(0, u'Tim', u'Gurra', u'Staff', u'991-509-9284')(1, u'Jim', u'Carey', u'Staff', u'832-252-1910')(2, u'Lee', u'Asher', u'Staff', u'897-747-1564')(3, u'Ben', u'Hayes', u'Staff', u'584-255-2631')