Python使用PyMySQL操作MySQL完整指南1. 安装依赖pip install pymysql2. 基础配置和数据库操作2.1 基础配置类import pymysql from typing import List, Dict, Optional from datetime import datetime class MySQLDB: def __init__(self): self.conn None self.cursor None def connect(self): 连接数据库 try: self.conn pymysql.connect( hostlocalhost, port3306, userroot, password123456, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor # 返回字典格式 ) self.cursor self.conn.cursor() except Exception as e: print(f连接数据库失败{e}) raise e def close(self): 关闭数据库连接 if self.cursor: self.cursor.close() if self.conn: self.conn.close()2.2 数据库和表操作def create_database(self): 创建数据库 try: sql CREATE DATABASE IF NOT EXISTS test DEFAULT CHARACTER SET utf8mb4 self.cursor.execute(sql) self.conn.select_db(test) # 选择数据库 except Exception as e: print(f创建数据库失败{e}) raise e def drop_database(self): 删除数据库 try: sql DROP DATABASE IF EXISTS test self.cursor.execute(sql) except Exception as e: print(f删除数据库失败{e}) raise e def create_table(self): 创建用户表 try: sql CREATE TABLE IF NOT EXISTS users ( id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT 自增主键, username VARCHAR(50) NOT NULL UNIQUE COMMENT 用户名, password VARCHAR(100) NOT NULL COMMENT 密码, age INT COMMENT 年龄, create_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT 创建时间, INDEX idx_username (username) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT用户表; self.cursor.execute(sql) self.conn.commit() except Exception as e: self.conn.rollback() print(f创建表失败{e}) raise e3. CRUD操作3.1 插入操作def insert_user(self, username: str, password: str, age: int) - int: 插入单个用户 try: sql INSERT INTO users(username, password, age) VALUES (%s, %s, %s) self.cursor.execute(sql, (username, password, age)) self.conn.commit() return self.cursor.lastrowid # 返回自增ID except Exception as e: self.conn.rollback() print(f插入用户失败{e}) raise e def batch_insert_users(self, users: List[Dict]): 批量插入用户 try: sql INSERT INTO users(username, password, age) VALUES (%s, %s, %s) values [(user[username], user[password], user[age]) for user in users] self.cursor.executemany(sql, values) self.conn.commit() except Exception as e: self.conn.rollback() print(f批量插入用户失败{e}) raise e3.2 查询操作def get_user_by_id(self, user_id: int) - Optional[Dict]: 根据ID查询用户 try: sql SELECT * FROM users WHERE id %s self.cursor.execute(sql, (user_id,)) return self.cursor.fetchone() except Exception as e: print(f查询用户失败{e}) raise e def get_users_by_age(self, min_age: int) - List[Dict]: 查询大于指定年龄的用户 try: sql SELECT * FROM users WHERE age %s self.cursor.execute(sql, (min_age,)) return self.cursor.fetchall() except Exception as e: print(f查询用户失败{e}) raise e def get_users_with_pagination(self, page: int, page_size: int) - List[Dict]: 分页查询用户 try: offset (page - 1) * page_size sql SELECT * FROM users LIMIT %s OFFSET %s self.cursor.execute(sql, (page_size, offset)) return self.cursor.fetchall() except Exception as e: print(f分页查询失败{e}) raise e3.3 更新操作def update_user(self, user_id: int, password: str None, age: int None) - bool: 更新用户信息 try: updates [] values [] if password is not None: updates.append(password %s) values.append(password) if age is not None: updates.append(age %s) values.append(age) if not updates: return False values.append(user_id) sql fUPDATE users SET {, .join(updates)} WHERE id %s result self.cursor.execute(sql, tuple(values)) self.conn.commit() return result 0 except Exception as e: self.conn.rollback() print(f更新用户失败{e}) raise e3.4 删除操作def delete_user(self, user_id: int) - bool: 删除用户 try: sql DELETE FROM users WHERE id %s result self.cursor.execute(sql, (user_id,)) self.conn.commit() return result 0 except Exception as e: self.conn.rollback() print(f删除用户失败{e}) raise e4. 使用示例def main(): # 创建数据库实例 db MySQLDB() try: # 连接数据库 db.connect() # 创建数据库和表 db.create_database() db.create_table() # 插入单个用户 user_id db.insert_user(张三, 123456, 25) print(f插入用户ID: {user_id}) # 批量插入用户 users [ {username: 李四, password: 123456, age: 26}, {username: 王五, password: 123456, age: 27} ] db.batch_insert_users(users) # 查询用户 user db.get_user_by_id(user_id) print(f查询到的用户: {user}) # 更新用户 db.update_user(user_id, age30) # 分页查询 users db.get_users_with_pagination(1, 10) print(f分页查询结果: {users}) # 删除用户 db.delete_user(user_id) except Exception as e: print(f操作失败{e}) finally: db.close() if __name__ __main__: main()5. 事务处理示例def transfer_money(self, from_id: int, to_id: int, amount: float): 转账操作示例 try: # 开始事务 self.conn.begin() # 扣除转出账户金额 sql1 UPDATE accounts SET balance balance - %s WHERE id %s AND balance %s result1 self.cursor.execute(sql1, (amount, from_id, amount)) if result1 0: raise Exception(余额不足或账户不存在) # 增加转入账户金额 sql2 UPDATE accounts SET balance balance %s WHERE id %s result2 self.cursor.execute(sql2, (amount, to_id)) if result2 0: raise Exception(转入账户不存在) # 提交事务 self.conn.commit() return True except Exception as e: # 回滚事务 self.conn.rollback() print(f转账失败{e}) raise e6. 注意事项连接管理及时关闭数据库连接使用连接池管理连接可以使用 DBUtils 等库处理连接超时情况安全性使用参数化查询防止 SQL 注入妥善保管数据库凭证最小权限原则性能优化合理使用索引批量操作代替循环单条操作避免查询无用字段错误处理完善的异常处理机制事务回滚日志记录