在 python 中运行多个 sql 语句的建议方法？

Posted 2023-02-24

【中文标题】在 python 中运行多个 sql 语句的建议方法？

【英文标题】：Suggested way to run multiple sql statements in python?

【发布时间】：2020-10-16 11:01:12

【问题描述】：

在 python 中运行类似以下内容的建议方法是什么：

self.cursor.execute('SET FOREIGN_KEY_CHECKS=0; DROP TABLE IF EXISTS %s; SET FOREIGN_KEY_CHECKS=1' % (table_name,))

例如，这应该是三个独立的self.cursor.execute(...) 语句吗？除了cursor.execute(...) 之外，还有什么特定的方法可以用来做这样的事情，或者建议的做法是什么？目前我的代码如下：

self.cursor.execute('SET FOREIGN_KEY_CHECKS=0;')
self.cursor.execute('DROP TABLE IF EXISTS %s;' % (table_name,))
self.cursor.execute('SET FOREIGN_KEY_CHECKS=1;')
self.cursor.execute('CREATE TABLE %s select * from mytable;' % (table_name,))

如您所见，一切都是单独运行的......所以我不确定这是否是一个好主意（或者更确切地说 - 执行上述操作的最佳方法是什么）。或许BEGIN...END？

【问题讨论】：

这对您有帮助吗zodb.readthedocs.io/en/latest/transactions.html ？

告诉我们大局是什么。您列出的内容很不正常；可能有更好的方法来解决“真正的”问题。

另外，如果这是一次性任务，只需在命令行 mysql 中执行即可，不必担心 python 的限制。

@RickJames 是的，同意。这是一个日常脚本，用于备份表以进行审计（我们将它们保留十天）。将其放入 DW 或其他任何东西都不够重要，但我们会保留它以用于一些临时查询，以便跟踪跟踪一周的数据，如果需要用于审计目的。

您使用DROP TABLE ... CREATE TABLE 超过TRUNCATE TABLE 的任何特殊原因？

【参考方案1】：

在MySQLCursor.execute()的文档中，他们建议使用multi=True参数：

operation = 'SELECT 1; INSERT INTO t1 VALUES (); SELECT 2'
for result in cursor.execute(operation, multi=True):
    ...

您可以在模块的源代码中找到another example。

【参考方案2】：

我在项目中多次遇到此类问题。经过大量研究，我发现了一些要点和建议。

execute() 方法一次可以很好地处理一个查询。因为在执行方法期间会处理状态。

我知道cursor.execute(operation, params=None, multi=True) 接受多个查询。但是在这种情况下参数不能很好地工作，有时内部错误异常也会破坏所有结果。代码变得庞大而模糊。甚至 docs 也提到了这一点。

executemany(operation, seq_of_params) 不是每次都实施的好习惯。因为产生一个或多个结果集的操作构成未定义的行为，并且允许（但不要求）实现在检测到结果集已通过调用操作创建时引发异常。 [来源-docs]

建议一：

制作一个查询列表，例如 -:

table_name = 'test'
quries = [
          'SET FOREIGN_KEY_CHECKS=0;',
          'DROP TABLE IF EXISTS ;'.format(table_name),
          'SET FOREIGN_KEY_CHECKS=1;',
          'CREATE TABLE  select * from mytable;'.format(table_name),
         ]
for query in quries:
    result = self.cursor.execute(query)
    # Do operation with result

建议2-：

用字典设置。 [you can also make this by executemany for recursive parameters for some special cases.]

quries = [
          'DROP TABLE IF EXISTS %(table_name);':'table_name': 'student',
          'CREATE TABLE %(table_name) select * from mytable;': 
           'table_name':'teacher',
          'SET FOREIGN_KEY_CHECKS=0;': ''
         ]
for data in quries:
    for query, parameter in data.iteritems():
        if parameter == '':
            result = self.cursor.execute(query)
            # Do something with result
        else:
            result = self.cursor.execute(query, parameter)
            # Do something with result

