企业官网建设流程全解析

前言

「Python 怎么做数据库编程」这个问题，标准答案是一份文档编号：PEP 249 — Python Database API Specification v2.0，通常简称DB-API 2.0。它不规定 Python 怎么连某个具体数据库，而是规定所有数据库驱动应该长成什么样：模块必须暴露哪些名字、连接对象有哪些方法、出错时抛什么异常。

理解这一点之后，很多现象就解释得通了：为什么换数据库时connect()/cursor()/execute()/fetchall()这套写法几乎不用改？因为它们本来就是同一份规范定下来的。为什么sqlite3的参数占位符是?，而 PyMySQL 里是%s？因为规范允许驱动在五种占位符风格里自选。

但有一个必须澄清的误解：「Python 有统一的数据库接口」不等于「一份代码能跑所有数据库」。DB-API 统一的是函数签名，不统一SQL 方言。LIMIT 10 OFFSET 20大部分库能用，可INSERT OR REPLACE、AUTOINCREMENT、日期函数、字符串函数各写各的。换库的工作量主要在 SQL，不在 API。

本文按规范的顺序讲：模块级约定 → 五步流程 → 参数风格 → 类型对象 → 事务与游标生命周期 → 怎么写出可移植的数据访问层。

一、模块级约定：驱动必须提供的东西

一个符合 DB-API 2.0 的模块，必须定义三个全局常量和一棵异常树。

三个全局常量

paramstyle的五种取值：

threadsafety是每个库自己声明的，不要假设。比如 SQLite 的sqlite3.threadsafety具体是几取决于底层库怎么编译；PyMySQL 声明的是1（可共享模块，不能共享连接）。跨线程共享连接或游标之前，先读这个常量，而不是凭「我记得它线程安全」下结论。

异常树

规范要求模块提供这棵继承结构，捕获的粒度从这里来：

Exception
|__Warning
|__Error ← 所有数据库错误的根，捕它等于「捕获所有数据库异常」
|__InterfaceError ← 接口层问题（驱动自身）
|__DatabaseError ← 数据库层问题
|__DataError ← 数据问题：除零、数值超范围等
|__OperationalError ← 运行问题：断连、找不到数据源、事务处理失败
|__IntegrityError ← 完整性约束：外键检查失败、唯一键冲突
|__InternalError ← 数据库内部错误：游标失效、事务失同步
|__ProgrammingError ← 编程错误：表不存在、语法错误、参数个数不匹配
|__NotSupportedError ← 用了数据库不支持的操作

实践建议：捕获Error而不是Exception。前者只会抓到数据库相关的错误，后者会把KeyboardInterrupt、TypeError这类完全无关的问题一起吞掉，掩盖真正的 bug。

顺带提一句旧代码：Python 2.7 已于 2020 年 1 月 1 日停止维护，网上大量数据库教程还停留在那个时代。如果你翻到的代码里有print "..."这种语句形式、有dict.iteritems()/xrange、有except Error, e这种写法，那就是 Python 2 的语法，在 Python 3 下直接是语法错误或AttributeError。现行写法分别是print(...)函数调用、.items()/range、except Error as e。看到这类代码，正确做法是整体按 Python 3 重写，而不是逐行打补丁。

二、五步流程

所有 DB-API 驱动都遵循同一套动作序列：

一个最小但完整的例子：

# 适用于 Python 3.8+（这里用标准库 sqlite3，无需安装第三方库）
import sqlite3

conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row

try:
with conn.cursor() as cur:
cur.execute("CREATE TABLE item(id INTEGER PRIMARY KEY, name TEXT, qty INTEGER)")
cur.execute("INSERT INTO item(name, qty) VALUES (?, ?)", ("螺栓", 120))
cur.execute("INSERT INTO item(name, qty) VALUES (?, ?)", ("螺母", 300))

cur.execute("SELECT id, name, qty FROM item WHERE qty > ? ORDER BY qty DESC", (100,))
for row in cur:
print(row["id"], row["name"], row["qty"])
conn.commit()
except sqlite3.Error:
conn.rollback()
raise
finally:
conn.close()

预期输出
2 螺母 300
1 螺栓 120

三、参数化：唯一正确的传值方式

规范里对应的是「execute(operation[, parameters])」的第二个参数。它不是格式化，是把值交给驱动按目标数据库的规则绑定。

对比一下两种做法，差别就在这里：

参数化能防注入，原理不是「转义了特殊字符」，而是「值和语法从一开始就不在同一条通道上」。这也是为什么它对各种花式引号、编码绕过都天然免疫。

三个常见的书写错误：

• 
  
• 单元素参数忘了逗号：(value)是标量，(value,)才是元组。
• 
  
• 占位符加了引号：'%s'会让驱动把整个占位符当字符串字面量，参数绑定不上。
• 
  
• 标识符想用参数化：表名、列名、排序方向不能参数化，必须用白名单。占位符只能绑「值」。
• 
  

四、类型对象与构造函数

规范还定义了一组类型对象（type object）和构造函数（constructor），用来在不依赖具体数据库类型名的前提下描述和构造值。

类型对象是单例，用于和cursor.description里的type_code比对：

构造函数把常见的 Python 值转换成驱动能接受的数据库值：

需要注意：各驱动对这些构造函数的支持程度不一，有的只提供一部分，有的直接复用 Python 的datetime类型。写可移植代码时，最保险的做法是用 Python 的datetime对象直接作为参数传进去，让驱动自己完成适配；只有在必须显式指定时才用这些构造函数。用之前以你所选驱动的官方文档为准。

