尧图网页设计 
简介一份面向高校Python课程教学的课件包聚焦Python操作数据库内容涵盖数据库基础概念、常见SQL语句、Python连接与操作数据库的核心API并通过综合案例完整演示SQLite的建表、增删改查与连接管理流程。课件以1个PPTX演示文档呈现整体大小约3.46MB方便教师课堂演示与学生课后对照复习。目前已有853人学习浏览尤其适合期末复习时快速梳理数据库操作知识体系。课件采用进阶篇视角从SQLite轻量级数据库出发逐步深入到结构化查询语言和Python操作数据库的实战细节包含建表、查询、条件过滤等典型代码示例可帮助读者理解关系型数据库核心概念掌握使用sqlite3模块进行数据操作的完整方法。压缩包文件结构简洁直接打开即可使用适合高校教师直接用于备课或学生考前背诵要点。1. Python 操作数据库第一步是选对驱动和连接方式很多人写 Python 操作数据库上来就在代码里拼 SQL结果连最基本的连接都没管好。Python 之所以适合做数据清洗、报表、自动化任务不是因为它能写出多复杂的 SQL而是因为它把“连接、游标、事务”这些数据库交互细节收敛成了一组统一 API也就是 DB-API 规范。无论是内置的 sqlite3、PyMySQL、psycopg2 还是 pymssql调用方式都长得差不多差别主要在驱动包、占位符和连接参数上。这篇文章以 sqlite3 和 PyMySQL 两条线讲解覆盖本地开发和线上 MySQL 场景。课程设计里常见的北风数据库通常通过 ODBC 或 SQL Server 访问但操作姿势和这里说的完全一致。新手可以先照着第二章的最小骨架跑通查询老手可以直接跳到第四章看连接池和批量写入的坑。比较反直觉的一点是Python 操作数据库的性能瓶颈大多不在 SQL 本身而在连接的创建和释放、参数传递方式以及事务提交时机。2. 用 sqlite3 和 PyMySQL 跑通连接-游标-查询的最小骨架2.1 先分清驱动 API内置 sqlite3 与第三方 PyMySQL 的差异sqlite3 是 Python 标准库的一部分不需要额外安装适合本地验证、单元测试和小型工具。PyMySQL 是纯 Python 实现的 MySQL 客户端可以兼容 MySQL 5.7、8.0 以及 MariaDB在无法安装 C 扩展的环境里特别实用。两者都遵循 DB-API 2.0 规范核心对象是 Connection 和 Cursor但占位符不同sqlite3 使用?PyMySQL 使用%s。驱动适用数据库参数占位符默认事务行为sqlite3SQLite 文件数据库?自动开启事务需要显式 commitPyMySQLMySQL / MariaDB%sautocommitFalse需要显式 commitpsycopg2PostgreSQL%sautocommitFalse新手最容易在这两个地方犯迷糊第一把 MySQL 的%s语法套到 sqlite3 上运行时报ValueError: unsupported format character第二以为commit()可有可无结果数据只写在当前连接里换一个连接查询就看不到。2.2 建立连接时容易被忽略的 4 个参数连接不是简单的 host、port、user、password。生产环境里连接超时、字符集、游标类型和自动提交这四项才决定后面好不好排查问题。下面这段以 PyMySQL 为例import os import pymysql conn pymysql.connect( host127.0.0.1, portint(os.getenv(DB_PORT, 3306)), userapp_user, passwordyour_password, databasenorthwind, charsetutf8mb4, connect_timeout5, cursorclasspymysql.cursors.DictCursor, autocommitFalse, ) cursor conn.cursor()参数说明database是库名不是dbMySQL 驱动里如果传错会直接报TypeError。charsetutf8mb4用来存储 emoji 和生僻字只写utf8在 MySQL 8.0 下也能用但不推荐。connect_timeout5让连接失败时快速返回否则默认可能卡住几十秒。cursorclassDictCursor让查询结果变成字典row[customer_id]比row[0]可读性好很多。sqlite3 的写法更简单但timeout10要记住它控制的是数据库文件被其他进程锁定时的等待时间import sqlite3 conn sqlite3.connect(northwind.db, timeout10) conn.row_factory sqlite3.Row cursor conn.cursor()row_factory sqlite3.Row之后行对象既支持下标访问也支持字段名访问接近 DictCursor 的效果。2.3 查询结果execute 之后 fetchone 和 fetchall 怎么选fetchall()会把所有结果一次性加载到内存里数据量小没问题但如果是百万行查询内存和网络都会承受压力。更稳妥的方式是fetchmany(size)分批次消费适合同步、导出和清洗任务。cursor.execute( SELECT customer_id, company_name FROM customers WHERE region %s, (华东,), ) while True: batch cursor.fetchmany(500) if not batch: break for row in batch: print(row[customer_id], row[company_name])代码逻辑是每次取 500 行处理完之后继续取下 500 行直到没有数据。这里的500是批次大小不是 SQL 里的LIMIT所以结果集仍然是完整的只是分块传输和消费。如果业务允许分段也可以直接改写 SQLLIMIT %s OFFSET %s但要注意大数据量下 OFFSET 越深越慢。2.4 参数化查询不要在 SQL 里用 f-string 拼参数写参数化查询不是“为了安全”这么简单。把参数交给驱动去转义数据库还能复用执行计划对频繁执行的语句有明显收益。下面的写法是错误的# 反例拼接 SQL运维看了会直接打回 cursor.execute(fSELECT * FROM customers WHERE customer_id {customer_id})正确写法是把 SQL 和参数分开def find_customer(conn, customer_id): with conn.cursor() as cur: cur.execute( SELECT * FROM customers WHERE customer_id %s, (customer_id,), ) return cur.fetchone()这里%s是占位符(customer_id,)是参数元组即使只有一个参数也要带逗号否则会被当成普通字符串而非可迭代对象。sqlite3 需要把%s换成?参数不变。另外要注意with conn.cursor()退出时只会关闭游标不会关闭连接with conn:在 PyMySQL 里也不等于自动提交事务控制要单独写。3. 增删改查与事务边界写进去不难难在写错了能撤回3.1 增删改查的标准姿势让 SQL 和参数分离CRUD 是所有数据库操作的骨架。以员工表为例插入一条记录后返回自增主键def insert_employee(conn, last_name, title): with conn.cursor() as cur: sql INSERT INTO employees (last_name, title) VALUES (%s, %s) cur.execute(sql, (last_name, title)) conn.commit() return cur.lastrowid更新和删除同样需要把条件放在参数里而不是拼进 SQLdef update_employee_title(conn, employee_id, new_title): with conn.cursor() as cur: cur.execute( UPDATE employees SET title %s WHERE employee_id %s, (new_title, employee_id), ) conn.commit() return cur.rowcount def delete_employee(conn, employee_id): with conn.cursor() as cur: cur.execute( DELETE FROM employees WHERE employee_id %s, (employee_id,), ) conn.commit() return cur.rowcountUPDATE和DELETE返回的rowcount表示实际影响的行数。如果条件没有匹配到任何记录返回值是 0这不一定是错误但需要业务层决定是继续还是告警。3.2 事务commit 和 rollback 该放在哪一层最常见的坏习惯是把commit()写进每一个数据操作函数。这样会导致一个问题一个业务动作涉及多个写操作时无法保证原子性。比如转账逻辑第一步扣款成功第二步加款失败如果各自提交资金就丢了。正确做法是把多个写操作放进同一个事务由最外层控制提交和回滚def transfer(conn, from_account, to_account, amount): try: with conn.cursor() as cur: cur.execute( UPDATE accounts SET balance balance - %s WHERE account_no %s, (amount, from_account), ) cur.execute( UPDATE accounts SET balance balance %s WHERE account_no %s, (amount, to_account), ) conn.commit() except Exception: conn.rollback() raise说明conn.rollback()会把从上一个commit()以来的所有未提交操作全部回滚所以这段代码中如果第二条 UPDATE 失败第一条也不会生效。这里的raise很关键它把原始异常继续向上抛调用方才能感知到失败原因。事务的粒度应该由业务决定而不是由数据访问函数决定。3.3 被忽略的 lastrowid 和 rowcount 细节lastrowid并不是在所有数据库、所有场景下都可靠。对于 sqlite3 和 MySQL 的自增主键插入成功后可以拿到生成的主键值。但如果表没有自增列或者执行的是 INSERT IGNORE返回值可能不准。rowcount也同样有坑MySQL 在默认配置下如果一条 UPDATE 更新前后值相同可能返回 0而不是匹配的行数。如果业务需要判断“是否存在这条记录”要用SELECT 1而不是依赖rowcount。def user_exists(conn, user_id): with conn.cursor() as cur: cur.execute(SELECT 1 FROM users WHERE id %s, (user_id,)) return cur.fetchone() is not None3.4 常见异常处理和可重试判断Python 数据库操作常见的异常有这几类异常含义处理方式IntegrityError主键冲突、唯一约束冲突通常是数据问题不建议重试OperationalError连接断开、表不存在、锁等待超时需要检查连接状态必要时重连ProgrammingErrorSQL 语法错误、参数个数不对属于代码缺陷直接修代码DataError数值超出范围、日期格式错误属于入参问题校验数据写重试时要区分可重试和不可重试。连接断开的OperationalError可以重建连接后重新执行一次主键冲突的IntegrityError重试没有意义应该落到“更新已存在记录”或“忽略冲突”的业务策略上。4. 连接池、executemany 和批量写入别让数据库被你的循环拖垮4.1 为什么课程作业也要连接池短任务里每次执行都新建连接性能问题还不明显但在 Web 服务、批量脚本和定时任务里频繁建连的代价很高。一个 MySQL 连接从建立到认证完成往往需要几十毫秒如果每执行一条 SQL 都重新连接整体的吞吐量立刻崩溃。更严重的是数据库服务端有连接数上限短连接风暴会直接打满max_connections导致其他应用也连不上。连接池的价值是把连接复用起来同时限制应用侧并发连接的数量。早期写课程设计时大家喜欢写一个全局conn pymysql.connect(...)然后在 SQLite 和 MySQL 之间来回切换这种全局连接在单线程下没问题一旦引入多线程连接状态就容易串掉。连接池比全局连接更合适因为它按需创建连接同时在归还时清理状态。4.2 用 dbutils 实现一个可复用的 MySQL 连接池常见做法是使用dbutils包新版本的导入路径是dbutils.pooled_db。安装命令pip install pymysql DBUtils连接池初始化示例import pymysql from dbutils.pooled_db import PooledDB db_pool PooledDB( creatorpymysql, mincached2, maxcached5, maxconnections10, blockingTrue, ping1, host127.0.0.1, port3306, userapp_user, passwordyour_password, databasenorthwind, charsetutf8mb4, ) def get_db_connection(): return db_pool.connection()参数说明mincached2池中至少保持 2 个空闲连接避免刚启动时频繁建连。maxcached5最多缓存 5 个空闲连接超过 5 个的可用连接会在用完后直接关闭。maxconnections10连接池能支持的最大连接数超过时会根据blocking决定等待还是抛异常。blockingTrue连接用完后不会立即报错而是等待其他线程归还连接。ping1从连接池取出连接时进行轻量探活避免把已断开的连接交给业务。具体探活频率与 DBUtils 版本有关生产环境建议根据版本查看源码注释确认。需要注意连接池里的连接是有状态的。如果某段代码执行了SET SESSION或USE语句连接归还后可能影响其他线程。稳妥的做法是在关键操作前后调用conn.rollback()或conn.cursor().execute(SET SESSION ...)恢复默认状态而不是寄希望于连接池自动清理。4.3 executemany 批处理插入以及 1000 条一提交的边界批量插入最简单的思路是写一个 for 循环逐条 execute但这样要频繁交互网络数据量稍微上来就很难看。PyMySQL 提供executemany方法可以在一次调用里传入多条参数。对于 INSERT 语句PyMySQL 会尝试把多条参数合并成单条多值 INSERT显著减少网络往返。data [(fuser_{i}, i) for i in range(10000)] batch_size 1000 with get_db_connection() as conn: with conn.cursor() as cur: for start in range(0, len(data), batch_size): cur.executemany( INSERT INTO users (username, score) VALUES (%s, %s), data[start:start batch_size], ) conn.commit()这里分批的原因有两个第一MySQL 客户端和服务端都有max_allowed_packet限制单次 INSERT 太大容易超过限制第二批次过大会让数据库在事务中持有大量锁影响并发读。1000是一个相对保守的数值如果插入字段多、单行体积大还要往下调整比如500甚至200。4.4 批量写入和事务全部成功还是部分保留批量任务里最纠结的问题是失败策略。如果 10000 条数据里第 5000 条是脏数据前面 4999 条要不要保留如果全量回滚代价太高如果保留又会产生半截数据。实际运维中可以根据业务容忍度选择“小事务”策略每批一个事务批量成功才提交批量失败只回滚当前批次。这样可以做到“批内原子批间隔离”日志里也能记录到哪个批次失败。实现起来不复杂for start in range(0, len(data), batch_size): batch data[start:start batch_size] try: with get_db_connection() as conn: with conn.cursor() as cur: cur.executemany(insert_sql, batch) conn.commit() except Exception as exc: logger.error(batch %s failed: %s, start, exc) raise如果是纯粹的导入任务MySQL 还有更高效的LOAD DATA LOCAL INFILE方案先把数据写到 CSV 再一次性导入但需要数据库端允许LOCAL INFILE权限控制要提前确认。5. 数据库操作完成之后对账、幂等导入与结果校验操作数据库不只在写入时提心吊胆写完之后的验证同样重要。一个低成本的对账方法是比较源表和目标表的行数以及关键字段的汇总值。def table_basic_stats(conn, table_name, key_column): with conn.cursor() as cur: cur.execute( fSELECT COUNT(*), COALESCE(SUM({key_column}), 0) FROM {table_name} ) return cur.fetchone()注意table_name和key_column不能通过参数占位符传入因为表名和字段名无法参数化。如果这是公共函数必须做白名单校验防止拼接进任意内容。对账时同时比较行数和SUM(id)一旦有任何不一致再通过EXCEPT找出差异数据。如果是需要反复执行的导入任务最好把 SQL 写成幂等形式。MySQL 下可以使用INSERT ... ON DUPLICATE KEY UPDATE重复执行同一批数据不会产生重复记录insert_sql INSERT INTO customers (customer_id, company_name, region) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE company_name VALUES(company_name), region VALUES(region) 这样跑第二次导入时遇到相同主键只会更新字段不会报IntegrityError。这个写法在 MySQL 8.0.20 之后VALUES()函数被标记为弃用但 MySQL 5.7 和 MariaDB 仍然普遍可用。如果目标库是 SQLite可以用INSERT OR REPLACE INTO但要注意它会先删除旧记录再插入新记录可能引发自增主键变化。最后一个技巧是校验“写入是否真的生效”。commit()成功只能说明数据库端接收了事务不能保证数据一定如预期。简单的做法是在写完数据后立刻用独立连接回查time.sleep(0.2) # 等待事务日志落盘 with get_db_connection() as conn: with conn.cursor() as cur: cur.execute(SELECT COUNT(*) FROM customers) total cur.fetchone()[COUNT(*)] assert total expected_count, fexpected {expected_count}, got {total}这条断言放在定时任务收尾处比人工看日志可靠得多。对账通过之后再做下游数据更新否则继续确认批次失败点并决定是否重跑。本文还有配套的精品资源点击获取 Python操作MySQL数据库：从驱动连接到连接池与事务实践
简介一份面向高校Python课程教学的课件包聚焦Python操作数据库内容涵盖数据库基础概念、常见SQL语句、Python连接与操作数据库的核心API并通过综合案例完整演示SQLite的建表、增删改查与连接管理流程。课件以1个PPTX演示文档呈现整体大小约3.46MB方便教师课堂演示与学生课后对照复习。目前已有853人学习浏览尤其适合期末复习时快速梳理数据库操作知识体系。课件采用进阶篇视角从SQLite轻量级数据库出发逐步深入到结构化查询语言和Python操作数据库的实战细节包含建表、查询、条件过滤等典型代码示例可帮助读者理解关系型数据库核心概念掌握使用sqlite3模块进行数据操作的完整方法。压缩包文件结构简洁直接打开即可使用适合高校教师直接用于备课或学生考前背诵要点。1. Python 操作数据库第一步是选对驱动和连接方式很多人写 Python 操作数据库上来就在代码里拼 SQL结果连最基本的连接都没管好。Python 之所以适合做数据清洗、报表、自动化任务不是因为它能写出多复杂的 SQL而是因为它把“连接、游标、事务”这些数据库交互细节收敛成了一组统一 API也就是 DB-API 规范。无论是内置的 sqlite3、PyMySQL、psycopg2 还是 pymssql调用方式都长得差不多差别主要在驱动包、占位符和连接参数上。这篇文章以 sqlite3 和 PyMySQL 两条线讲解覆盖本地开发和线上 MySQL 场景。课程设计里常见的北风数据库通常通过 ODBC 或 SQL Server 访问但操作姿势和这里说的完全一致。新手可以先照着第二章的最小骨架跑通查询老手可以直接跳到第四章看连接池和批量写入的坑。比较反直觉的一点是Python 操作数据库的性能瓶颈大多不在 SQL 本身而在连接的创建和释放、参数传递方式以及事务提交时机。2. 用 sqlite3 和 PyMySQL 跑通连接-游标-查询的最小骨架2.1 先分清驱动 API内置 sqlite3 与第三方 PyMySQL 的差异sqlite3 是 Python 标准库的一部分不需要额外安装适合本地验证、单元测试和小型工具。PyMySQL 是纯 Python 实现的 MySQL 客户端可以兼容 MySQL 5.7、8.0 以及 MariaDB在无法安装 C 扩展的环境里特别实用。两者都遵循 DB-API 2.0 规范核心对象是 Connection 和 Cursor但占位符不同sqlite3 使用?PyMySQL 使用%s。驱动适用数据库参数占位符默认事务行为sqlite3SQLite 文件数据库?自动开启事务需要显式 commitPyMySQLMySQL / MariaDB%sautocommitFalse需要显式 commitpsycopg2PostgreSQL%sautocommitFalse新手最容易在这两个地方犯迷糊第一把 MySQL 的%s语法套到 sqlite3 上运行时报ValueError: unsupported format character第二以为commit()可有可无结果数据只写在当前连接里换一个连接查询就看不到。2.2 建立连接时容易被忽略的 4 个参数连接不是简单的 host、port、user、password。生产环境里连接超时、字符集、游标类型和自动提交这四项才决定后面好不好排查问题。下面这段以 PyMySQL 为例import os import pymysql conn pymysql.connect( host127.0.0.1, portint(os.getenv(DB_PORT, 3306)), userapp_user, passwordyour_password, databasenorthwind, charsetutf8mb4, connect_timeout5, cursorclasspymysql.cursors.DictCursor, autocommitFalse, ) cursor conn.cursor()参数说明database是库名不是dbMySQL 驱动里如果传错会直接报TypeError。charsetutf8mb4用来存储 emoji 和生僻字只写utf8在 MySQL 8.0 下也能用但不推荐。connect_timeout5让连接失败时快速返回否则默认可能卡住几十秒。cursorclassDictCursor让查询结果变成字典row[customer_id]比row[0]可读性好很多。sqlite3 的写法更简单但timeout10要记住它控制的是数据库文件被其他进程锁定时的等待时间import sqlite3 conn sqlite3.connect(northwind.db, timeout10) conn.row_factory sqlite3.Row cursor conn.cursor()row_factory sqlite3.Row之后行对象既支持下标访问也支持字段名访问接近 DictCursor 的效果。2.3 查询结果execute 之后 fetchone 和 fetchall 怎么选fetchall()会把所有结果一次性加载到内存里数据量小没问题但如果是百万行查询内存和网络都会承受压力。更稳妥的方式是fetchmany(size)分批次消费适合同步、导出和清洗任务。cursor.execute( SELECT customer_id, company_name FROM customers WHERE region %s, (华东,), ) while True: batch cursor.fetchmany(500) if not batch: break for row in batch: print(row[customer_id], row[company_name])代码逻辑是每次取 500 行处理完之后继续取下 500 行直到没有数据。这里的500是批次大小不是 SQL 里的LIMIT所以结果集仍然是完整的只是分块传输和消费。如果业务允许分段也可以直接改写 SQLLIMIT %s OFFSET %s但要注意大数据量下 OFFSET 越深越慢。2.4 参数化查询不要在 SQL 里用 f-string 拼参数写参数化查询不是“为了安全”这么简单。把参数交给驱动去转义数据库还能复用执行计划对频繁执行的语句有明显收益。下面的写法是错误的# 反例拼接 SQL运维看了会直接打回 cursor.execute(fSELECT * FROM customers WHERE customer_id {customer_id})正确写法是把 SQL 和参数分开def find_customer(conn, customer_id): with conn.cursor() as cur: cur.execute( SELECT * FROM customers WHERE customer_id %s, (customer_id,), ) return cur.fetchone()这里%s是占位符(customer_id,)是参数元组即使只有一个参数也要带逗号否则会被当成普通字符串而非可迭代对象。sqlite3 需要把%s换成?参数不变。另外要注意with conn.cursor()退出时只会关闭游标不会关闭连接with conn:在 PyMySQL 里也不等于自动提交事务控制要单独写。3. 增删改查与事务边界写进去不难难在写错了能撤回3.1 增删改查的标准姿势让 SQL 和参数分离CRUD 是所有数据库操作的骨架。以员工表为例插入一条记录后返回自增主键def insert_employee(conn, last_name, title): with conn.cursor() as cur: sql INSERT INTO employees (last_name, title) VALUES (%s, %s) cur.execute(sql, (last_name, title)) conn.commit() return cur.lastrowid更新和删除同样需要把条件放在参数里而不是拼进 SQLdef update_employee_title(conn, employee_id, new_title): with conn.cursor() as cur: cur.execute( UPDATE employees SET title %s WHERE employee_id %s, (new_title, employee_id), ) conn.commit() return cur.rowcount def delete_employee(conn, employee_id): with conn.cursor() as cur: cur.execute( DELETE FROM employees WHERE employee_id %s, (employee_id,), ) conn.commit() return cur.rowcountUPDATE和DELETE返回的rowcount表示实际影响的行数。如果条件没有匹配到任何记录返回值是 0这不一定是错误但需要业务层决定是继续还是告警。3.2 事务commit 和 rollback 该放在哪一层最常见的坏习惯是把commit()写进每一个数据操作函数。这样会导致一个问题一个业务动作涉及多个写操作时无法保证原子性。比如转账逻辑第一步扣款成功第二步加款失败如果各自提交资金就丢了。正确做法是把多个写操作放进同一个事务由最外层控制提交和回滚def transfer(conn, from_account, to_account, amount): try: with conn.cursor() as cur: cur.execute( UPDATE accounts SET balance balance - %s WHERE account_no %s, (amount, from_account), ) cur.execute( UPDATE accounts SET balance balance %s WHERE account_no %s, (amount, to_account), ) conn.commit() except Exception: conn.rollback() raise说明conn.rollback()会把从上一个commit()以来的所有未提交操作全部回滚所以这段代码中如果第二条 UPDATE 失败第一条也不会生效。这里的raise很关键它把原始异常继续向上抛调用方才能感知到失败原因。事务的粒度应该由业务决定而不是由数据访问函数决定。3.3 被忽略的 lastrowid 和 rowcount 细节lastrowid并不是在所有数据库、所有场景下都可靠。对于 sqlite3 和 MySQL 的自增主键插入成功后可以拿到生成的主键值。但如果表没有自增列或者执行的是 INSERT IGNORE返回值可能不准。rowcount也同样有坑MySQL 在默认配置下如果一条 UPDATE 更新前后值相同可能返回 0而不是匹配的行数。如果业务需要判断“是否存在这条记录”要用SELECT 1而不是依赖rowcount。def user_exists(conn, user_id): with conn.cursor() as cur: cur.execute(SELECT 1 FROM users WHERE id %s, (user_id,)) return cur.fetchone() is not None3.4 常见异常处理和可重试判断Python 数据库操作常见的异常有这几类异常含义处理方式IntegrityError主键冲突、唯一约束冲突通常是数据问题不建议重试OperationalError连接断开、表不存在、锁等待超时需要检查连接状态必要时重连ProgrammingErrorSQL 语法错误、参数个数不对属于代码缺陷直接修代码DataError数值超出范围、日期格式错误属于入参问题校验数据写重试时要区分可重试和不可重试。连接断开的OperationalError可以重建连接后重新执行一次主键冲突的IntegrityError重试没有意义应该落到“更新已存在记录”或“忽略冲突”的业务策略上。4. 连接池、executemany 和批量写入别让数据库被你的循环拖垮4.1 为什么课程作业也要连接池短任务里每次执行都新建连接性能问题还不明显但在 Web 服务、批量脚本和定时任务里频繁建连的代价很高。一个 MySQL 连接从建立到认证完成往往需要几十毫秒如果每执行一条 SQL 都重新连接整体的吞吐量立刻崩溃。更严重的是数据库服务端有连接数上限短连接风暴会直接打满max_connections导致其他应用也连不上。连接池的价值是把连接复用起来同时限制应用侧并发连接的数量。早期写课程设计时大家喜欢写一个全局conn pymysql.connect(...)然后在 SQLite 和 MySQL 之间来回切换这种全局连接在单线程下没问题一旦引入多线程连接状态就容易串掉。连接池比全局连接更合适因为它按需创建连接同时在归还时清理状态。4.2 用 dbutils 实现一个可复用的 MySQL 连接池常见做法是使用dbutils包新版本的导入路径是dbutils.pooled_db。安装命令pip install pymysql DBUtils连接池初始化示例import pymysql from dbutils.pooled_db import PooledDB db_pool PooledDB( creatorpymysql, mincached2, maxcached5, maxconnections10, blockingTrue, ping1, host127.0.0.1, port3306, userapp_user, passwordyour_password, databasenorthwind, charsetutf8mb4, ) def get_db_connection(): return db_pool.connection()参数说明mincached2池中至少保持 2 个空闲连接避免刚启动时频繁建连。maxcached5最多缓存 5 个空闲连接超过 5 个的可用连接会在用完后直接关闭。maxconnections10连接池能支持的最大连接数超过时会根据blocking决定等待还是抛异常。blockingTrue连接用完后不会立即报错而是等待其他线程归还连接。ping1从连接池取出连接时进行轻量探活避免把已断开的连接交给业务。具体探活频率与 DBUtils 版本有关生产环境建议根据版本查看源码注释确认。需要注意连接池里的连接是有状态的。如果某段代码执行了SET SESSION或USE语句连接归还后可能影响其他线程。稳妥的做法是在关键操作前后调用conn.rollback()或conn.cursor().execute(SET SESSION ...)恢复默认状态而不是寄希望于连接池自动清理。4.3 executemany 批处理插入以及 1000 条一提交的边界批量插入最简单的思路是写一个 for 循环逐条 execute但这样要频繁交互网络数据量稍微上来就很难看。PyMySQL 提供executemany方法可以在一次调用里传入多条参数。对于 INSERT 语句PyMySQL 会尝试把多条参数合并成单条多值 INSERT显著减少网络往返。data [(fuser_{i}, i) for i in range(10000)] batch_size 1000 with get_db_connection() as conn: with conn.cursor() as cur: for start in range(0, len(data), batch_size): cur.executemany( INSERT INTO users (username, score) VALUES (%s, %s), data[start:start batch_size], ) conn.commit()这里分批的原因有两个第一MySQL 客户端和服务端都有max_allowed_packet限制单次 INSERT 太大容易超过限制第二批次过大会让数据库在事务中持有大量锁影响并发读。1000是一个相对保守的数值如果插入字段多、单行体积大还要往下调整比如500甚至200。4.4 批量写入和事务全部成功还是部分保留批量任务里最纠结的问题是失败策略。如果 10000 条数据里第 5000 条是脏数据前面 4999 条要不要保留如果全量回滚代价太高如果保留又会产生半截数据。实际运维中可以根据业务容忍度选择“小事务”策略每批一个事务批量成功才提交批量失败只回滚当前批次。这样可以做到“批内原子批间隔离”日志里也能记录到哪个批次失败。实现起来不复杂for start in range(0, len(data), batch_size): batch data[start:start batch_size] try: with get_db_connection() as conn: with conn.cursor() as cur: cur.executemany(insert_sql, batch) conn.commit() except Exception as exc: logger.error(batch %s failed: %s, start, exc) raise如果是纯粹的导入任务MySQL 还有更高效的LOAD DATA LOCAL INFILE方案先把数据写到 CSV 再一次性导入但需要数据库端允许LOCAL INFILE权限控制要提前确认。5. 数据库操作完成之后对账、幂等导入与结果校验操作数据库不只在写入时提心吊胆写完之后的验证同样重要。一个低成本的对账方法是比较源表和目标表的行数以及关键字段的汇总值。def table_basic_stats(conn, table_name, key_column): with conn.cursor() as cur: cur.execute( fSELECT COUNT(*), COALESCE(SUM({key_column}), 0) FROM {table_name} ) return cur.fetchone()注意table_name和key_column不能通过参数占位符传入因为表名和字段名无法参数化。如果这是公共函数必须做白名单校验防止拼接进任意内容。对账时同时比较行数和SUM(id)一旦有任何不一致再通过EXCEPT找出差异数据。如果是需要反复执行的导入任务最好把 SQL 写成幂等形式。MySQL 下可以使用INSERT ... ON DUPLICATE KEY UPDATE重复执行同一批数据不会产生重复记录insert_sql INSERT INTO customers (customer_id, company_name, region) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE company_name VALUES(company_name), region VALUES(region) 这样跑第二次导入时遇到相同主键只会更新字段不会报IntegrityError。这个写法在 MySQL 8.0.20 之后VALUES()函数被标记为弃用但 MySQL 5.7 和 MariaDB 仍然普遍可用。如果目标库是 SQLite可以用INSERT OR REPLACE INTO但要注意它会先删除旧记录再插入新记录可能引发自增主键变化。最后一个技巧是校验“写入是否真的生效”。commit()成功只能说明数据库端接收了事务不能保证数据一定如预期。简单的做法是在写完数据后立刻用独立连接回查time.sleep(0.2) # 等待事务日志落盘 with get_db_connection() as conn: with conn.cursor() as cur: cur.execute(SELECT COUNT(*) FROM customers) total cur.fetchone()[COUNT(*)] assert total expected_count, fexpected {expected_count}, got {total}这条断言放在定时任务收尾处比人工看日志可靠得多。对账通过之后再做下游数据更新否则继续确认批次失败点并决定是否重跑。本文还有配套的精品资源点击获取