您还可以将 split 与脚本一起使用。 Not recommended

with connection.cursor() as cursor:
    for statement in script.split(';'):
        if len(statement) > 0:
             cursor.execute(statement + ';')

注意-：我主要使用list of query 方法，但在一些复杂的地方使用make dictionary 方法。

【参考方案3】：

查看 MySQLCursor.execute() 的文档。

它声称您可以传入一个multi 参数，该参数允许您在一个字符串中运行多个查询。

如果multi设置为True，execute()可以执行操作字符串中指定的多条语句。

multi 是 execute() 调用的可选第二个参数：

---

operation = 'SELECT 1; INSERT INTO t1 VALUES (); SELECT 2'
for result in cursor.execute(operation, multi=True):

---

【参考方案4】：

import mysql.connector

您可以执行以下命令，只需将 t1 和 episodes 替换为您自己的 tabaes 即可

tablename= "t1"
 mycursor.execute("SET FOREIGN_KEY_CHECKS=0; DROP TABLE IF EXISTS ; SET FOREIGN_KEY_CHECKS=1;CREATE TABLE  select * from episodes;".format(tablename, tablename),multi=True)

虽然这将运行，但您必须确保启用后生效的外键限制不会导致问题。

如果 tablename 是用户可以输入的，您应该考虑一个表名白名单。

Prepared statemnts 不适用于表名和列名，因此我们必须使用字符串替换来在正确的位置获取正确的表名，这会使您的代码 易受攻击 到 sql注入

multi=True 是在连接器中运行 4 个命令所必需的，当我测试它时，调试器要求它。

【讨论】：

为什么是 import mysql.connector ？我在代码中没有看到。

有三个不同的python mysql连接器，这个使用官方的，其他的使用稍微不同的语法..你没有指定你使用的是哪个库，所以我澄清了，，，

【参考方案5】：

所有答案都是完全有效的，所以我将使用静态类型和closing 上下文管理器添加我的解决方案。

from contextlib import closing
from typing import List

import mysql.connector
import logging

logger = logging.getLogger(__name__)


def execute(stmts: List[str]) -> None:
    logger.info("Starting daily execution")

    with closing(mysql.connector.connect()) as connection:
        try:
            with closing(connection.cursor()) as cursor:
                cursor.execute(' ; '.join(stmts), multi=True)
        except Exception:
            logger.exception("Rollbacking changes")
            connection.rollback()
            raise
        else:
            logger.info("Finished successfully")

如果我没记错连接或游标可能不是上下文管理器，这取决于您拥有的 mysql 驱动程序的版本，所以这是一个 pythonic 安全的解决方案。

【参考方案6】：

执行脚本() 这是一次执行多个 SQL 语句的便捷方法。它执行作为参数获取的 SQL 脚本。 语法：

sqlite3.connect.executescript(script)

示例代码：

import sqlite3 

# Connection with the DataBase 
# 'library.db' 
connection = sqlite3.connect("library.db") 
cursor = connection.cursor() 

# SQL piece of code Executed 
# SQL piece of code Executed 
cursor.executescript(""" 
    CREATE TABLE people( 
        firstname, 
        lastname, 
        age 
    ); 

    CREATE TABLE book( 
        title, 
        author, 
        published 
    ); 

    INSERT INTO 
    book(title, author, published) 
    VALUES ( 
        'Dan Clarke''s GFG Detective Agency', 
        'Sean Simpsons', 
        1987 
    ); 
    """) 

sql = """ 
SELECT COUNT(*) FROM book;"""

cursor.execute(sql) 

# The output in fetched and returned 
# as a List by fetchall() 
result = cursor.fetchall() 
print(result) 

sql = """ 
SELECT * FROM book;"""

cursor.execute(sql) 

result = cursor.fetchall() 
print(result) 

# Changes saved into database 
connection.commit() 

