Python连接MySQL通常使用PyMySQL库，它通过提供接口实现数据增删改查，建立连接需安装库、配置参数、创建游标、执行SQL、提交事务并关闭连接；推荐使用环境变量或配置文件管理数据库凭证以提升安全性，避免硬编码；PyMySQL为纯Python实现，兼容Python 3且安装简便，相较MySQLdb更适用于现代项目；处理中文时应设置charset='utf8mb4'防止乱码，并利用DictCursor返回字典形式结果；常见问题如连接失败需检查网络、权限、用户名密码及端口，SQL错误则需验证语法、表名、列名和数据类型；事务管理需手动关闭自动提交，通过commit()提交或rollback()回滚以保证数据一致性；防范SQL注入必须使用参数化查询，禁止字符串拼接SQL。
Python连接MySQL数据库，通常我们会借助专门的数据库连接库，比如非常流行的
。它提供了一套接口，让你的Python程序能够像和本地对象交互一样，发送SQL指令、获取数据、管理事务，实现数据的增删改查。说白了，它就是Python和MySQL之间的一座桥梁，让你能用Python的语法去操作MySQL。
在Python项目里使用PyMySQL连接MySQL数据库，其实并没有想象中那么复杂。我们通常会分几步走：先是安装库，然后建立连接，接着创建游标来执行SQL语句，处理完数据后别忘了提交更改或者回滚，最后关闭连接。
首先，你得确保你的Python环境里安装了
。这通常通过
命令就能搞定：
安装好了之后，就可以开始写代码了。一个基本的连接和操作流程大概是这样：
立即学习
“
Python免费学习笔记（深入）
”；
这段代码展示了从连接到执行增删改查的基本流程。特别要注意的是
，这对于处理中文或其他多语言字符非常重要，可以有效避免乱码问题。另外，使用
这种上下文管理器的方式，可以确保游标在使用完毕后被正确关闭，这是一种良好的编程习惯。
PyMySQL与MySQLdb：Python数据库连接库的选择困境与我的看法
在Python连接MySQL的生态里，
和
是两个绕不开的名字。很多初学者可能都会纠结，到底用哪个好？我个人觉得，这其实是一个时代演进的问题。
是比较早期的、非常成熟的Python MySQL连接器，它底层是用C语言实现的，所以性能上确实有优势。但它的主要问题在于，它最初是为Python 2设计的，对Python 3的支持就显得有些力不从心，安装起来也常常需要编译C扩展，这在不同
操作系统
环境下可能会遇到各种依赖问题，挺让人头疼的。我记得以前在Windows上安装
，那简直就是一场与编译器的斗争。
而
呢，它是一个纯Python实现的库，这意味着它不依赖任何C语言扩展，安装起来就简单得多，直接
就行，跨平台兼容性非常好。更重要的是，它从一开始就是为Python 3设计的，完美支持Python 3的各种特性。虽然纯Python实现理论上性能会比C实现略低，但在大多数Web应用和日常数据操作场景下，这种差异几乎可以忽略不计。它的API设计也和DB-API 2.0规范保持一致，用起来很顺手。
所以，我的建议是，如果你正在开发一个全新的Python 3项目，或者打算将现有项目迁移到Python 3，那么毫不犹豫地选择
。它更现代、更易于安装和维护，社区也比较活跃。如果你的项目还在Python 2上跑，或者有非常苛刻的性能要求且能忍受
的安装复杂性，那可能才会考虑
。但就目前而言，
无疑是更主流、更推荐的选择。
安全连接MySQL：如何妥善管理Python中的数据库凭证？
将数据库的用户名、密码等敏感信息直接写在代码里（就像我上面那个示例里那样），在开发阶段图个方便可能还行，但一旦项目上线，这简直就是个安全大漏洞。想象一下，如果你的代码库被泄露，数据库的门就直接敞开了。所以，妥善管理数据库凭证是生产环境中非常重要的一环。
有几种比较常见的做法，可以有效提升安全性：
使用环境变量：
这是我最推荐的一种方式。你可以在部署服务器上设置环境变量，比如
、
、
等。Python程序启动时，通过
来获取这些值。这样，敏感信息就不会出现在代码仓库里，也方便在不同的部署环境（开发、测试、生产）中使用不同的配置。
在Linux/macOS中，你可以在终端这样设置：
配置文件：
可以使用
、
、
等格式的配置文件来存储数据库凭证。比如，创建一个
文件：
然后在Python代码中读取：
关键点在于：
这个
文件本身不应该被提交到公共的代码仓库中。你应该把它添加到
文件里。在部署时，手动将配置文件放到服务器上，或者通过部署工具注入。
通义灵码 windows版
通义灵码 0.8.0 是一款面向智能开发场景的轻量级集成开发环境。该版本聚焦效率与协作，提供代码智能补全、实时语法检查与版本管理功能，并优化了插件扩展能力，适配主流编程语言。0.8.0 版本进一步提升了启动速度与界面响应流畅度，内置任务运行器与调试工具，帮助开发者更快捷地完成项目构建与问题定位。简洁、现代的设计让初学者与专业团队都能快速上手。
下载
云服务商的秘密管理服务：
如果你的应用部署在云平台上（如AWS Secrets Manager, Azure Key Vault, Google Secret Manager），这些服务提供了更高级的凭证管理方案，可以动态获取、轮换凭证，安全性更高。这通常适用于大型、复杂的云原生应用。
无论你选择哪种方式，核心原则都是将敏感信息与代码分离，并且确保敏感信息本身在传输和存储过程中是安全的。
PyMySQL常见问题排查：连接失败？SQL报错？这里有解决思路
在使用PyMySQL连接MySQL数据库时，遇到各种错误是家常便饭。很多时候，错误信息本身就能提供不少线索，但有时也需要一些排查技巧。
连接失败 (或)
检查网络连接和主机地址：
确保你的Python应用能够访问到MySQL服务器的IP地址或域名。
命令可以初步测试网络连通性。如果MySQL服务器在远程，检查
防火墙
是否阻止了连接（通常MySQL默认端口是3306）。
检查MySQL服务是否运行：
确保MySQL服务本身正在运行。在Linux上，
或
。
检查端口：
确认
和
参数是否正确。默认是
，但有时会被修改。
检查用户名和密码：
这是最常见的错误之一。仔细核对
和
是否正确，注意大小写。
检查用户权限：
即使用户名密码正确，该用户可能没有从你的客户端IP地址连接的权限，或者没有访问特定数据库的权限。在MySQL里，你可以用
来授予权限，然后
。如果是
则表示任何IP。
检查数据库名称：
确认
参数拼写无误，且该数据库确实存在。
SQL语句执行错误 (或)
SQL语法错误 ()：
这是最直接的。错误信息通常会指出SQL语句的哪一部分有问题。仔细检查你的SQL语句，看看有没有拼写错误、缺少逗号、引号不匹配、关键词使用不当等。可以尝试在MySQL客户端（如
命令行工具或
）中直接执行这条SQL，看是否能复现错误，这样定位问题会更快。
表或列不存在 (或)：
确认你引用的表名和列名是正确的，并且数据库中确实存在。注意大小写，MySQL在某些操作系统上（如Linux）对表名是大小写敏感的。
数据类型不匹配或约束违规 (或)：
比如，你尝试将一个过长的字符串插入到
的列中，或者将非数字字符插入到
列。
通常是违反了唯一约束（
）、非空约束（
）或外键约束。例如，你插入了一个已经存在的唯一值。
字符集问题：
如果你看到乱码，或者在插入特定字符时报错，很可能是字符集设置不正确。确保数据库、表、列以及PyMySQL连接时的
都设置为
，并且客户端的终端也支持
。
调试技巧：
打印SQL语句和参数：
在执行
之前，把完整的SQL语句（包括替换后的参数）打印出来，这样可以直观地看到发送给数据库的到底是什么。
查看MySQL错误日志：
MySQL服务器本身会记录错误日志，这些日志通常包含更详细的错误信息，可以帮助你定位服务器端的问题。
逐步调试：
使用Python的调试器（如
或IDE的调试功能）逐步执行代码，检查变量的值，特别是SQL语句和传递的参数。
遇到问题不要慌，先看错误信息，它往往就是最好的老师。结合这些排查思路，大多数连接和SQL执行问题都能迎刃而解。
进阶应用：PyMySQL如何实现事务管理与防范SQL注入？
数据库操作，尤其是涉及多个步骤的数据修改，事务管理就显得尤为重要。同时，面对无处不在的网络攻击，SQL注入防范更是重中之重。PyMySQL在这两方面都提供了成熟的解决方案。
事务管理
事务（Transaction）是一组原子性的数据库操作。要么这些操作全部成功提交，要么全部失败回滚，不会出现部分成功的情况。这对于保持数据的一致性和完整性至关重要，比如银行转账，从一个账户扣钱，给另一个账户加钱，这两个操作必须同时成功或同时失败。
PyMySQL默认是开启自动提交（autocommit）的，这意味着你每执行一条SQL语句，数据库都会立即提交更改。但在需要事务管理的场景下，我们需要手动关闭自动提交，然后显式地进行提交或回滚。
在这个例子中，
是关键。它告诉PyMySQL，后续的SQL操作不会立即生效，直到你调用
。如果中间出现任何异常，
会撤销所有未提交的更改，确保数据的一致性。
防范SQL注入
SQL注入是一种常见的Web安全漏洞，攻击者通过在输入字段中插入恶意的SQL代码，来操纵数据库查询，可能导致数据泄露、篡改甚至删除。
PyMySQL，以及所有遵循DB-API 2.0规范的Python数据库连接库，都提供了强大的机制来防范SQL注入，那就是
参数化查询（Parameterized Queries）
。
错误的做法（容易被SQL注入）：
正确的做法（使用参数化查询）：
PyMySQLPyMySQLpippip install PyMySQLimport pymysql
# 数据库连接参数，这里建议不要硬编码在代码里，后面会提到更安全的做法
DB_CONFIG = {
'host': 'localhost', # 数据库服务器地址
'user': 'your_username', # 数据库用户名
'password': 'your_password', # 数据库密码
'database': 'your_database', # 要连接的数据库名称
'charset': 'utf8mb4', # 字符集，很重要，避免乱码
'cursorclass': pymysql.cursors.DictCursor # 让查询结果以字典形式返回，方便操作
}
connection = None # 初始化连接变量
try:
# 建立数据库连接
connection = pymysql.connect(**DB_CONFIG)
print("数据库连接成功！")
# 创建一个游标对象，用于执行SQL查询
# DictCursor 会让查询结果以字典形式返回，方便通过列名访问数据
with connection.cursor() as cursor:
# 示例：创建一个表（如果不存在）
create_table_sql = """
CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
email VARCHAR(255) UNIQUE NOT NULL
) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
"""
cursor.execute(create_table_sql)
print("表 'users' 检查或创建完成。")
# 示例：插入一条数据
insert_sql = "INSERT INTO users (name, email) VALUES (%s, %s)"
user_data = ('张三', 'zhangsan@example.com')
cursor.execute(insert_sql, user_data)
connection.commit() # 提交事务，保存更改
print(f"插入数据成功：{user_data}")
# 示例：查询所有数据
select_sql = "SELECT id, name, email FROM users"
cursor.execute(select_sql)
results = cursor.fetchall() # 获取所有查询结果
print("\n当前用户列表：")
if results:
for row in results:
print(f"ID: {row['id']}, 姓名: {row['name']}, 邮箱: {row['email']}")
else:
print("没有找到用户数据。")
# 示例：更新一条数据
update_sql = "UPDATE users SET name = %s WHERE email = %s"
update_data = ('李四', 'zhangsan@example.com')
cursor.execute(update_sql, update_data)
connection.commit()
print(f"更新数据成功：将 'zhangsan@example.com' 的姓名更新为 '李四'")
# 再次查询确认更新
cursor.execute(select_sql)
results_after_update = cursor.fetchall()
print("\n更新后的用户列表：")
for row in results_after_update:
print(f"ID: {row['id']}, 姓名: {row['name']}, 邮箱: {row['email']}")
except pymysql.Error as e:
print(f"数据库操作失败: {e}")
if connection:
connection.rollback() # 发生错误时回滚事务
print("事务已回滚。")
except Exception as e:
print(f"发生未知错误: {e}")
finally:
if connection:
connection.close() # 确保连接被关闭
print("数据库连接已关闭。")
charset='utf8mb4'with connection.cursor() as cursor:PyMySQLMySQLdbMySQLdbMySQLdbPyMySQLpip install PyMySQLPyMySQLMySQLdbMySQLdbPyMySQLMYSQL_HOSTMYSQL_USERMYSQL_PASSWORDos.getenv()import os
import pymysql
DB_CONFIG = {
'host': os.getenv('MYSQL_HOST', 'localhost'), # 默认值，防止环境变量未设置
'user': os.getenv('MYSQL_USER', 'root'),
'password': os.getenv('MYSQL_PASSWORD', ''),
'database': os.getenv('MYSQL_DATABASE', 'test_db'),
'charset': 'utf8mb4',
'cursorclass': pymysql.cursors.DictCursor
}
# ... 后续连接代码 ...export MYSQL_HOST='your_host'export MYSQL_USER='your_user'export MYSQL_PASSWORD='your_password'.ini.json.yamlconfig.ini[database]
host = localhost
user = your_username
password = your_password
database = your_databaseimport configparser
import os
import pymysql
config = configparser.ConfigParser()
# 尝试从当前目录或指定路径加载配置文件
config_path = os.path.join(os.path.dirname(__file__), 'config.ini')
if os.path.exists(config_path):
config.read(config_path)
else:
print("Warning: config.ini not found, falling back to environment variables or defaults.")
# 如果配置文件不存在，可以考虑从环境变量获取或使用默认值
DB_CONFIG = {
'host': config.get('database', 'host', fallback=os.getenv('MYSQL_HOST', 'localhost')),
'user': config.get('database', 'user', fallback=os.getenv('MYSQL_USER', 'root')),
'password': config.get('database', 'password', fallback=os.getenv('MYSQL_PASSWORD', '')),
'database': config.get('database', 'database', fallback=os.getenv('MYSQL_DATABASE', 'test_db')),
'charset': 'utf8mb4',
'cursorclass': pymysql.cursors.DictCursor
}
# ... 后续连接代码 ...config.ini.gitignoreOperationalError: (2003, "Can't connect to MySQL server on ...")(1045, "Access denied for user ...")ping your_mysql_hostsudo systemctl status mysqlsudo service mysql statushostport3306userpasswordGRANT ALL PRIVILEGES ON your_database.* TO 'your_username'@'your_client_ip' IDENTIFIED BY 'your_password';FLUSH PRIVILEGES;'%'databaseProgrammingErrorDataErrorProgrammingError: (1064, "You have an error in your SQL syntax...")mysqlMySQL WorkbenchProgrammingError: (1146, "Table 'your_database.your_table' doesn't exist")(1054, "Unknown column '...' in 'field list'")DataErrorIntegrityErrorVARCHAR(10)INTIntegrityErrorUNIQUENOT NULLcharsetutf8mb4utf8mb4cursor.execute()pdbimport pymysql
DB_CONFIG = {
'host': 'localhost',
'user': 'your_username',
'password': 'your_password',
'database': 'your_database',
'charset': 'utf8mb4',
'cursorclass': pymysql.cursors.DictCursor
}
connection = None
try:
connection = pymysql.connect(**DB_CONFIG)
# 关闭自动提交，手动管理事务
connection.autocommit(False)
print("数据库连接成功，自动提交已关闭。")
with connection.cursor() as cursor:
# 假设我们有一个 accounts 表，有 id 和 balance 字段
# 示例：创建 accounts 表
create_table_sql = """
CREATE TABLE IF NOT EXISTS accounts (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
balance DECIMAL(10, 2) NOT NULL DEFAULT 0.00
) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
"""
cursor.execute(create_table_sql)
# 确保有测试数据
cursor.execute("INSERT IGNORE INTO accounts (id, name, balance) VALUES (1, 'Alice', 1000.00)")
cursor.execute("INSERT IGNORE INTO accounts (id, name, balance) VALUES (2, 'Bob', 500.00)")
connection.commit() # 提交初始数据
print("\n开始转账事务：从 Alice 转 200 到 Bob")
# 1. Alice 账户扣除 200
update_alice_sql = "UPDATE accounts SET balance = balance - %s WHERE id = %s"
cursor.execute(update_alice_sql, (200.00, 1))
print("Alice 账户扣除 200。")
# 模拟一个可能失败的操作，比如 Bob 的账户不存在，或者余额不足等
# if True: # 假设这里有个条件判断，如果失败就抛异常
# raise ValueError("模拟转账失败！")
# 2. Bob 账户增加 200
update_bob_sql = "UPDATE accounts SET balance = balance + %s WHERE id = %s"
cursor.execute(update_bob_sql, (200.00, 2))
print("Bob 账户增加 200。")
# 如果所有操作都成功，则提交事务
connection.commit()
print("转账事务成功提交！")
except pymysql.Error as e:
print(f"数据库操作失败: {e}")
if connection:
connection.rollback() # 任何一步出错，都回滚所有操作
print("事务已回滚，数据恢复到转账前状态。")
except Exception as e:
print(f"发生未知错误: {e}")
if connection:
connection.rollback()
print("事务已回滚。")
finally:
if connection:
connection.close()
print("数据库连接已关闭。")
# 再次查询确认结果
try:
connection = pymysql.connect(**DB_CONFIG)
with connection.cursor() as cursor:
cursor.execute("SELECT id, name, balance FROM accounts WHERE id IN (1, 2)")
results = cursor.fetchall()
print("\n转账后的账户余额：")
for row in results:
print(f"ID: {row['id']}, 姓名: {row['name']}, 余额: {row['balance']}")
except pymysql.Error as e:
print(f"查询余额失败: {e}")
finally:
if connection:
connection.close()connection.autocommit(False)connection.commit()connection.rollback()# 千万不要这样做！
user_input = "'; DROP TABLE users; --" # 攻击者输入
sql = f"SELECT * FROM users WHERE name = '{user_input}'"
# 最终SQL会变成：SELECT * FROM users WHERE name = ''; DROP TABLE users; --'
# 这会导致你的 users 表被删除！
cursor.execute(sql)