下面是一个Python脚本，使用PyMySQL库实现了连接远程数据库、执行SQL查询、格式化数据为CSV文件并处理潜在错误的需求。脚本支持通过命令行参数传入数据库账户信息文件路径和CSV保存路径。
先安装依赖库：
pip install pymysql
Python脚本代码
import pymysql
import csv
import argparse
import json
import sys
def load_db_credentials(file_path):
"""
Load database connection credentials from a given JSON file.
"""
try:
with open(file_path, 'r') as file:
credentials = json.load(file)
return credentials
except FileNotFoundError:
print(f"Error: Credential file '{file_path}' not found.")
sys.exit(1)
except json.JSONDecodeError:
print(f"Error: Unable to parse the JSON file '{file_path}'. Ensure it's valid.")
sys.exit(1)
def connect_to_database(credentials):
"""
Establish a connection to the database with error handling.
"""
try:
connection = pymysql.connect(
host=credentials['host'],
user=credentials['user'],
password=credentials['password'],
database=credentials['database'],
port=credentials.get('port', 3306), # Default port for MySQL is 3306
cursorclass=pymysql.cursors.DictCursor
)
return connection
except pymysql.MySQLError as e:
print(f"Database connection error: {e}")
sys.exit(1)
def execute_query(connection, query):
"""
Execute the given SQL query and fetch results.
"""
try:
with connection.cursor() as cursor:
cursor.execute(query)
results = cursor.fetchall()
return results
except pymysql.MySQLError as e:
print(f"SQL execution error: {e}")
sys.exit(1)
def save_to_csv(data, output_file):
"""
Save the query results to a CSV file.
"""
try:
if len(data) == 0:
print("No data returned to save as CSV file.")
sys.exit(0)
# Extract headers from the keys of the first dictionary in the data list
headers = data[0].keys()
with open(output_file, mode='w', newline='', encoding='utf-8') as file:
writer = csv.DictWriter(file, fieldnames=headers)
# Write header row
writer.writeheader()
# Write data rows
writer.writerows(data)
print(f"Data successfully saved to {output_file}")
except Exception as e:
print(f"Error while saving CSV: {e}")
sys.exit(1)
def main():
# Set up argument parser
parser = argparse.ArgumentParser(description="Connect to a database, execute an SQL query, and save results to a CSV file.")
parser.add_argument("credentials", help="Path to JSON file with database connection credentials.")
parser.add_argument("query", help="SQL query to execute.")
parser.add_argument("output", help="Path to save the CSV file.")
args = parser.parse_args()
# Load database credentials
credentials = load_db_credentials(args.credentials)
# Connect to the database
connection = connect_to_database(credentials)
# Execute the SQL query
results = execute_query(connection, args.query)
# Close the database connection
connection.close()
# Save results to CSV
save_to_csv(results, args.output)
if __name__ == "__main__":
main()
功能说明
-
脚本参数:
credentials: 数据库连接信息保存为一个JSON文件，包含如下格式：
{
"host": "your_host",
"user": "your_username",
"password": "your_password",
"database": "your_database",
"port": 3306
}
port参数是可选的（默认为3306）。query: 用于执行的SQL查询语句。output: CSV文件保存路径。
-
脚本步骤:
- 从用户指定的账户信息JSON文件加载数据库连接配置。
- 使用PyMySQL库连接到MySQL数据库。
- 执行SQL查询语句。
- 查询数据并保存为CSV文件。
-
错误处理:
- 捕获文件读取错误（如账户信息文件不存在或JSON语法错误）。
- 数据库连接错误（如用户名密码错误，主机不可达）。
- SQL语法错误或查询问题。
- CSV文件写入过程中的错误。
-
运行方式:
使用如下命令运行：
python script.py credentials.json "SELECT * FROM your_table WHERE condition" output.csv
-
可扩展性:
- 可自行修改JSON文件中的键名或增加更多数据库配置。
- 自定义数据过滤SQL语句。
- 扩展支持其他数据库类型（通过替换PyMySQL库）。
示例
假设有以下SQL查询：
SELECT id, name, email FROM users WHERE status = 'active';
命令行运行示例：
python script.py db_credentials.json "SELECT id, name, email FROM users WHERE status = 'active'" active_users.csv
结果：保存查询到的id, name, email数据于active_users.csv。
通过这份脚本，你可以快速实现远程数据库操作并生成CSV，非常适用于脚本化查询任务的场景。
