如何在SQLAlchemy-flask应用程序中执行原始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 text

sql = text('select name from penguins')
result = db.engine.execute(sql)
names = [row[0] for row in result]
print names

Answer 2

如果您想使用会话(如您的问题所示),请execute直接使用其方法:

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

以下可能特定于我的数据库驱动程序(psycopg2); 我不确定.无论如何,这是我如何取出我的价值观.

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

关键是db.session电话.这execute部分只是我发现通过提供基于名称的访问使我的生活更轻松的东西.

此外,这是事务性的,无需手动管理.Say :val是一个创建会话的函数:

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)

Answer 3

docs: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, 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')

Answer 4

对于 SQLAlchemy \xe2\x89\xa5 1.4

\n

从 SQLAlchemy 1.4 开始，无连接或隐式执行已被弃用，即

\n

db.engine.execute(...) # DEPRECATED\n

\n

以及作为查询的裸字符串。

\n

新的 API 需要显式连接，例如

\n

from sqlalchemy import text\n\nwith db.engine.connect() as connection:\n    result = connection.execute(text("SELECT * FROM ..."))\n    for row in result:\n        # ...\n

\n

同样，it\xe2\x80\x99s 鼓励使用现有会话（如果可用）：

\n

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

\n

或使用参数：

\n

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

\n

有关更多详细信息，请参阅文档中的“无连接执行、隐式执行”。

\n

Answer 5

你可以使用SELECT SQL查询的结果from_statement(),并text()如图所示这里.你不必以这种方式处理元组.作为类名为'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

Answer 6

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

<sql here>除非你处于autocommit模式,否则执行但不提交它.因此,插入和更新不会反映在数据库中.

要在更改后提交,请执行

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