如何使用 MySQLdb python 驱动程序执行存储在 *.sql 文件中的 sql 脚本。我在尝试
cursor.execute(file(PATH_TO_FILE).read())
但这不起作用，因为 cursor.execute 一次只能运行一个 sql 命令。我的 sql 脚本包含几个 sql 语句。我也在尝试
cursor.execute('source %s'%PATH_TO_FILE)
但也没有成功。
【问题讨论】：
for line in open(PATH_TO_FILE):
cursor.execute(line)
这假设您的文件中的每一行都有一个 SQL 语句。否则，您需要编写一些规则来将线条连接在一起。
【讨论】：
line 传入的字符串已经被转义。或者它期待像execute('SOME SQL COMMAND blah WITH PARAMS %s', params) 这样的东西，然后它会为你正确地转义params ......但这在我描述的情况下不起作用。在这种情况下，@jotmicron 的答案可能会更好。幸运的是，我能够改用 django 固定装置。
从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]
【讨论】：
output = process.communicate(str.encode('source ' + filename))[0]
至少MySQLdb 1.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)
【讨论】：
我还需要执行一个 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 = ""
我确信还有改进的余地，但现在它对我来说效果很好。希望有人觉得它有用。
【讨论】：
/* ... comment ... comment ... comment ... comment ... */
加载mysqldump文件：
for line in open(PATH_TO_FILE).read().split(';\n'):
cursor.execute(line)
【讨论】：
;\n 不仅是可能的分隔符。就像这里的几个答案一样，这无法解释您可以（并且脚本经常这样做）在处理过程中自己更改分隔符的事实。此外，您可以将 MySQL 配置为默认使用不同的分隔符。在定义 sql 函数、过程、触发器等时，通常使用 $$ 作为分隔符，然后在其中使用 ;。
当您的 sql 脚本包含空行并且您的查询语句跨越多行时，接受的答案将遇到问题。相反，使用以下方法将解决问题：
f = open(filename, 'r')
query = " ".join(f.readlines())
c.execute(query)
【讨论】：
这是一个代码 sn-p，它将导入来自导出的典型 .sql。 （我成功地将它与 Sequel Pro 的导出一起使用。）处理多行查询和 cmets (#)。
import MySQLdb
from configdb import DB_HOST, DB_PASS, DB_USER, DB_DATABASE_NAME
db = MySQLdb.connect(host=DB_HOST, # your host, usually localhost
user=DB_USER, # your username
passwd=DB_PASS, # your password
db=DB_DATABASE_NAME) # name of the data base
cur = db.cursor()
PATH_TO_FILE = "db-testcases.sql"
fullLine = ''
for line in open(PATH_TO_FILE):
tempLine = line.strip()
# Skip empty lines.
# However, it seems "strip" doesn't remove every sort of whitespace.
# So, we also catch the "Query was empty" error below.
if len(tempLine) == 0:
continue
# Skip comments
if tempLine[0] == '#':
continue
fullLine += line
if not ';' in line:
continue
# You can remove this. It's for debugging purposes.
print "[line] ", fullLine, "[/line]"
try:
cur.execute(fullLine)
except MySQLdb.OperationalError as e:
if e[1] == 'Query was empty':
continue
raise e
fullLine = ''
db.close()
【讨论】：
另一个允许在不进行任何解析的情况下利用 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)
它避免了任何解析错误，例如，如果您有一个带有笑脸 ;-) 的字符串变量，或者如果您检查 ; 作为最后一个字符，如果您之后有 cmets，例如 SELECT * FROM foo_table; # selecting data
【讨论】：
system，这是一个阻塞调用。如果你觉得迂腐，可以使用Popen 和wait。
这里的许多答案都有严重的缺陷......
首先不要尝试自己解析开放式 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 中抛出这些异常之一。
【讨论】：
-p 提供密码时，MySQL 客户端本身总是对您大喊大叫。我发现很难想象您的操作系统将密码隐藏在您的进程列表中。这一定是 MySQL 现在内置的一些技巧。也许他们在“最新消息”文档中列出了这些内容？你用什么版本试过这个？
这对我有用：
with open('schema.sql') as f:
cursor.execute(f.read().decode('utf-8'), multi=True)
【讨论】：
decode 错误 - 删除它可以解决问题，效果很好，谢谢。
您可以使用不同的数据库驱动程序吗？
如果是：使用 MySQL 的 MySQL Connector/Python 驱动程序可以实现您想要的。
它的cursor.execute method支持通过Multi=True一次执行多条SQL语句。
不需要用分号分割文件中的 SQL 语句。
简单示例（主要是从第二个链接复制粘贴，我只是添加了从文件中读取SQL）：
import mysql.connector
file = open('test.sql')
sql = file.read()
cnx = mysql.connector.connect(user='uuu', password='ppp', host='hhh', database='ddd')
cursor = cnx.cursor()
for result in cursor.execute(sql, multi=True):
if result.with_rows:
print("Rows produced by statement '{}':".format(
result.statement))
print(result.fetchall())
else:
print("Number of rows affected by statement '{}': {}".format(
result.statement, result.rowcount))
cnx.close()
我正在使用它将 MySQL 转储（通过将整个数据库导出到 SQL 文件在 phpMyAdmin 中创建）从 *.sql 文件导入回数据库。
【讨论】：
使用pexpect library 怎么样？这个想法是，您可以启动一个进程pexpect.spawn(...)，并等到该进程的输出包含特定模式process.expect(pattern)。
其实我是用这个来连接mysql客户端，执行一些sql脚本的。
正在连接：
import pexpect
process = pexpect.spawn("mysql", ["-u", user, "-p"])
process.expect("Enter password")
process.sendline(password)
process.expect("mysql>")
这样密码不会硬编码到命令行参数中（消除安全风险）。
执行甚至几个sql脚本：
error = False
for script in sql_scripts:
process.sendline("source {};".format(script))
index = process.expect(["mysql>", "ERROR"])
# Error occurred, interrupt
if index == 1:
error = True
break
if not error:
# commit changes of the scripts
process.sendline("COMMIT;")
process.expect("mysql>")
print "Everything fine"
else:
# don't commit + print error message
print "Your scripts have errors"
请注意，您始终调用expect(pattern)，并且它匹配，否则您将收到超时错误。我需要这段代码来执行几个 sql 脚本，并且只有在没有发生错误时才提交它们的更改，但它很容易适应只有一个脚本的用例。
【讨论】：
下面还有一个 sqlite 示例（它也适用于 MySQL）。
示例文件.sql
INSERT INTO actors (name) VALUES ('Evan Goldberg')
INSERT INTO actors (name) VALUES ('Jay Baruchel')
INSERT INTO actors (name) VALUES ('Ray Downey')
示例 app.py
import sqlite3
from sqlite3 import OperationalError
#Connect to sqlite database
conn = sqlite3.connect('./database.db')
cur = conn.cursor()
#Create Table if not exist
create_table="""CREATE TABLE actors
( id INTEGER PRIMARY KEY AUTOINCREMENT,
name VARCHAR NOT NULL);"""
try:
execute = cur.execute(create_table)
except OperationalError as e:
print(e)
#Read File line by line and execute sql
with open("file.sql") as f:
query = f.readlines()
for sql in query:
execute = cur.execute(sql)
conn.commit()
#Check if data was submitted
execute = cur.execute('Select * From actors Order By id asc Limit 3')
rows = cur.fetchall()
print(rows)
conn.close()
你应该看到类似的结果：
[(1, 'Evan Goldberg'), (2, 'Jay Baruchel'), (3, 'Ray Downey')]
【讨论】：
正如其中一个 cmets 所述，如果您确定每个命令都以分号结尾，您可以这样做：
import mysql.connector
connection = mysql.connector.connect(
host=host,
user=user,
password=password
)
cursor = connection.cursor()
with open(script, encoding="utf-8") as f:
commands = f.read().split(';')
for command in commands:
cursor.execute(command)
print(command)
connection.close()
【讨论】：
你可以使用这样的东西-
def write_data(schema_name: str, table_name: str, column_names: str, data: list):
try:
data_list_template = ','.join(['%s'] * len(data))
insert_query = f"insert into {schema_name}.{table_name} ({column_names}) values {data_list_template}"
db.execute(insert_query, data)
conn_obj.commit()
except Exception as e:
db.execute("rollback")
raise e
【讨论】：