# Connection closed(broken) 
# with DataBase 
connection.close() 

输出：

[(1,)] [（“丹·克拉克的 GFG 侦探社”，“肖恩·辛普森一家”，1987 年）]

executemany() 通常情况下，必须将大量数据从数据文件插入数据库（对于更简单的情况，采用列表、数组）。多次迭代代码比每次写入数据库更简单，每一行都写入数据库。但是在这种情况下不适合使用循环，下面的例子说明了原因。下面解释了 executemany() 的语法和用法，以及如何像循环一样使用它：

来源：GeeksForGeeks: SQL Using Python 看看这个来源..这有很多很棒的东西给你。

【参考方案7】：

我不会依赖execute 函数的任何multi=True 参数，它非常依赖于驱动程序，也不会尝试尝试在; 字符上拆分字符串，这可能嵌入在字符串文字中。最直接的方法是创建一个函数execute_multiple，它采用要执行的语句列表和rollback_on_error 参数来确定在任何语句导致异常时要执行的操作。

我对 MySQLdb 和 PyMySQL 的体验是，默认情况下它们从 autocommit=0 开始，换句话说，就像您已经在事务中并且需要显式提交一样。无论如何，这个假设适用于下面的代码。如果不是这种情况，那么您应该 1. 在连接后显式设置 autocommit=0 或 2. 修改此代码以在 try 语句之后启动事务

def execute_multiple(conn, statements, rollback_on_error=True):
    """
    Execute multiple SQL statements and returns the cursor from the last executed statement.

    :param conn: The connection to the database
    :type conn: Database connection

    :param statements: The statements to be executed
    :type statements: A list of strings

    :param: rollback_on_error: Flag to indicate action to be taken on an exception
    :type rollback_on_error: bool

    :returns cursor from the last statement executed
    :rtype cursor
    """

    try:
        cursor = conn.cursor()
        for statement in statements:
            cursor.execute(statement)
            if not rollback_on_error:
                conn.commit() # commit on each statement
    except Exception as e:
        if rollback_on_error:
            conn.rollback()
        raise
    else:
        if rollback_on_error:
            conn.commit() # then commit only after all statements have completed successfully

您还可以拥有一个使用参数列表处理预准备语句的版本：

def execute_multiple_prepared(conn, statements_and_values, rollback_on_error=True):
    """
    Execute multiple SQL statements and returns the cursor from the last executed statement.

    :param conn: The connection to the database
    :type conn: Database connection

    :param statements_and_values: The statements and values to be executed
    :type statements_and_values: A list of lists. Each sublist consists of a string, the SQL prepared statement with %s placeholders, and a list or tuple of its parameters

    :param: rollback_on_error: Flag to indicate action to be taken on an exception
    :type rollback_on_error: bool

    :returns cursor from the last statement executed
    :rtype cursor
    """

    try:
        cursor = conn.cursor()
        for s_v in statements_and_values:
            cursor.execute(s_v[0], s_v[1])
            if not rollback_on_error:
                conn.commit() # commit on each statement
    except Exception as e:
        if rollback_on_error:
            conn.rollback()
        raise
    else:
        if rollback_on_error:
            conn.commit() # then commit only after all statements have completed successfully
        return cursor # return the cursor in case there are results to be processed

例如：

cursor = execute_multiple_prepared(conn, [('select * from test_table where count = %s', (2000,))], False)

虽然，不可否认，上述调用只有一个带参数的 SQL 准备语句。

【讨论】：

这是一个很好的答案，感谢您的撰写和示例。

【参考方案8】：

美丽在旁观者的眼中，所以做某事的最佳方法是主观的，除非你明确告诉我们如何衡量。我可以看到三个假设选项：

使用 MySQLCursor 的 multi 选项（不理想） 将查询保留在多行中 将查询保留在一行中

或者，您还可以更改查询以避免一些不必要的工作。

---

关于multi 选项，MySQL documentation 对此非常清楚

