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

【问题标题】：How to execute raw SQL in Flask-SQLAlchemy app如何在 Flask-SQLAlchemy 应用程序中执行原始 SQL
【发布时间】：2013-08-01 01:48:30
【问题描述】：

如何在 SQLAlchemy 中执行原始 SQL？

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

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

我试过了：

connection = db.session.connection()
connection.execute( <sql here> )

但我不断收到网关错误。

【问题讨论】：

  • 我以前看过，但我找不到有关运行更新的教程。我也不想学习语法并隐藏一个相当长（大约 20 行）的 SQL 查询。
  • @MarkusUnterwaditzer 我曾经这么认为，但现在我强烈反对。原始的、经过适当参数化的 SQL 通常比一堆函数调用和生成它的对象更容易阅读和维护。它还为您提供了数据库的全部功能，而无需费力地让 ORM 生成正确的语法（如果可能的话）并防止 ORM 做意外的事情。你可能会问，“那为什么要使用 SQLAlchemy？”，而我唯一的答案是，“现有的应用程序使用它，而改变一切都太昂贵了。”
  • @jpmc26 更新了您的评论——作为 SQL 的爱好者，我很难接受将“数据库的密钥”交给不负责任的炼金术士的想法，并且倾向于支持ORM is an antipattern :) 话虽如此，我很想加速某些组件，例如用户注册/管理，以及生成带有按钮序列的表，我可以为其编写操作 + SQL。您是否遇到过一些对 ORM 怀疑论者友好且在 Python 框架中非常适合您的工具？
  • @jpmc26 您在 Python 框架中使用什么来仅使用 SQL 或与 C# Dapper 非常接近？我在 Python Web 框架中看到的所有内容都希望我使用 SQLAlchemy，而我不喜欢 ORM，如果我使用 ORM，它会非常少。
  • @johnny 我还没有机会亲自尝试，但原始数据库连接库可能就足够了。例如，psycopg2 有直接返回namedtuple 和dict 的游标：initd.org/psycopg/docs/extras.html。

标签： python sql sqlalchemy flask flask-sqlalchemy


【解决方案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

注意db.engine.execute() 是“无连接”，即deprecated in SQLAlchemy 2.0。

【讨论】：

  • 如果进行插入或更新，如何提交事务？
  • 如果您使用的是原始 SQL，那么您可以控制事务，因此您必须自己发出 BEGIN 和 COMMIT 语句。
  • db.engine.execute(text("&lt;sql here&gt;")).execution_options(autocommit=True)) 也执行并提交它。
  • @Miguel “如果您使用原始 SQL，那么您可以控制事务，因此您必须自己发出 BEGIN 和 COMMIT 语句。”这是不正确的。您可以将原始 SQL 与会话对象一起使用。刚刚注意到这条评论，但你可以看到我的回答，了解如何使用原始 SQL 会话。
  • 我可以知道为什么我们需要将查询包装在text(...) 中吗？
【解决方案2】：

您是否尝试过使用connection.execute(text( &lt;sql here&gt; ), &lt;bind params here&gt; ) 并按照in the docs 的描述绑定参数？这可以帮助解决许多参数格式和性能问题。也许网关错误是超时？绑定参数往往会使复杂查询的执行速度大大加快。

【讨论】：

  • 根据docs，应该是connection.execute(text(&lt;sql here&gt;), &lt;bind params&gt; )。 bind params 不应该在 text() 中。 将绑定参数输入到 execute() 方法
  • Jake 的链接已损坏。我认为这是现在相关的 URL：docs.sqlalchemy.org/en/latest/core/…
【解决方案3】：

文档：SQL Expression Language Tutorial - Using Text

示例：

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

