如何在SQLAlchemy-flask应用程序中执行原始SQL
你如何在SQLAlchemy中执行原始SQL？
我有一个python web应用程序运行在瓶和通过SQLAlchemy接口到数据库。
我需要一种方法来运行原始的SQL。 查询涉及多个表连接以及embedded视图。
我试过了：
connection = db.session.connection() connection.execute( <sql here> ) 但我不断收到网关错误。
你有没有尝试过：
result = db.engine.execute("<sql here>")
要么：
from sqlalchemy import text sql = text('select name from penguins') result = db.engine.execute(sql) names = [] for row in result: names.append(row[0]) print names
如果您想使用会话（如您的问题所示），请直接使用其execute方法：
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})
以下可能是特定于我的数据库驱动程序（psycopg2）; 我不确定。 无论如何，这是我如何拉出我的价值观。
from collections import namedtuple Record = namedtuple('Record', result.keys()) records = [Record(*r) for r in result.fetchall()] for r in records: print(r)
关键是fetchall()调用。 namedtuple部分只是我发现通过给予基于名称的访问使我的生活更容易。
而且，这是事务性的， 不需要手动pipe理。 说make_session是一个创build会话的函数：
>>> s1 = make_session() >>> s1.execute('CREATE TABLE blah (id INTEGER)') <sqlalchemy.engine.result.ResultProxy object at 0x02CD86F0> >>> s1.commit() >>> >>> s1.execute('INSERT INTO blah VALUES (1)') <sqlalchemy.engine.result.ResultProxy object at 0x02CD8870> >>> s1.execute('SELECT * FROM blah').fetchall() [(1,)] >>> >>> s2 = make_session() >>> s2.execute('SELECT * FROM blah').fetchall() [] >>> s2.close() >>> >>> s1.commit() >>> >>> s2 = make_session() >>> s2.execute('SELECT * FROM blah').fetchall() [(1,)] >>> s2.close() >>> s1.close()
文档： SQLexpression式语言教程 – 使用文本
例：
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, 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')
您可以使用from_statement()和text()来获得SELECT SQL查询的结果，如下所示。 你不必用这种方法来处理混淆。 作为一个具有表名'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
result = db.engine.execute(text("<sql here>"))执行<sql here>但不会提交它，除非您处于自动提交模式。 所以，插入和更新不会反映在数据库中。
要在变更后提交，请执行
result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))
你有没有尝试过使用connection.execute(text( <sql here> ), <bind params here> )和绑定参数的文档中所述 ？ 这可以帮助解决许多参数格式和性能问题。 也许网关错误是超时？ 绑定参数往往会使复杂的查询执行速度大大加快。