如果 multi 设置为 True，execute() 能够执行操作字符串中指定的多个语句。它返回一个迭代器，可以处理每个语句的结果。但是，在这种情况下使用参数效果不佳，通常最好单独执行每个语句。

---

关于选项 2. 和 3. 这纯粹是您希望如何查看代码的偏好。回想一下，连接对象默认具有autocommit=FALSE，因此游标实际上将cursor.execute(...) 调用批处理到单个事务中。换言之，以下两个版本是等价的。

self.cursor.execute('SET FOREIGN_KEY_CHECKS=0;')
self.cursor.execute('DROP TABLE IF EXISTS %s;' % (table_name,))
self.cursor.execute('SET FOREIGN_KEY_CHECKS=1;')
self.cursor.execute('CREATE TABLE %s select * from mytable;' % (table_name,))

对

self.cursor.execute(
    'SET FOREIGN_KEY_CHECKS=0;'
    'DROP TABLE IF EXISTS %s;' % (table_name,)
    'SET FOREIGN_KEY_CHECKS=1;'
    'CREATE TABLE %s select * from mytable;' % (table_name,)
)

Python 3.6 引入了非常优雅的 f 字符串，如果可以的话，你应该使用它们。 :)

self.cursor.execute(
    'SET FOREIGN_KEY_CHECKS=0;'
    f'DROP TABLE IF EXISTS table_name;'
    'SET FOREIGN_KEY_CHECKS=1;'
    f'CREATE TABLE table_name select * from mytable;'
)

请注意，当您开始操作行时，这不再成立；在这种情况下，它成为特定于查询的，如果相关，您应该进行分析。一个相关的 SO 问题是What is faster, one big query or many small queries?

---

最后，使用TRUNCATE 代替DROP TABLE 可能更优雅，除非你有特定的理由不这样做。

self.cursor.execute(
    f'CREATE TABLE IF NOT EXISTS table_name;'
    'SET FOREIGN_KEY_CHECKS=0;'
    f'TRUNCATE TABLE table_name;'
    'SET FOREIGN_KEY_CHECKS=1;'
    f'INSERT INTO table_name SELECT * FROM mytable;'
)

【参考方案9】：

我会创建一个存储过程：

DROP PROCEDURE IF EXISTS CopyTable;
DELIMITER $$
CREATE PROCEDURE CopyTable(IN _mytable VARCHAR(64), _table_name VARCHAR(64))
BEGIN
    SET FOREIGN_KEY_CHECKS=0;
    SET @stmt = CONCAT('DROP TABLE IF EXISTS ',_table_name);
    PREPARE stmt1 FROM @stmt;
    EXECUTE stmt1;
    SET FOREIGN_KEY_CHECKS=1;
    SET @stmt = CONCAT('CREATE TABLE ',_table_name,' as select * from ', _mytable);
    PREPARE stmt1 FROM @stmt;
    EXECUTE stmt1;
    DEALLOCATE PREPARE stmt1;
END$$
DELIMITER ;

然后运行：

args = ['mytable', 'table_name']
cursor.callproc('CopyTable', args)

保持简单和模块化。当然你应该做一些错误检查，你甚至可以让存储过程返回一个代码来指示成功或失败。

【讨论】：

谢谢，这是一个不错的方法。出于好奇，为什么CopyTable 函数需要两个参数？ -- 一个是old_table，另一个是new_table，或者它是如何工作的？

是的，完全正确。我在模拟你在做什么，我什至用了和你一样的名字。 'table_name' 不存在，如果存在，它将被删除； 'mytable' 是我们从 [to 'table_name'] 复制的那个。诀窍，如果我们可以这样称呼它，就在 PREPARE 命令中:)

我仍然不清楚some kind of error checking and you could even have the store procedure return a code to indicate success or failure。能否请您详细说明这一点并给我一些参考？

看看本页底部dev.mysql.com/doc/connector-python/en/… 就是一个完美的例子。