【讨论】：

  • sqlalchemy 文档的链接似乎已过期。这是最近的：docs.sqlalchemy.org/en/latest/core/…
  • 请问我们为什么使用==？
  • @Jake Berger 非常感谢你。我已经浪费了将近一天的时间来寻找这个答案。我只是直接执行sql而不转换为文本。每当我们的 where 子句中有 %students% 时，它就会抛出错误。为您的回答鼓掌。
  • @NamGVU 因为与大多数编程语言一样，= 通常保留用于分配一个值；而== 保留用于比较值
  • @JakeBerger 你有那个链接吗？ SQL 不是这样的语言，从SQLAlchemy docs 来看，情况并非如此。
【解决方案4】：

SQL Alchemy 会话对象有自己的execute 方法：

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

所有您的应用程序查询都应该通过会话对象，无论它们是否是原始 SQL。这确保了查询是正确的managed by a transaction，这允许同一请求中的多个查询作为一个单元提交或回滚。使用engine 或connection 退出事务会使您面临更大的风险，即可能难以检测到可能导致数据损坏的错误。每个请求应该只与一个事务相关联，并且使用db.session 将确保您的应用程序是这种情况。

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

假设它是一个 SELECT 查询，这将返回 an iterable 的 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

我个人更喜欢将结果转换成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)

如果您没有使用 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})

【讨论】：

  • 一个 Select 将返回一个 ResultProxy。
  • @AlanB 是的。当我称它为序列时，我用词不当，暗示它实现了序列协议。我已经更正并澄清了。谢谢。
  • @jpmc26 应该在执行 db.session.close() 之类的查询后关闭会话？它还会有连接池的好处吗？
  • dict(r.items()) 它的作品。确保您拥有这两个软件包的正确版本。 SQLAlchemy=
  • 有了这些我总是得到列表和TypeError: list indices错误......所以最后我去了results = db_session.query("my_table")........ .all()然后for row in results:然后row.my_column工作，或者如果你有一个专栏变量中的名称 => 然后我推荐 getattr(row, column_name_str_variable)
【解决方案5】：

您可以使用from_statement() 和text() 获得SELECT SQL 查询的结果，如here 所示。您不必以这种方式处理元组。以表名users 的类User 为例，您可以试试，

from sqlalchemy.sql import text

user = session.query(User).from_statement(
    text("""SELECT * FROM users where name=:name""")
).params(name="ed").all()

return user

【讨论】：

    【解决方案6】： 
    result = db.engine.execute(text("<sql here>"))

    执行&lt;sql here&gt;，但除非您处于autocommit 模式，否则不会提交它。因此，插入和更新不会反映在数据库中。

    要在更改后提交，请执行

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

    【讨论】：

      【解决方案7】：

      这是关于如何从 Flask Shell 运行 SQL 查询的简化答案

      首先，映射您的模块（如果您的模块/应用是主文件夹中的 manage.py 并且您在 UNIX 操作系统中），运行：

      export FLASK_APP=manage

      运行 Flask 外壳

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

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

      【讨论】：

        【解决方案8】：

        如果你想避免元组，另一种方法是调用first、one 或all 方法：

        query = db.engine.execute("SELECT * FROM blogs "
                           "WHERE id = 1 ")

assert query.first().name == "Welcome to my blog"

        【讨论】：

          【解决方案9】：

          对于 SQLAlchemy ≥ 1.4

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

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

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

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

          from sqlalchemy import text

with db.engine.connect() as connection:
    result = connection.execute(text("SELECT * FROM ..."))
    for row in result:
        # ...

          同样，如果有可用的Session，我们鼓励使用：

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

          或使用参数：

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

          有关详细信息，请参阅文档中的“Connectionless Execution, Implicit Execution”。

          【讨论】：

          • 现在，这应该是公认的答案。
          猜你喜欢
          • 2015-12-05
          • 1970-01-01
          • 1970-01-01
          • 2018-09-08
          • 2014-06-06
          • 2016-09-04
          • 2015-11-08
          • 1970-01-01
          • 2015-10-20
          相关资源
          最近更新 更多
          热门标签
          Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode