使用python MySQLdb执行*.sql文件

Question

如何使用MySQLdb python驱动程序执行存储在*.sql文件中的sql脚本.我在努力

cursor.execute(file(PATH_TO_FILE).read())

但这不起作用,因为cursor.execute一次只能运行一个sql命令.我的sql脚本包含几个sql语句.我也在努力

cursor.execute('source %s'%PATH_TO_FILE)

但也没有成功.

Answer 1

从python,我启动一个mysql进程来为我执行该文件:

from subprocess import Popen, PIPE
process = Popen(['mysql', db, '-u', user, '-p', passwd],
                stdout=PIPE, stdin=PIPE)
output = process.communicate('source ' + filename)[0]

Answer 2

我还需要执行一个SQL文件,但问题是每行没有一个语句,所以接受的答案对我不起作用.

我想要执行的SQL文件如下所示:

-- SQL script to bootstrap the DB:
--
CREATE USER 'x'@'%' IDENTIFIED BY 'x';
GRANT ALL PRIVILEGES ON mystore.* TO 'x'@'%';
GRANT ALL ON `%`.* TO 'x'@`%`;
FLUSH PRIVILEGES;
--
--
CREATE DATABASE oozie;
GRANT ALL PRIVILEGES ON oozie.* TO 'oozie'@'localhost' IDENTIFIED BY 'oozie';
GRANT ALL PRIVILEGES ON oozie.* TO 'oozie'@'%' IDENTIFIED BY 'oozie';
FLUSH PRIVILEGES;
--
USE oozie;
--
CREATE TABLE `BUNDLE_ACTIONS` (
  `bundle_action_id` varchar(255) NOT NULL,
  `bundle_id` varchar(255) DEFAULT NULL,
  `coord_id` varchar(255) DEFAULT NULL,
  `coord_name` varchar(255) DEFAULT NULL,
  `critical` int(11) DEFAULT NULL,
  `last_modified_time` datetime DEFAULT NULL,
  `pending` int(11) DEFAULT NULL,
  `status` varchar(255) DEFAULT NULL,
  `bean_type` varchar(31) DEFAULT NULL,
  PRIMARY KEY (`bundle_action_id`),
  KEY `I_BNDLTNS_DTYPE` (`bean_type`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;
--
--

上面文件中的一些语句位于一行,一些语句也跨越多行(如末尾的CREATE TABLE).还有一些SQL内联注释行以" - "开头.

正如ThomasK所建议的那样,我必须编写一些简单的规则来将行连接到一个语句中.我最终得到了一个执行sql文件的函数:

def exec_sql_file(cursor, sql_file):
    print "\n[INFO] Executing SQL script file: '%s'" % (sql_file)
    statement = ""

    for line in open(sql_file):
        if re.match(r'--', line):  # ignore sql comment lines
            continue
        if not re.search(r';$', line):  # keep appending lines that don't end in ';'
            statement = statement + line
        else:  # when you get a line ending in ';' then exec statement and reset for next statement
            statement = statement + line
            #print "\n\n[DEBUG] Executing SQL statement:\n%s" % (statement)
            try:
                cursor.execute(statement)
            except (OperationalError, ProgrammingError) as e:
                print "\n[WARN] MySQLError during execute statement \n\tArgs: '%s'" % (str(e.args))

            statement = ""

我确信还有改进的余地,但是现在它对我来说效果很好.希望有人发现它有用.

Answer 3

for line in open(PATH_TO_FILE):
    cursor.execute(line)

这假设您的文件中每行有一个SQL语句.否则,您需要编写一些规则来将线连接在一起.

Answer 4

这对我有用:

with open('schema.sql') as f:
    cursor.execute(f.read().decode('utf-8'), multi=True)

Answer 5

至少MySQLdb1.2.3似乎允许开箱即用,你只需要调用cursor.nextset()循环返回的结果集.

db = conn.cursor()
db.execute('SELECT 1; SELECT 2;')

more = True
while more:
    print db.fetchall()
    more = db.nextset()

如果您想完全确定已启用对此的支持,和/或禁用支持,您可以使用以下内容:

MYSQL_OPTION_MULTI_STATEMENTS_ON = 0
MYSQL_OPTION_MULTI_STATEMENTS_OFF = 1

conn.set_server_option(MYSQL_OPTION_MULTI_STATEMENTS_ON)
# Multiple statement execution here...
conn.set_server_option(MYSQL_OPTION_MULTI_STATEMENTS_OFF)

Answer 6

当您的 sql 脚本包含空行并且您的查询语句跨越多行时，接受的答案将遇到问题。相反，使用以下方法将解决问题：

f = open(filename, 'r')
query = " ".join(f.readlines())
c.execute(query)

Answer 7

另一个无需任何解析即可利用MySQL解释器的解决方案是使用以下os.system命令在python内部直接运行MySQL提示命令：

from os import system
USERNAME = "root"
PASSWORD = "root"
DBNAME = "pablo"
HOST = "localhost"
PORT = 3306
FILE = "file.sql"
command = """mysql -u %s -p"%s" --host %s --port %s %s < %s""" %(USERNAME, PASSWORD, HOST, PORT, DBNAME, FILE)
system(command)

这样可以避免任何解析错误，例如，当您有一个带有笑脸的字符串变量，;-)或者如果您将_ ;作为最后一个字符进行检查，以及之后是否有注释，例如SELECT * FROM foo_table; # selecting data

Answer 8

这里的许多答案都有严重的缺陷。

首先，不要尝试自己解析开放式sql脚本！如果您认为这很容易做到，那么您就不会知道sql多么健壮和复杂。严肃的sql脚本当然涉及跨越多行的语句和过程定义。在脚本的中间显式声明和更改定界符也很常见。您也可以在彼此之间嵌套源命令。由于许多原因，您希望通过MySQL客户端运行脚本并允许其处理繁重的工作。试图重塑这是充满危险和时间的巨大浪费。也许如果您是唯一编写这些脚本的人，并且您编写的并不是复杂的东西，而又为什么要限制自己呢？机器生成的脚本或其他开发人员编写的脚本呢？

@jdferreira的答案是正确的，但也存在问题和不足。最重要的是，通过以这种方式向过程发送连接参数来打开安全漏洞。

这是您复制和粘贴乐趣的解决方案/示例。我的扩展讨论如下：

首先，创建一个单独的配置文件以保存您的用户名和密码。

db-creds.cfg

[client]
user     = XXXXXXX
password = YYYYYYY

在其上添加正确的文件系统权限，以便python进程可以从中读取内容，但是没有人可以查看谁不应该这样做。

然后，使用此Python（在我的示例中为creds文件与py脚本相邻）：

#!/usr/bin/python

import os
import sys
import MySQLdb
from subprocess import Popen, PIPE, STDOUT

__MYSQL_CLIENT_PATH = "mysql"

__THIS_DIR = os.path.dirname( os.path.realpath( sys.argv[0] ) )

__DB_CONFIG_PATH    = os.path.join( __THIS_DIR, "db-creds.cfg" )
__DB_CONFIG_SECTION = "client"

__DB_CONN_HOST = "localhost"
__DB_CONN_PORT = 3306

# ----------------------------------------------------------------

class MySqlScriptError( Exception ):

    def __init__( self, dbName, scriptPath, stdOut, stdErr ):
        Exception.__init__( self )
        self.dbName = dbName
        self.scriptPath = scriptPath
        self.priorOutput = stdOut
        self.errorMsg = stdErr                
        errNumParts = stdErr.split("(")        
        try : self.errorNum = long( errNumParts[0].replace("ERROR","").strip() )
        except: self.errorNum = None        
        try : self.sqlState = long( errNumParts[1].split(")")[0].strip() )
        except: self.sqlState = None

    def __str__( self ): 
        return ("--- MySqlScriptError ---\n" +
                "Script: %s\n" % (self.scriptPath,) +
                "Database: %s\n" % (self.dbName,) +
                self.errorMsg ) 

    def __repr__( self ): return self.__str__()

# ----------------------------------------------------------------

def databaseLoginParms() :        
    from ConfigParser import RawConfigParser
    parser = RawConfigParser()
    parser.read( __DB_CONFIG_PATH )   
    return ( parser.get( __DB_CONFIG_SECTION, "user" ).strip(), 
             parser.get( __DB_CONFIG_SECTION, "password" ).strip() )

def databaseConn( username, password, dbName ):        
    return MySQLdb.connect( host=__DB_CONN_HOST, port=__DB_CONN_PORT,
                            user=username, passwd=password, db=dbName )

def executeSqlScript( dbName, scriptPath, ignoreErrors=False ) :       
    scriptDirPath = os.path.dirname( os.path.realpath( scriptPath ) )
    sourceCmd = "SOURCE %s" % (scriptPath,)
    cmdList = [ __MYSQL_CLIENT_PATH,                
               "--defaults-extra-file=%s" % (__DB_CONFIG_PATH,) , 
               "--database", dbName,
               "--unbuffered" ] 
    if ignoreErrors : 
        cmdList.append( "--force" )
    else:
        cmdList.extend( ["--execute", sourceCmd ] )
    process = Popen( cmdList 
                   , cwd=scriptDirPath
                   , stdout=PIPE 
                   , stderr=(STDOUT if ignoreErrors else PIPE) 
                   , stdin=(PIPE if ignoreErrors else None) )
    stdOut, stdErr = process.communicate( sourceCmd if ignoreErrors else None )
    if stdErr is not None and len(stdErr) > 0 : 
        raise MySqlScriptError( dbName, scriptPath, stdOut, stdErr )
    return stdOut

如果要对其进行测试，请添加以下内容：

if __name__ == "__main__": 

    ( username, password ) = databaseLoginParms()
    dbName = "ExampleDatabase"

    print "MySQLdb Test"
    print   
    conn = databaseConn( username, password, dbName )
    cursor = conn.cursor()
    cursor.execute( "show tables" )
    print cursor.fetchall()
    cursor.close()
    conn.close()
    print   

    print "-----------------"
    print "Execute Script with ignore errors"
    print   
    scriptPath = "test.sql"
    print executeSqlScript( dbName, scriptPath, 
                            ignoreErrors=True )
    print   

    print "-----------------"
    print "Execute Script WITHOUT ignore errors"                            
    print   
    try : print executeSqlScript( dbName, scriptPath )
    except MySqlScriptError as e :        
        print "dbName: %s" % (e.dbName,)
        print "scriptPath: %s" % (e.scriptPath,)
        print "errorNum: %s" % (str(e.errorNum),)
        print "sqlState: %s" % (str(e.sqlState),)
        print "priorOutput:"        
        print e.priorOutput
        print
        print "errorMsg:"
        print e.errorMsg           
        print
        print e
    print   

出于良好的考虑，下面是一个示例sql脚本供您输入：

test.sql

show tables;
blow up;
show tables;

因此，现在进行一些讨论。

首先，我说明了如何在执行此外部脚本的同时使用MySQLdb，同时将凭据存储在一个共享文件中，您可以将两者同时使用。

通过--defaults-extra-file在命令行上使用，您可以安全地传入连接参数。

结合--force使用stdin流源命令或--execute在外部运行该命令，您可以决定脚本的运行方式。那是通过忽略错误并继续运行，或者一旦发生错误就停止运行。

结果恢复的顺序也将通过保留--unbuffered。否则，您的stdout和stderr流将混乱且无法按顺序定义，这使得与输入sql进行比较时，很难确定哪些有效，哪些无效。

使用Popen，cwd=scriptDirPath您可以使用相对路径在彼此之间嵌套源命令。如果所有脚本都位于同一目录（或相对于该目录的已知路径）中，那么您可以参考相对于顶级脚本所在位置的那些目录。

最后，我抛出了一个异常类，该类携带了您可能想要有关发生的事情的所有信息。如果您没有使用ignoreErrors选项，则当出现问题并且脚本已停止运行该错误时，将在Python中引发这些异常之一。