如何使用Python恢复MySQL数据库？加载SQL备份文件至备用MySQL库时遇‘str’对象不可调用错误的排查

我有一个数据库备份生成的SQL文件，想要把里面的内容加载到备用MySQL数据库中。我写了一个Python函数来完成这个操作，但执行时出现了错误：'str' object is not callable。以下是我的脚本：

def load_database_dump_to_secondary_mysql(file_path='db_backup_file.sql'):
    query = f'source {file_path}'
    try:
        connection = mysql_hook.get_conn() # connection to secondary db
        cursor = connection.cursor(query)
        print('LOAD TO MYSQL COMPLETE')
    except Exception as xerror:
        print("LOAD ERROR: ", xerror)

注：mysql_hook是Airflow的连接器，包含备用数据库的连接信息（主机、用户名/密码、数据库名）。另外我无法连接主数据库，只能拿到SQL备份文件。请问我哪里做错了？

这个错误主要来自两个关键问题，我来一步步帮你解决：

1. 错误的cursor()用法导致类型错误 #

你这里写了cursor = connection.cursor(query)，但**cursor()方法并不接受查询字符串作为参数**。正确的用法是先创建游标对象，再调用cursor.execute()执行查询。不过更重要的是，你想用的source命令根本不能这么执行——这是MySQL命令行客户端的内置命令，不是标准SQL语句，Python的数据库驱动（比如pymysql、mysql-connector）根本不认识它。

2. 正确的执行方式：读取SQL文件内容并执行 #

因为source无法通过DBAPI执行，你需要直接读取SQL备份文件的内容，然后将内容作为SQL语句执行。结合你用的Airflow MySqlHook，这里有两种靠谱的方案：

方案一：用Airflow Hook的run方法（适合普通SQL备份） #

如果你的备份文件只包含常规的DDL/DML语句（没有存储过程、触发器这类需要修改分隔符的内容），直接用MySqlHook的run方法即可，它会自动处理连接和事务：

from airflow.providers.mysql.hooks.mysql import MySqlHook

def load_database_dump_to_secondary_mysql(file_path='db_backup_file.sql'):
    try:
        # 替换成你的备用数据库连接ID
        mysql_hook = MySqlHook(mysql_conn_id='secondary_mysql_conn')
        # 读取SQL文件内容
        with open(file_path, 'r', encoding='utf-8') as f:
            sql_content = f.read()
        
        # 执行SQL内容，autocommit=True确保所有语句都提交
        mysql_hook.run(sql_content, autocommit=True)
        print('LOAD TO MYSQL COMPLETE')
    except Exception as xerror:
        print("LOAD ERROR: ", xerror)

方案二：调用MySQL命令行客户端（适合含存储过程/触发器的备份） #

如果你的备份文件里有DELIMITER命令（比如存储过程、函数），Python的DBAPI驱动无法正确处理这种自定义分隔符，这时候直接调用MySQL命令行客户端是更可靠的方式：

import subprocess
from airflow.providers.mysql.hooks.mysql import MySqlHook

def load_database_dump_to_secondary_mysql(file_path='db_backup_file.sql'):
    try:
        mysql_hook = MySqlHook(mysql_conn_id='secondary_mysql_conn')
        # 从Hook获取连接配置
        conn = mysql_hook.get_connection('secondary_mysql_conn')
        
        # 构造MySQL客户端命令
        cmd = [
            'mysql',
            '-h', conn.host,
            '-u', conn.login,
            f'-p{conn.password}',
            conn.schema
        ]
        
        # 执行命令并传入SQL文件内容
        with open(file_path, 'r', encoding='utf-8') as f:
            subprocess.run(cmd, stdin=f, check=True)
        
        print('LOAD TO MYSQL COMPLETE')
    except subprocess.CalledProcessError as xerror:
        print("LOAD ERROR: 命令执行失败", xerror)
    except Exception as xerror:
        print("LOAD ERROR: ", xerror)

总结你遗漏的点 #

• 混淆了MySQL命令行命令和标准SQL：source是客户端内置命令，不能通过Python数据库连接执行
• 错误地给cursor()方法传入了查询参数，导致了'str' object is not callable的类型错误
• 没有正确读取SQL文件内容并执行其中的语句

内容的提问来源于stack exchange，提问作者Shadow Walker