【Python】PyMySQL 实战：从基础连接到高效数据操作

1. PyMySQL快速入门连接MySQL数据库第一次用PyMySQL操作数据库时我踩了不少坑。记得当时连最简单的连接语句都写错把port参数写成字符串导致报错。现在回头看这些经验反而成了最好的教学案例。PyMySQL是Python3中最流行的MySQL驱动库它用纯Python实现安装简单到只需一行命令pip install pymysql安装完成后我们可以用5行代码建立数据库连接import pymysql conn pymysql.connect( hostlocalhost, # 数据库IP本地可以用127.0.0.1 useryour_username, # 你的MySQL用户名 passwordyour_password, # 你的MySQL密码 databasetest_db, # 要连接的数据库名 port3306, # MySQL默认端口 charsetutf8mb4 # 推荐使用utf8mb4支持emoji )这里有几个新手常犯的错误忘记指定charset参数导致中文乱码port参数应该传整数而非字符串密码包含特殊字符时没加引号连接成功后我们需要创建游标对象来执行SQL语句cursor conn.cursor() # 创建普通游标 dict_cursor conn.cursor(pymysql.cursors.DictCursor) # 创建返回字典的游标实际项目中我强烈推荐使用DictCursor它会让返回结果以字典形式呈现通过字段名而非数字索引访问数据代码可读性大幅提升。验证连接是否成功的最佳方式是执行一个简单查询cursor.execute(SELECT VERSION()) version cursor.fetchone() print(fMySQL版本: {version[0]})最后别忘了关闭连接这是很多初学者会遗漏的重要步骤cursor.close() # 先关闭游标 conn.close() # 再关闭连接我见过太多因为未正确关闭连接导致数据库连接数耗尽的生产事故。在后续章节我们会介绍更安全的上下文管理器用法。2. CRUD操作实战增删改查全掌握数据库操作的核心就是CRUDCreate, Read, Update, Delete。让我用一个用户管理系统的例子带你完整走一遍这四种操作。2.1 创建数据表首先我们创建用户表包含id、用户名、邮箱和注册时间字段create_table_sql CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(50) NOT NULL UNIQUE, email VARCHAR(100) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 cursor.execute(create_table_sql) conn.commit() # DDL语句也需要commit这里有几个设计要点使用AUTO_INCREMENT让ID自增UNIQUE约束保证用户名唯一使用InnoDB引擎支持事务DEFAULT CURRENT_TIMESTAMP自动设置创建时间2.2 插入数据插入单条数据的标准写法insert_sql INSERT INTO users(username, email) VALUES(%s, %s) try: cursor.execute(insert_sql, (python_lover, pythonexample.com)) conn.commit() print(f新增用户ID: {cursor.lastrowid}) # 获取自增ID except pymysql.IntegrityError: print(用户名已存在) conn.rollback()批量插入数据时使用executemany效率更高users [ (user1, user1test.com), (user2, user2test.com), (user3, user3test.com) ] cursor.executemany(insert_sql, users) conn.commit()2.3 查询数据基础查询非常简单cursor.execute(SELECT id, username, email FROM users) for row in cursor.fetchall(): print(fID: {row[0]}, 用户名: {row[1]}, 邮箱: {row[2]})使用字典游标会让代码更清晰dict_cursor.execute(SELECT * FROM users WHERE username%s, (python_lover,)) user dict_cursor.fetchone() print(f用户邮箱: {user[email]})分页查询是常见需求page 1 page_size 10 offset (page - 1) * page_size cursor.execute( SELECT * FROM users ORDER BY id DESC LIMIT %s OFFSET %s, (page_size, offset) )2.4 更新与删除更新用户信息update_sql UPDATE users SET email%s WHERE username%s try: cursor.execute(update_sql, (new_emailtest.com, python_lover)) conn.commit() print(f影响行数: {cursor.rowcount}) except: conn.rollback()删除用户delete_sql DELETE FROM users WHERE username%s cursor.execute(delete_sql, (inactive_user,)) conn.commit()3. 高级技巧提升操作效率与安全性当数据量增大后基础CRUD操作的性能问题就会显现。下面分享几个我在实战中总结的优化技巧。3.1 批量操作提速executemany的底层其实是循环执行单条SQL对于大批量插入我们可以使用更高效的方式# 传统方式 - 慢 data [(fuser_{i}, fuser_{i}test.com) for i in range(1000)] cursor.executemany(insert_sql, data) # 优化方式 - 快 values ,.join([cursor.mogrify((%s,%s), row).decode(utf-8) for row in data]) cursor.execute(fINSERT INTO users(username,email) VALUES {values})实测万条数据插入时间从12秒降到0.8秒。原理是合并多条INSERT为一条SQL减少网络往返。3.2 事务处理转账场景是经典的事务案例def transfer_money(conn, from_user, to_user, amount): try: with conn.cursor() as cursor: # 检查余额 cursor.execute(SELECT balance FROM accounts WHERE user%s, (from_user,)) balance cursor.fetchone()[0] if balance amount: raise ValueError(余额不足) # 扣款 cursor.execute( UPDATE accounts SET balancebalance-%s WHERE user%s, (amount, from_user) ) # 入账 cursor.execute( UPDATE accounts SET balancebalance%s WHERE user%s, (amount, to_user) ) conn.commit() # 提交事务 except: conn.rollback() # 回滚事务 raise关键点使用with语句确保游标自动关闭所有操作成功后才commit任何异常触发rollback3.3 防SQL注入永远不要拼接SQL字符串# 危险可能被SQL注入 username input(用户名: ) cursor.execute(fSELECT * FROM users WHERE username{username}) # 安全方式 cursor.execute(SELECT * FROM users WHERE username%s, (username,))PyMySQL的参数化查询会自动处理特殊字符转义这是最基本的安全防护。4. 上下文管理器与连接池4.1 使用with自动管理资源手动管理连接容易遗忘关闭操作Python的上下文管理器是更好的选择class DB: def __init__(self, **kwargs): self.conn_kwargs kwargs def __enter__(self): self.conn pymysql.connect(**self.conn_kwargs) return self.conn.cursor(pymysql.cursors.DictCursor) def __exit__(self, exc_type, exc_val, exc_tb): self.conn.commit() if not exc_type else self.conn.rollback() self.conn.close() # 使用示例 with DB(hostlocalhost, userroot, databasetest) as cursor: cursor.execute(SELECT * FROM users) for row in cursor: print(row)这种写法确保在任何情况下连接都会被正确关闭事务也会根据是否发生异常自动提交或回滚。4.2 连接池优化性能Web应用中频繁创建关闭连接会消耗大量资源连接池是必备优化from dbutils.pooled_db import PooledDB pool PooledDB( creatorpymysql, maxconnections20, # 最大连接数 hostlocalhost, userroot, databasetest ) def get_users(): conn pool.connection() try: with conn.cursor() as cursor: cursor.execute(SELECT * FROM users) return cursor.fetchall() finally: conn.close()连接池的主要参数mincached: 初始空闲连接数maxcached: 最大空闲连接数maxshared: 最大共享连接数blocking: 当连接不足时是否阻塞等待5. 实战用户管理系统综合运用前面所学我们实现一个完整的用户管理系统import pymysql from datetime import datetime class UserManager: def __init__(self): self.conn pymysql.connect( hostlocalhost, userroot, databaseuser_system, cursorclasspymysql.cursors.DictCursor ) def add_user(self, username, email): with self.conn.cursor() as cursor: try: cursor.execute( INSERT INTO users(username, email) VALUES(%s, %s), (username, email) ) self.conn.commit() return cursor.lastrowid except pymysql.IntegrityError: self.conn.rollback() raise ValueError(用户名已存在) def get_user(self, user_id): with self.conn.cursor() as cursor: cursor.execute( SELECT id, username, email, created_at FROM users WHERE id%s, (user_id,) ) return cursor.fetchone() def list_users(self, page1, per_page10): with self.conn.cursor() as cursor: offset (page - 1) * per_page cursor.execute( SELECT * FROM users ORDER BY id DESC LIMIT %s OFFSET %s, (per_page, offset) ) return cursor.fetchall() def update_email(self, user_id, new_email): with self.conn.cursor() as cursor: cursor.execute( UPDATE users SET email%s WHERE id%s, (new_email, user_id) ) self.conn.commit() return cursor.rowcount def delete_user(self, user_id): with self.conn.cursor() as cursor: cursor.execute(DELETE FROM users WHERE id%s, (user_id,)) self.conn.commit() return cursor.rowcount def __del__(self): self.conn.close() # 使用示例 if __name__ __main__: manager UserManager() # 添加用户 user_id manager.add_user(test_user, testexample.com) # 查询用户 user manager.get_user(user_id) print(f新用户: {user[username]}) # 分页列表 print(用户列表:) for user in manager.list_users(): print(f{user[id]}: {user[username]}) # 更新邮箱 manager.update_email(user_id, new_emailtest.com) # 删除用户 manager.delete_user(user_id)这个案例涵盖了PyMySQL的大部分核心功能包括连接管理事务处理参数化查询分页查询异常处理在实际开发中你可以基于这个模板扩展更多功能如用户登录验证、密码加密、权限管理等。