1. 项目概述用Python与MySQL构建数据桥梁作为一名和数据打了十几年交道的开发者我处理过无数种数据源但MySQL和Python的组合始终是我工具箱里最趁手、最可靠的“黄金搭档”。无论是快速搭建一个内容管理系统还是为复杂的业务分析提供数据支撑这套组合拳都能游刃有余。今天我们就来深入聊聊如何用Python实现对MySQL数据库最核心的四个操作查询、添加、修改和删除。这不仅仅是写几条SQL语句那么简单它关乎到代码的健壮性、性能的优化以及生产环境下的安全与稳定。如果你正在为课程设计寻找一个扎实的数据库操作案例或者在工作中需要快速上手一个数据驱动的Python应用那么这篇从一线实战中总结出来的经验或许能让你少走不少弯路。2. 环境准备与核心工具选型2.1 Python环境与MySQL驱动工欲善其事必先利其器。首先确保你有一个可用的Python环境。我个人推荐使用Python 3.8或更高版本它们在异步支持和类型提示上更加完善。安装Python本身很简单从官网下载安装包即可这里不再赘述。关键在于MySQL驱动也叫连接器的选择。在Python生态中mysql-connector-python和PyMySQL是两个最主流的选择。它们各有侧重mysql-connector-python这是MySQL官方提供的纯Python驱动。它的优点是“官方出品”理论上兼容性最好且完全遵循Python DB API 2.0规范。但有时在安装或某些特定功能上可能会遇到小麻烦。PyMySQL这是一个纯Python的第三方驱动同样遵循DB API 2.0规范。它的最大优点是安装极其简单pip install pymysql并且在大多数场景下表现稳定可靠社区活跃是我个人在大多数项目中的首选。对于新手和绝大多数应用场景我强烈建议从PyMySQL开始。它足够简单、稳定能让你快速把注意力集中在业务逻辑上而不是环境配置上。安装命令非常简单pip install pymysql2.2 数据库准备与连接配置在写代码之前我们需要一个目标数据库。假设你已经在本地或远程服务器上安装并运行了MySQL服务关于MySQL的安装与基础配置网上教程很多核心是记住root密码和服务端口通常是3306。我们创建一个用于演示的数据库和表-- 创建一个名为 demo_db 的数据库 CREATE DATABASE IF NOT EXISTS demo_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; -- 使用这个数据库 USE demo_db; -- 创建一个用户表 CREATE TABLE IF NOT EXISTS users ( id INT NOT NULL AUTO_INCREMENT COMMENT 用户ID主键, username VARCHAR(50) NOT NULL COMMENT 用户名, email VARCHAR(100) NOT NULL COMMENT 邮箱, age INT COMMENT 年龄, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT 创建时间, PRIMARY KEY (id), UNIQUE KEY uk_username (username), UNIQUE KEY uk_email (email) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT用户信息表;这个表结构包含了自增主键、唯一约束、默认时间戳等常见设计比较有代表性。接下来就是Python连接数据库的核心步骤。这里有一个非常重要的原则永远不要在代码中硬编码数据库密码等敏感信息。正确的做法是使用配置文件或环境变量。注意将数据库连接信息尤其是密码直接写在代码里并上传到版本控制系统如Git是极其危险的行为可能导致严重的数据泄露。一个安全的连接示例我们假设通过环境变量获取密码import pymysql import os from dotenv import load_dotenv # 需要安装 python-dotenv: pip install python-dotenv # 加载.env文件中的环境变量 load_dotenv() def get_connection(): 创建并返回一个数据库连接。 连接信息从环境变量读取确保安全。 try: connection pymysql.connect( hostos.getenv(DB_HOST, localhost), # 主机默认localhost portint(os.getenv(DB_PORT, 3306)), # 端口默认3306 useros.getenv(DB_USER, root), # 用户名 passwordos.getenv(DB_PASSWORD), # 密码必须从环境变量读取 databaseos.getenv(DB_NAME, demo_db), # 数据库名 charsetutf8mb4, # 字符集支持emoji等 cursorclasspymysql.cursors.DictCursor # 让查询结果以字典形式返回更易处理 ) print(数据库连接成功) return connection except pymysql.MySQLError as e: print(f连接数据库失败: {e}) return None # 在你的项目根目录创建一个 .env 文件内容如下不要提交到Git # DB_HOSTlocalhost # DB_PORT3306 # DB_USERyour_username # DB_PASSWORDyour_strong_password # DB_NAMEdemo_db这里我特意将连接逻辑封装成函数并使用了pymysql.cursors.DictCursor。这个设置会让查询返回的结果是一个个字典键是字段名值是字段值这比默认的元组格式直观得多后续处理数据也更方便。3. 核心操作一数据查询Read查询是数据库操作中最频繁、也最灵活的部分。PyMySQL执行查询的基本流程是建立连接 - 创建游标 - 执行SQL - 获取结果 - 关闭游标和连接。3.1 基础查询与结果处理让我们先从最简单的查询所有数据开始def fetch_all_users(): 查询所有用户信息 conn get_connection() if not conn: return try: with conn.cursor() as cursor: # 使用with语句管理游标自动关闭 sql SELECT id, username, email, age, created_at FROM users cursor.execute(sql) # 获取所有结果 results cursor.fetchall() for row in results: # 因为使用了DictCursorrow是一个字典 print(fID: {row[id]}, 用户名: {row[username]}, 邮箱: {row[email]}, 年龄: {row.get(age)}) except pymysql.MySQLError as e: print(f查询出错: {e}) finally: conn.close() # 确保连接被关闭 fetch_all_users()fetchall()会一次性获取所有结果行。对于小数据量没问题但如果表里有几万、几十万行数据这会瞬间耗尽内存。这时就需要使用fetchone()或fetchmany(size)。def fetch_users_in_batches(batch_size10): 分批查询用户适用于大数据量 conn get_connection() if not conn: return try: with conn.cursor() as cursor: sql SELECT id, username FROM users ORDER BY id cursor.execute(sql) while True: batch cursor.fetchmany(batch_size) # 一次获取batch_size条 if not batch: break print(f获取到 {len(batch)} 条记录:) for row in batch: print(row) # 这里可以加入处理每一批数据的逻辑 # process_batch(batch) except pymysql.MySQLError as e: print(f分批查询出错: {e}) finally: conn.close()3.2 条件查询与参数化重中之重直接拼接SQL字符串是万恶之源会导致SQL注入攻击。必须使用参数化查询。def find_user_by_name(username): 根据用户名查找用户安全的方式 conn get_connection() if not conn: return try: with conn.cursor() as cursor: # 使用 %s 作为占位符参数以元组形式传入 sql SELECT * FROM users WHERE username %s cursor.execute(sql, (username,)) # 注意(username,)是单元素元组 user cursor.fetchone() # 只期望一条结果 if user: print(f找到用户: {user}) else: print(f未找到用户名为 {username} 的用户) return user except pymysql.MySQLError as e: print(f条件查询出错: {e}) finally: conn.close() # 模糊查询示例 def search_users_by_keyword(keyword): 根据关键词模糊搜索用户名或邮箱 conn get_connection() if not conn: return try: with conn.cursor() as cursor: # LIKE 查询也需要参数化 search_pattern f%{keyword}% sql SELECT id, username, email FROM users WHERE username LIKE %s OR email LIKE %s ORDER BY created_at DESC LIMIT 20 cursor.execute(sql, (search_pattern, search_pattern)) users cursor.fetchall() print(f找到 {len(users)} 个相关用户:) for u in users: print(u) finally: conn.close()关键点cursor.execute(sql, params)中的params可以是元组或列表。PyMySQL的底层驱动会负责对参数进行正确的转义防止SQL注入。即使参数看起来无害也必须养成使用参数化查询的习惯。3.3 复杂查询连接、分组与排序在实际项目中单表查询往往不够。我们再来创建一个orders表并演示一个简单的连接查询。CREATE TABLE IF NOT EXISTS orders ( order_id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL COMMENT 关联用户ID, amount DECIMAL(10, 2) NOT NULL COMMENT 订单金额, status ENUM(pending, paid, shipped, completed) DEFAULT pending, order_date DATE, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE );def get_user_orders(): 查询用户及其订单信息内连接 conn get_connection() if not conn: return try: with conn.cursor() as cursor: sql SELECT u.username, u.email, o.order_id, o.amount, o.status, o.order_date FROM users u INNER JOIN orders o ON u.id o.user_id WHERE o.order_date CURDATE() - INTERVAL 30 DAY -- 最近30天的订单 ORDER BY o.order_date DESC, o.amount DESC cursor.execute(sql) orders cursor.fetchall() for order in orders: print(f用户[{order[username]}] 在 {order[order_date]} 有订单: ID{order[order_id]}, 金额{order[amount]}, 状态{order[status]}) except pymysql.MySQLError as e: print(f连接查询出错: {e}) finally: conn.close()4. 核心操作二数据添加Create向数据库插入新记录同样要严格使用参数化查询。4.1 单条插入与获取自增IDdef add_single_user(username, email, ageNone): 添加一个新用户 conn get_connection() if not conn: return None try: with conn.cursor() as cursor: sql INSERT INTO users (username, email, age) VALUES (%s, %s, %s) # 执行插入 cursor.execute(sql, (username, email, age)) # 提交事务对于INSERT/UPDATE/DELETE必须提交才会生效 conn.commit() print(f用户 {username} 添加成功。) # 获取刚刚插入记录的自增ID new_user_id cursor.lastrowid print(f新用户的ID是: {new_user_id}) return new_user_id except pymysql.IntegrityError as e: # 捕获唯一约束冲突等完整性错误 conn.rollback() # 回滚事务 if Duplicate entry in str(e): print(f添加失败用户名 {username} 或邮箱 {email} 已存在。) else: print(f添加失败数据完整性冲突: {e}) return None except pymysql.MySQLError as e: conn.rollback() # 出错时回滚 print(f添加用户出错: {e}) return None finally: conn.close() # 调用示例 # add_single_user(张三, zhangsanexample.com, 25) # add_single_user(李四, lisiexample.com) # age为NULL关键经验事务提交INSERT,UPDATE,DELETE操作后必须执行conn.commit()才能使更改永久化。错误处理与回滚在try块中如果发生异常必须在except块中执行conn.rollback()来回滚所有未提交的更改保持数据一致性。获取自增IDcursor.lastrowid是获取最后一次插入操作的自增主键值的最可靠方法。处理唯一约束错误通过捕获pymysql.IntegrityError并检查错误信息可以给用户更友好的提示。4.2 批量插入提升性能当需要插入大量数据时逐条插入的效率极低。应该使用executemany()方法。def add_multiple_users(users_list): 批量插入用户大幅提升性能 conn get_connection() if not conn: return False # 准备数据列表中的每个元素是一个元组对应SQL中的一组值 data [(user[username], user[email], user.get(age)) for user in users_list] if not data: print(数据列表为空。) return False try: with conn.cursor() as cursor: sql INSERT INTO users (username, email, age) VALUES (%s, %s, %s) # 使用executemany affected_rows cursor.executemany(sql, data) conn.commit() print(f批量插入成功影响了 {affected_rows} 行。) return True except pymysql.IntegrityError as e: conn.rollback() print(f批量插入失败可能存在重复数据或其他约束冲突: {e}) # 更精细的处理可以尝试将数据拆分成更小的批次重试或记录失败的行 return False except pymysql.MySQLError as e: conn.rollback() print(f批量插入出错: {e}) return False finally: conn.close() # 调用示例 # new_users [ # {username: 王五, email: wangwuexample.com, age: 30}, # {username: 赵六, email: zhaoliuexample.com, age: 22}, # {username: 孙七, email: sunqiexample.com} # age为None # ] # add_multiple_users(new_users)executemany()会将所有数据组合成一条高效的SQL语句或少量几条发送到数据库比循环执行execute()快几个数量级尤其是在网络延迟较高的情况下。5. 核心操作三数据修改Update与删除Delete5.1 更新数据更新操作需要谨慎务必带上WHERE条件否则会更新整张表def update_user_email(user_id, new_email): 更新用户的邮箱 conn get_connection() if not conn: return False try: with conn.cursor() as cursor: sql UPDATE users SET email %s WHERE id %s cursor.execute(sql, (new_email, user_id)) conn.commit() # 判断是否真的更新了数据 if cursor.rowcount 0: print(f成功更新了用户ID为 {user_id} 的邮箱。) return True else: # rowcount为0可能是提供的user_id不存在或者新旧邮箱相同 print(f未更新任何记录。请检查用户ID {user_id} 是否存在或新邮箱是否与原邮箱相同。) return False except pymysql.IntegrityError as e: conn.rollback() print(f更新失败新邮箱 {new_email} 可能已被其他用户使用: {e}) return False except pymysql.MySQLError as e: conn.rollback() print(f更新用户出错: {e}) return False finally: conn.close() def increment_user_age(user_id): 让用户的年龄增加1岁演示使用字段原值 conn get_connection() if not conn: return try: with conn.cursor() as cursor: # 使用 SQL 表达式更新 sql UPDATE users SET age age 1 WHERE id %s cursor.execute(sql, (user_id,)) conn.commit() if cursor.rowcount 0: print(f用户 {user_id} 的年龄已增加。) except pymysql.MySQLError as e: conn.rollback() print(f更新年龄出错: {e}) finally: conn.close()关键点cursor.rowcount属性返回受上一次execute()影响的行数。这对于判断UPDATE或DELETE操作是否真正生效非常有用。5.2 删除数据删除操作风险最高一定要有确认机制并且优先考虑“软删除”用一个字段标记为已删除而不是物理删除。def delete_user_by_id(user_id): 根据ID删除用户物理删除谨慎操作 # 在实际应用中这里应该有一个额外的确认比如弹出对话框或二次验证。 confirm input(f你确定要删除ID为 {user_id} 的用户吗此操作不可逆(输入 yes 确认): ) if confirm.lower() ! yes: print(操作已取消。) return False conn get_connection() if not conn: return False try: with conn.cursor() as cursor: # 先查询一下用户是否存在可选提供更好体验 cursor.execute(SELECT username FROM users WHERE id %s, (user_id,)) user cursor.fetchone() if not user: print(f用户ID {user_id} 不存在。) return False # 执行删除 sql DELETE FROM users WHERE id %s cursor.execute(sql, (user_id,)) conn.commit() if cursor.rowcount 0: print(f用户 {user[username]} (ID: {user_id}) 已被删除。) return True else: print(删除操作未影响任何行。) return False except pymysql.MySQLError as e: conn.rollback() print(f删除用户出错: {e}) # 如果表上有外键约束删除可能会失败 if foreign key constraint in str(e).lower(): print(删除失败该用户可能关联了其他数据如订单请先处理关联数据。) return False finally: conn.close()关于“软删除”的强烈建议在生产系统中我几乎从不使用物理DELETE。取而代之的是在表中增加一个is_deleted(TINYINT) 或deleted_at(TIMESTAMP) 字段。删除操作只是更新这个标记查询时默认加上WHERE is_deleted 0的条件。这保留了数据历史便于审计和恢复也避免了外键约束带来的麻烦。6. 实战进阶封装与上下文管理器每次都写try...except...finally来管理连接和游标太繁琐了。我们可以利用Python的上下文管理器with语句和面向对象思想进行优雅的封装。6.1 创建一个简单的数据库助手类import pymysql import os from contextlib import contextmanager from dotenv import load_dotenv load_dotenv() class DBHelper: 一个简单的数据库操作助手类 def __init__(self): self.db_config { host: os.getenv(DB_HOST, localhost), port: int(os.getenv(DB_PORT, 3306)), user: os.getenv(DB_USER, root), password: os.getenv(DB_PASSWORD), database: os.getenv(DB_NAME, demo_db), charset: utf8mb4, cursorclass: pymysql.cursors.DictCursor } contextmanager def get_cursor(self, commitFalse): 获取游标的上下文管理器。 :param commit: 是否在退出时自动提交事务针对写操作 conn pymysql.connect(**self.db_config) cursor None try: cursor conn.cursor() yield cursor # 将游标提供给with块内的代码使用 if commit: conn.commit() else: # 对于纯读操作显式rollback可以释放一些锁取决于事务隔离级别 conn.rollback() except Exception as e: conn.rollback() print(f数据库操作出错: {e}) raise e # 将异常向上抛出 finally: if cursor: cursor.close() conn.close() # 封装查询方法 def query_all(self, sql, paramsNone): 执行查询并返回所有结果 with self.get_cursor() as cursor: cursor.execute(sql, params or ()) return cursor.fetchall() def query_one(self, sql, paramsNone): 执行查询并返回第一条结果 with self.get_cursor() as cursor: cursor.execute(sql, params or ()) return cursor.fetchone() # 封装写操作方法 def execute_write(self, sql, paramsNone): 执行INSERT/UPDATE/DELETE操作返回影响的行数 with self.get_cursor(commitTrue) as cursor: cursor.execute(sql, params or ()) return cursor.rowcount def execute_many_write(self, sql, params_list): 批量执行写操作返回影响的行数 with self.get_cursor(commitTrue) as cursor: cursor.executemany(sql, params_list) return cursor.rowcount # 使用示例 if __name__ __main__: db DBHelper() # 查询变得非常简单 users db.query_all(SELECT * FROM users WHERE age %s, (20,)) for u in users: print(u[username]) # 插入也简化了 new_id_sql INSERT INTO users (username, email) VALUES (%s, %s) affected db.execute_write(new_id_sql, (测试用户, testexample.com)) print(f插入成功影响行数: {affected}) # 复杂的业务逻辑也可以在一个连接事务中完成 def transfer_points(from_user_id, to_user_id, points): 模拟一个转账业务需要原子性 db DBHelper() try: # 注意这里需要确保多个操作在同一个事务中。 # 更严谨的做法是让 get_cursor 支持嵌套或者使用连接池。 # 这里为简化假设get_cursor(commitTrue)内的操作是原子的。 # 实际复杂事务应使用 start_transaction(), commit(), rollback() 手动控制。 pass # 业务逻辑省略 except Exception as e: print(f转账失败: {e})这个封装将重复的数据库连接、游标管理和错误处理逻辑隐藏起来让业务代码更加清晰。contextmanager装饰器让我们能自定义一个支持with语句的游标管理器。6.2 连接池的考量对于Web应用或高并发场景频繁创建和销毁数据库连接开销很大。这时就需要使用连接池。DBUtils或SQLAlchemy等库提供了成熟的连接池实现。例如使用PyMySQL配合DBUtilspip install DBUtilsfrom dbutils.pooled_db import PooledDB import pymysql # 创建连接池 pool PooledDB( creatorpymysql, # 使用PyMySQL作为底层驱动 maxconnections10, # 池中最大连接数 mincached2, # 初始化时创建的空闲连接 maxcached5, # 池中空闲连接的最大数 blockingTrue, # 连接池满时是否阻塞等待 **db_config # 传入之前的连接配置字典 ) # 从池中获取连接 conn pool.connection() try: with conn.cursor() as cursor: cursor.execute(SELECT 1) result cursor.fetchone() print(result) finally: conn.close() # 注意这里不是真正关闭而是将连接归还给池连接池能显著提升高并发下的数据库访问性能是生产级应用的标配。7. 常见问题、性能优化与安全备忘在实际开发中你会遇到各种各样的问题。下面是我总结的一些典型场景和解决方案。7.1 高频问题速查表问题现象可能原因解决方案与排查步骤pymysql.err.OperationalError: (2003, “Can’t connect to MySQL server”)1. MySQL服务未运行。2. 主机名、端口错误。3. 防火墙阻止了连接。4. 用户无权从该主机连接。1. 检查MySQL服务状态 (sudo systemctl status mysql)。2. 确认连接参数host, port。3. 检查防火墙规则如3306端口。4. 在MySQL中执行GRANT ALL PRIVILEGES ON *.* TO ‘user’’%’(生产环境需限制IP)。pymysql.err.ProgrammingError: (1064, “You have an error in your SQL syntax”)SQL语句语法错误。1. 将打印出的SQL语句复制到MySQL客户端如Navicat, MySQL Workbench中直接执行看具体报错。2. 检查引号、括号是否匹配关键字是否拼写正确。3.特别注意参数化查询时占位符是%s不要写成?或其他。pymysql.err.IntegrityError: (1062, “Duplicate entry ‘xxx’ for key ‘uk_username’”)违反了唯一约束插入了重复的值如用户名、邮箱。1. 插入前先查询是否存在。2. 使用INSERT IGNORE或ON DUPLICATE KEY UPDATE语法。3. 在业务逻辑层做好数据校验。pymysql.err.InternalError: (1054, “Unknown column ‘xxx’ in ‘field list’”)查询或插入的字段在表中不存在。1. 检查表结构确认字段名拼写、大小写MySQL在Linux下默认区分大小写。2. 使用DESC table_name;命令查看表结构。查询结果中文乱码数据库、连接、终端的字符集不统一。1. 确保建表时使用utf8mb4。2. 确保Python连接参数设置charset’utf8mb4’。3. 确保Python文件本身也是UTF-8编码。cursor.fetchall()内存溢出查询结果集太大一次性加载到内存。1. 使用cursor.fetchone()逐行处理。2. 使用cursor.fetchmany(size)分批处理。3. 优化SQL增加LIMIT子句或更精确的WHERE条件。执行UPDATE或DELETE后rowcount为0WHERE 条件不匹配任何行或新旧数据相同。1. 确认传入的条件参数是否正确。2. 先执行一个SELECT查询确认目标数据存在。3. 检查程序逻辑确保不是重复执行了相同数据的更新。7.2 性能优化小贴士使用连接池如前所述对于Web应用必不可少。合理使用索引在WHERE,ORDER BY,GROUP BY涉及的列上建立索引能极大提升查询速度。可以使用EXPLAIN命令分析SQL执行计划。批量操作如前文的executemany()对于大量数据插入/更新批量操作比循环单条操作快得多。只查询需要的列避免使用SELECT *明确列出需要的字段名减少网络传输和内存消耗。关闭游标和连接使用with语句或确保在finally块中关闭避免连接泄漏。注意事务范围将多个相关的写操作放在一个事务中可以减少提交次数提升性能。但事务也不宜过大过长否则会锁定过多资源。7.3 安全红线SQL注入防御再次强调永远使用参数化查询cursor.execute(sql, params)绝对不要用字符串格式化如f”SELECT * FROM users WHERE name ‘{name}”或字符串拼接来构建SQL。密码等敏感信息必须通过环境变量或配置文件管理绝不能写在代码里。最小权限原则为应用数据库用户分配最小必要的权限如只读、只写特定表而不是直接使用root账户。验证输入即使在参数化查询防止了SQL注入也应在业务逻辑层对用户输入进行验证如邮箱格式、用户名长度等。走到这里你已经掌握了使用Python操作MySQL数据库从基础到进阶的绝大部分核心技能。从环境搭建、驱动选择到安全的CRUD操作、高效的批量处理再到生产级别的封装、连接池管理和避坑指南这些内容都来源于我多年实战中积累的经验和教训。数据库操作是后端开发的基石写出高效、安全、健壮的数据库访问代码是每个合格开发者的必修课。记住多思考“为什么这么做”多动手实践遇到问题善用搜索和官方文档你会越来越得心应手。如果在实际操作中遇到了上面没覆盖的奇怪问题不妨先回头检查一下最基础的连接参数和SQL语法很多时候问题就藏在最不起眼的地方。
![[Android ] Ampere Pro v4.37.0-专业的 Android 设备充电+放电评测应用](http://pic.xiahunao.cn/yaotu/[Android ] Ampere Pro v4.37.0-专业的 Android 设备充电+放电评测应用)
![[具身智能-649]：个人电脑搭建 RTSP 服务完整方案（Windows / Ubuntu 双平台，适配 RDK X5 rtsp2display 调试）](http://pic.xiahunao.cn/yaotu/[具身智能-649]：个人电脑搭建 RTSP 服务完整方案（Windows / Ubuntu 双平台，适配 RDK X5 rtsp2display 调试）)