cursor.description每一列是一个7 元序列，依次是：

(name, type_code, display_size, internal_size, precision, scale, null_ok)

其中只有前两项name和type_code是必须提供的，其余可能为None。所以想动态拿到列名列表，可以这样写：

# 适用于 Python 3.8+
def column_names(cursor):
"""从 description 里取出列名；没有结果集时返回空列表。"""
if cursor.description is None:
return []
return [col[0] for col in cursor.description]

description在没有执行过execute或执行的语句不返回结果集（比如INSERT）时为None，所以上面这个空值判断是必要的。

五、事务、游标生命周期与可选扩展

事务

规范要求Connection提供commit()和rollback()，并且自动提交在初始状态必须是关闭的。这就是为什么各驱动里写操作都必须显式提交。

关闭一个未提交的连接会触发隐式回滚。这条行为在规范里有明确说明，但不要依赖它来做事务控制——显式调用rollback()才是可读、可控的写法。

游标

游标是有明确「有效期」的对象：关闭之后，任何操作都应抛异常。常见做法是用上下文管理器包住，出作用域自动关闭：

# 适用于 Python 3.8+
with conn.cursor() as cur:
cur.execute("SELECT ...")

注意区分两种with：with conn:管的是事务（提交/回滚），with conn.cursor() as cur:管的是游标（关闭）。两者互不替代，连接本身的关闭在任何驱动里都要自己负责。

可选扩展

规范把一批能力标为「可选扩展」，因为不是所有数据库都支持：

写库代码时，callproc/nextset这类扩展必须先确认你的驱动支持，再加到关键路径上。它们在 sqlite3 里就没有对应实现。

六、怎么写出可移植的数据访问层

既然 API 统一、SQL 不统一，可移植的关键就是把「不统一的东西」集中到一处。

# 适用于 Python 3.8+
class Database:
"""把驱动差异关在门后：上层只看到业务方法。"""

def __init__(self, connect_factory):
self._factory = connect_factory
self.conn = None

def open(self, **kwargs):
self.conn = self._factory(**kwargs)
return self

def close(self):
if self.conn is not None:
self.conn.close()
self.conn = None

def query(self, sql, params=()):
with self.conn.cursor() as cur:
cur.execute(sql, params)
return cur.fetchall()

def execute(self, sql, params=()):
"""执行一条写语句，返回受影响行数；失败则回滚。"""
try:
with self.conn.cursor() as cur:
cur.execute(sql, params)
affected = cur.rowcount
self.conn.commit()
return affected
except Exception:
self.conn.rollback()
raise

要点是：

• 
  
• 连接由一个工厂函数创建，换库时只换这个工厂，不动上层。
• 
  
• execute把「提交 / 回滚」收口在一个地方，业务代码不再各自记忆commit。
• 
  
• SQL 集中管理：把?和%s的差异收敛到每种数据库各自的语句定义里。
• 
  

需要提醒的是：rowcount在语句执行失败或结果不可判定时可能是-1，规范对此有明确说明。所以affected == 0和affected == -1含义不同，业务判断时要区分。

底线是：可移植性是设计出来的，不是靠驱动「兼容」出来的。

常见坑点

1. 捕获Exception吞掉一切

❌try: ... except Exception: pass——连KeyboardInterrupt、TypeError都会被吞掉，真正的 bug 被埋掉。

✅ 捕Error（所有数据库错误的公共基类）或更具体的子类：

try:
...
except sqlite3.Error:
conn.rollback()
raise

2. 假定所有驱动的占位符都是%s

❌ 在 SQLite 上写cur.execute("... WHERE id = %s", (1,))，sqlite3不认这个风格。

✅ 先看模块.paramstyle：SQLite 用?或:name；多数 MySQL / PostgreSQL 驱动用%s。写之前读常量，不要背。

3. 以为「DB-API 统一」就等于「SQL 通用」

❌ 直接把 SQLite 上的INSERT OR REPLACE、AUTOINCREMENT搬到别的库。

✅ API 统一、SQL 不统一；迁移前把方言差异列成清单逐条替换。

4. 把表名/列名放进参数

❌cur.execute("SELECT * FROM ? WHERE ? = ?", (table, col, val))——表名列名不能参数化，必然报错或行为异常。

✅ 值走占位符，标识符走白名单校验后再拼接。

5. 忘commit或忘rollback

❌ 写操作跑完不提交，数据「消失了」；异常路径不回滚，连接卡在失败事务里，后续操作连环报错。

✅ 写操作后commit()；except分支里rollback()。

6. 混淆两种with

❌ 以为with conn:用完就把连接关了，于是长期运行的服务连接数持续增长。

✅with conn:管事务，with conn.cursor() as cur:管游标，连接关闭自己负责。

7. 假设cursor.description一定有值

❌names = [c[0] for c in cur.description]在INSERT之后执行——description是None，抛TypeError。

✅ 先判空：if cur.description is None: return []。

8. 用rowcount当作精确结果数

❌cur.execute("SELECT ..."); print(cur.rowcount)——规范允许返回-1（结果不可判定时），很多驱动对SELECT就是返回-1。

✅ 取结果用fetchall()后算len()；rowcount只在写语句后判受影响行数。

总结

数据库编程在 Python 里之所以「换库不太痛」，靠的就是这份规范；而之所以「换库还是会痛」，是因为规范管不到 SQL 方言。想清楚这条分界线，就知道哪些代码该抽象、哪些必须为具体数据库单独写。