Python 操作 MySQL 数据库从入门到精通

前言PythonMySQL 黄金技术栈Python 与 MySQL 的组合是 Web 后端、数据分析、自动化运维领域的核心技术栈。Python 简洁高效MySQL 开源稳定二者结合可快速搭建各类业务系统。本文精简核心内容聚焦实用技能避开冗余理论从零基础到企业级应用一站式掌握操作要点。第一章 环境搭建快速构建开发环境1.1 MySQL 安装与初始化1.1.1 快速安装主流系统系统安装命令初始化步骤Windows官网下载 MySQL 8.0 安装包选择 “Developer Default”1. 设置 root 密码2. 默认端口 33063. 启动服务Macbrew install mysql1.brew services start mysql2.mysql_secure_installation设密码Linux (Ubuntu)sudo apt update sudo apt install mysql-server -y1.sudo systemctl start mysql2. 执行初始化脚本设密码1.1.2 创建测试库表-- 创建数据库支持emoji CREATE DATABASE IF NOT EXISTS python_mysql_demo CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci; USE python_mysql_demo; -- 创建用户表核心测试表 CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY COMMENT 用户ID, username VARCHAR(50) NOT NULL UNIQUE COMMENT 用户名, password VARCHAR(100) NOT NULL COMMENT 加密密码, nickname VARCHAR(50) COMMENT 昵称, age INT DEFAULT 0 COMMENT 年龄, gender ENUM(男,女,未知) DEFAULT 未知 COMMENT 性别, email VARCHAR(100) UNIQUE COMMENT 邮箱, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT 创建时间 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT用户信息表; -- 插入测试数据 INSERT INTO users (username, password, nickname, age, gender, email) VALUES (zhangsan, MD5(123456), 张三, 25, 男, zhangsandemo.com), (lisi, MD5(123456), 李四, 28, 女, lisidemo.com);1.2 Python 驱动安装与选择主流驱动两种按需选择驱动特点安装命令适用场景mysql-connector-pythonOracle 官方稳定兼容pip install mysql-connector-python企业级项目、追求稳定性PyMySQL纯 Python 实现轻量便捷pip install pymysql小型项目、快速开发第二章 基础入门连接与核心流程2.1 数据库连接两种方式2.1.1 mysql-connector 连接推荐import mysql.connector from mysql.connector import Error def connect_mysql(): 连接MySQL官方驱动 config { host: localhost, user: root, password: 123456, database: python_mysql_demo, charset: utf8mb4, auth_plugin: mysql_native_password # 兼容MySQL8.0认证 } try: conn mysql.connector.connect(**config) if conn.is_connected(): print(✅ 连接成功) return conn except Error as e: print(f❌ 连接失败{e}) return None # 测试连接 conn connect_mysql() if conn: conn.close() # 用完关闭2.1.2 PyMySQL 连接替代import pymysql from pymysql import OperationalError def connect_pymysql(): 连接MySQLPyMySQL config { host: localhost, user: root, password: 123456, database: python_mysql_demo, charset: utf8mb4, cursorclass: pymysql.cursors.DictCursor # 字典游标可读性强 } try: conn pymysql.connect(**config) print(✅ PyMySQL连接成功) return conn except OperationalError as e: print(f❌ 连接失败{e}) return None # 测试连接 conn connect_pymysql() if conn: conn.close()2.2 游标与核心流程核心六步连接→创建游标→执行 SQL→处理结果→提交事务→关闭资源def core_demo(): 完整操作流程演示 conn connect_mysql() if not conn: return # 1. 创建字典游标推荐 cursor conn.cursor(dictionaryTrue) try: # 2. 执行SQL查询前2条用户数据 sql SELECT * FROM users LIMIT 2 cursor.execute(sql) # 3. 处理结果 results cursor.fetchall() for row in results: print(fID{row[id]}昵称{row[nickname]}) # 4. 提交事务增删改必须查询无需 # conn.commit() except Error as e: # 5. 异常回滚 conn.rollback() print(f❌ 操作失败{e}) finally: # 6. 关闭资源必做防连接泄漏 cursor.close() conn.close() print( 资源已释放) # 运行演示 core_demo()第三章 CRUD 实战核心操作全解3.1 插入数据单条 / 批量3.1.1 单条插入防注入核心原则必须用%s占位符禁止字符串拼接def insert_single(): 单条插入用户 conn connect_mysql() if not conn: return cursor conn.cursor() try: sql INSERT INTO users (username, password, nickname, age) VALUES (%s, %s, %s, %s) params (wangwu, MD5(123456), 王五, 30) cursor.execute(sql, params) conn.commit() # 必须提交 print(f✅ 插入成功新增ID{cursor.lastrowid}) except Error as e: conn.rollback() print(f❌ 插入失败{e}) finally: cursor.close() conn.close() # 测试 insert_single()3.1.2 批量插入高效def insert_batch(): 批量插入用户 conn connect_mysql() if not conn: return cursor conn.cursor() try: sql INSERT INTO users (username, password, nickname, age) VALUES (%s, %s, %s, %s) # 批量参数列表嵌套元组 params_list [ (zhaoliu, MD5(123456), 赵六, 22), (sunqi, MD5(123456), 孙七, 24) ] cursor.executemany(sql, params_list) conn.commit() print(f✅ 批量插入成功影响行数{cursor.rowcount}) except Error as e: conn.rollback() print(f❌ 批量插入失败{e}) finally: cursor.close() conn.close() # 测试 insert_batch()3.2 查询数据基础 / 分页 / 聚合3.2.1 基础查询def query_basic(): 基础查询条件/全表 conn connect_mysql() cursor conn.cursor(dictionaryTrue) # 1. 全表查询 cursor.execute(SELECT username, nickname FROM users) all_users cursor.fetchall() print( 所有用户, all_users) # 2. 条件查询年龄25 cursor.execute(SELECT * FROM users WHERE age %s, (25,)) filter_users cursor.fetchall() print( 年龄25的用户, filter_users) # 3. 模糊查询昵称含张 cursor.execute(SELECT * FROM users WHERE nickname LIKE %s, (%张%,)) like_users cursor.fetchall() print( 昵称含张的用户, like_users) cursor.close() conn.close() # 测试 query_basic()3.2.2 分页查询企业必备def query_page(page1, page_size2): 分页查询 conn connect_mysql() cursor conn.cursor(dictionaryTrue) offset (page - 1) * page_size # 计算偏移量 sql SELECT * FROM users LIMIT %s OFFSET %s cursor.execute(sql, (page_size, offset)) results cursor.fetchall() print(f 第{page}页每页{page_size}条) for row in results: print(row) cursor.close() conn.close() # 测试 query_page(page1)3.2.3 聚合查询统计数据def query_aggregate(): 用户统计查询 conn connect_mysql() cursor conn.cursor(dictionaryTrue) sql SELECT COUNT(*) AS total, AVG(age) AS avg_age, MAX(age) AS max_age FROM users cursor.execute(sql) stats cursor.fetchone() print( 统计信息) print(f总用户数{stats[total]}) print(f平均年龄{round(stats[avg_age], 1)}) print(f最大年龄{stats[max_age]}) cursor.close() conn.close() # 测试 query_aggregate()3.3 更新数据单条 / 批量3.3.1 单条更新def update_single(user_id, new_nickname): 更新用户昵称 conn connect_mysql() cursor conn.cursor() sql UPDATE users SET nickname %s WHERE id %s cursor.execute(sql, (new_nickname, user_id)) conn.commit() if cursor.rowcount 0: print(f✅ 更新成功影响行数{cursor.rowcount}) else: print(⚠️ 未找到该用户) cursor.close() conn.close() # 测试更新ID1的用户昵称 update_single(1, 张三_更新)3.3.2 批量更新def update_batch(age, new_age): 批量更新年龄 conn connect_mysql() cursor conn.cursor() sql UPDATE users SET age %s WHERE age %s cursor.execute(sql, (new_age, age)) conn.commit() print(f✅ 批量更新成功影响行数{cursor.rowcount}) cursor.close() conn.close() # 测试将25岁用户改为26岁 update_batch(25, 26)3.4 删除数据单条 / 条件3.4.1 单条删除def delete_single(user_id): 根据ID删除用户 conn connect_mysql() cursor conn.cursor() sql DELETE FROM users WHERE id %s cursor.execute(sql, (user_id,)) conn.commit() if cursor.rowcount 0: print(f✅ 删除成功) else: print(⚠️ 未找到该用户) cursor.close() conn.close() # 测试删除ID3的用户 delete_single(3)3.4.2 条件删除慎用def delete_condition(age): 删除年龄小于指定值的用户 conn connect_mysql() cursor conn.cursor() sql DELETE FROM users WHERE age %s cursor.execute(sql, (age,)) conn.commit() print(f✅ 条件删除成功影响行数{cursor.rowcount}) cursor.close() conn.close() # 测试删除年龄23的用户 delete_condition(23)第四章 事务管理保证数据安全4.1 事务核心概念事务是不可分割的操作序列要么全成功要么全回滚具备 ACID 特性原子性、一致性、隔离性、持久性。4.2 Python 事务实现模拟转账def transaction_transfer(from_user, to_user, amount): 模拟转账事务安全 conn connect_mysql() if not conn: return cursor conn.cursor() try: conn.autocommit False # 关闭自动提交开启事务 # 1. 扣款 sql1 UPDATE users SET balance balance - %s WHERE username %s cursor.execute(sql1, (amount, from_user)) # 2. 入账 sql2 UPDATE users SET balance balance %s WHERE username %s cursor.execute(sql2, (amount, to_user)) conn.commit() # 提交事务 print(✅ 转账成功) except Error as e: conn.rollback() # 回滚事务 print(f❌ 转账失败已回滚{e}) finally: conn.autocommit True # 恢复自动提交 cursor.close() conn.close() # 测试需先给users表添加balance字段ALTER TABLE users ADD COLUMN balance DECIMAL(10,2) DEFAULT 0.00; # transaction_transfer(zhangsan, lisi, 100)4.3 事务隔离级别快速配置MySQL 默认隔离级别为REPEATABLE READ可快速切换def set_isolation_level(): 设置事务隔离级别 conn connect_mysql() cursor conn.cursor() # 查看当前级别 cursor.execute(SELECT transaction_isolation) print(当前级别, cursor.fetchone()[0]) # 切换为READ COMMITTED平衡性能与一致性 cursor.execute(SET SESSION transaction_isolation READ-COMMITTED) print(✅ 已切换为READ COMMITTED) cursor.close() conn.close() # 测试 set_isolation_level()第五章 高级特性进阶开发要点5.1 连接池企业必备避免频繁创建 / 销毁连接提升并发性能from mysql.connector import pooling # 全局连接池 pool None def init_pool(): 初始化连接池 global pool config { host: localhost, user: root, password: 123456, database: python_mysql_demo, charset: utf8mb4, pool_name: python_pool, pool_size: 5 # 池大小5 } pool pooling.MySQLConnectionPool(**config) print(✅ 连接池初始化完成) def get_conn(): 从池获取连接 if not pool: init_pool() return pool.get_connection() # 使用连接池查询 def query_with_pool(): conn get_conn() cursor conn.cursor(dictionaryTrue) cursor.execute(SELECT username FROM users LIMIT 3) print(cursor.fetchall()) cursor.close() conn.close() # 归还连接到池非关闭 # 测试 query_with_pool()总结本文从环境搭建、基础 CRUD、事务管理、高级特性、性能优化到实战项目全面覆盖了 Python 操作 MySQL 的全流程知识体系。核心要点总结驱动选择优先 mysql-connector-python官方稳定备用 PyMySQL轻量便捷核心流程连接→游标→执行→处理→提交→关闭资源必须释放安全第一参数化查询防注入密码加密存储事务保证一致性效率提升批量操作、连接池、索引优化、ORM 框架实战落地封装工具类异常处理 日志记录适配企业开发