Python 的DB-API

从哪里可以找到数据库相关的接口呢？很简单，只需要前往Python 官网的数据库主题部
分即可。在那里，你可以找到全面的当前版本DB-API（2.0 版本）的链接，包括数据库模块、
文档、特殊兴趣小组等。从一开始起，DB-API 就被移动至PEP 249 中（PEP 248 中的老版
DB-API 1.0 标准已经废弃）。那么，DB-API 是什么呢？

DB-API 是阐明一系列所需对象和数据库访问机制的标准，它可以为不同的数据库适配
器和底层数据库系统提供一致性的访问。就像很多基于社区的成果一样，DB-API 也是强需
求驱动的。

在过去的日子里，曾有这样一种场景：有很多种数据库，并且很多人实现了他们自己
的数据库适配器。就像做无用功一样。这些数据库和适配器是在不同的时间被不同的人实
现的，在功能上完全没有一致性可言。但是，这意味着使用这些接口的应用代码需要与他
们选择使用的数据库模块进行定制化处理，接口的任何改变都会导致应用代码的变更。

由此，为了解决Python 数据库连接问题的特殊兴趣小组成立了，并且撰写了1.0 版本的
DB-API。该API 为不同的关系数据库提供了一致性的接口，并且使不同数据库间移植代码变
得更加简单，通常只需要修改几行代码即可。本章后面的部分你会看到一个相关的示例。

模块属性

DB-API 标准要求必须提供下文列出的功能和属性。一个兼容DB-API 的模块必须定义表
6-1 所示的几个全局属性。

数据属性
apilevel
该字符串（注意，不是浮点型）指明了模块需要兼容的DB-API 最高版本，比如，1.0、
2.0 等。该属性的默认值是1.0。

threadsafety
这是一个整型值，可选值如下。

• 0：不支持线程安全。线程间不能共享模块。
• 1：最小化线程安全支持：线程间可以共享模块，但是不能共享连接。
• 2：适度的线程安全支持：线程间可以共享模块和连接，但是不能共享游标。
• 3：完整的线程安全支持：线程间可以共享模块、连接和游标。

如果有资源需要进行共享，那么就需要诸如自旋锁、信号量之类的同步原语来达到原子
锁定的目的。由于此目的，磁盘文件和全局变量都并不可靠，甚至还会干扰到标准的互斥操
作。查阅threading 模块，或回顾第4 章，以获取如何使用锁的更多信息。

参数风格
DB-API 支持以不同的方式指明如何将参数与SQL 语句进行整合，并最终传递给服务器中执
行。该参数是一个字符串，用于指定构建查询行或命令时使用的字符串替代形式（见表6-2）。

函数属性
connect()函数通过Connection 对象访问数据库。兼容模块必须实现connect()函数，该函
数创建并返回一个Connection 对象。表6-3 所示为connect()的参数。
可以使用包含多个参数的字符串（DSN）来传递数据库连接信息，也可以按照位置传递每个
参数，或者是使用关键字参数的形式传入。下面是PEP 249 中给出的使用connect()函数的例子。

connect(dsn='myhost:MYDB',user='guido',password='234$')

使用DSN 还是独立参数主要基于所连接的系统。比如，如果你使用的是像ODBC（Open
Database Connectivity）或JDBC（Java Database Connectivity）的API，则需要使用DSN；
而如果你直接使用数据库，则更倾向于使用独立的登录参数。另一个使用独立参数的原因
是很多数据库适配器并没有实现对DSN 的支持。下面是一些没有使用DSN 的connect()调
用。需要注意的是，并不是所有的适配器都会严格按照标准实现，比如MySQLdb 使用了
db 而不是database。

MySQLdb.connect(host='dbserv', db='inv', user='smith')
• PgSQL.connect(database='sales')
• psycopg.connect(database='template1', user='pgsql')
• gadfly.dbapi20.connect('csrDB', '/usr/local/database')
• sqlite3.connect('marketing/test')

异常
异常同样需要包含在兼容的模块中，如表6-4 所示。

Connection 对象

应用与数据库之间进行通信需要建立数据库连接。它是最基本的机制，只有通过数据库
连接才能把命令传递到服务器，并得到返回的结果。当一个连接（或一个连接池）建立后，
可以创建一个游标，向数据库发送请求，然后从数据库中接收回应。

Connection 对象方法
Connection 对象不需要包含任何数据属性，不过应当定义表6-5 所示的几个方法。

当使用close()时，这个连接将不能再使用，否则会进入到异常处理中。
如果数据库不支持事务处理，或启用了自动提交功能，commit()方法都将无法使用。如
果你愿意，可以实现单独的方法用来启动或关闭自动提交功能。因为本方法是DB-API 中的
一部分，所以对于不支持事务处理的数据库而言，只需要在方法中实现“pass”即可。

如果RDBMS 不支持游标，那么cursor()仍然会返回一个尽可能模仿真实游标的对象。这
是最基本的要求。每个适配器开发者都可以为他的接口或数据库专门添加特殊属性。

DB-API 建议适配器开发者为连接编写所有的数据库模块异常（见表6-4），但并没
有做强制要求。如果没有，则认为Connection 对象将会抛出对应模块级别的异常。当你
完成数据库连接并关闭游标时，需要对所有操作执行commit()，并对你的连接执行
close()。

Cursor 对象

当建立连接后，就可以和数据库进行通信了。正如6.1 节所述，游标可以让用户提交数
据库命令，并获得查询的结果行。Python DB-API 游标对象总能提供游标的功能，即使是那
些不支持游标的数据库。此时，如果你创建了一个数据库适配器，还必须要实现cursor 对象，
以扮演类似游标的角色。这样，无论你将数据库系统切换到支持游标的数据库还是不支持游
标的数据库，都能保持Python 代码的一致性。

当游标创建好后，就可以执行查询或命令（或多个查询和命令），并从结果集中取回一行
或多行结果。表6-6 所示为Cursor 对象的数据属性和方法。

游标对象最重要的属性是execute*()和fetch*()方法，所有针对数据库的服务请求都
是通过它们执行的。arraysize 数据属性在为fetchmany()设置默认大小时非常有用。当然，
在不需要时关闭游标是个好主意， 而如果你的数据库支持存储过程，可能会用到
callproc()。

类型对象和构造函数

通常，两个不同系统间的接口是最脆弱的。比如在Python 对象和C 类型中进行转换就是
这样。类似地，在Python 对象和原生数据库对象间也存在这个问题。作为一个使用Python
的DB-API 的程序员，虽然你传递给数据库的参数是字符串，但是数据库可能需要将其转换
为多种不同的类型，从而可以对任何特定查询都能给出正确的数据类型。

例如，一个Python 字符串是应该转换成VARCHAR、TEXT、BLOB，还是原生BINARY
对象，抑或是DATE 或TIME 对象呢？为数据库提供期望格式的输入必须非常小心，因此，
DB-API 的另一个需求是创建构造函数，从而构建可以简单地转换成适当数据库对象的特
殊对象。表6-7 给出了用于此目标的一些类。SQL 的NULL 值对应于Python 的NULL
对象None。

API 版本变更
从1.0 版本（1996 年）修订为2.0 版本（1999 年），DB-API 做了几个重要的变更。

• 从API 中移除了之前必须的dbi 模块。

• 更新了类型对象。

• 添加了新属性用于提供更好的数据库绑定。

• 重新定义了callproc()的语义以及execute()的返回值。

• 转换为基于类的异常。
  DB-API 2.0 版本发布后，曾在2002 年添加了一些刚才提到过的可选的扩展。自此之后，
  就再也没有重大变更了。关于DB-API 的持续讨论一直在DB-SIG 的邮件列表中进行。过去
  五年，对DB-API 的下一版本的可能性进行了讨论，该版本暂时命名为DB-API 3.0。它将包
  含如下特性。

• 当有一个新的结果集时，nextset()可以给出一个更好的结果值。

• 将float 转为Decimal。

• 改善参数风格的灵活性以及对其的支持情况。

• 预编译语句或语句缓存。

• 完善事务模型。

• 确定DB-API 可移植性的角色。

• 添加单元测试。

关系数据库

现在我们可以准备开始学习了，不过首先会有一个问题摆在我们面前：“Python 有哪些
数据库系统的接口呢？”或者说是：“Python 支持哪些平台呢？”答案是：“几乎所有的数据
库系统。”下面是一个大概（但不详尽）的接口列表。

商业RDBMS

• IBM Informix
• Sybase
• Oracle
• Microsoft SQL Server
• IBM DB2
• SAP
• Embarcadero Interbase
• Ingres
  开源RDBMS
• MySQL
• PostgreSQL
• SQLite
• Gadfly
  数据库API
• JDBC
• ODBC
  非关系数据库
• MongoDB
• Redis
• Cassandra
• SimpleDB
• Tokyo Cabinet
• CouchDB
• Bigtable（通过Google App Engine 的数据库API）

数据库和Python：适配器

对于每种支持的数据库，Python 都有一个或多个适配器用于连接Python 中的目标数
据库系统。比如Sybase、SAP、Oracle 和SQLServer 这些数据库就都存在多个可用的适配
器。我们需要做的事情就是挑选出最合适的适配器。你的挑选标准可能包括：它的性能如
何，它的文档和/或网站是否有用，是否有一个活跃的社区，驱动的质量和稳定性如何等。
需要记住的是，大多数适配器只提供给你连接数据库的基本需求，所以你还需要寻找一些
额外的特性。请记住，你需要负责编写更高级别的代码，比如线程管理和数据库连接池管
理等。

如果你不希望有太多的交互操作，比如你希望少写一些SQL 语句，或者尽可能少地参与
数据库管理的细节，那么你可以考虑ORM，本话题将在本章后面的小节中进行展开。

现在，让我们看几个使用适配器模块与关系数据库进行通信的例子。真正的秘密在于建
立连接。一旦你建立连接，并使用DB-API 的对象、属性和对象方法，你的核心代码就会看
起来很相似，而无须去管它使用了哪个适配器以及RDBMS。

使用数据库适配器的示例

首先，让我们看一些代码片段，包括创建数据库、创建表和使用表。这里会给出使用
MySQL、PostgreSQL 和SQLite 的例子。

MySQL
本例中我们将使用MySQL，以及Python 中著名的MySQL 适配器：MySQLdb（即
MySQL-Python）。而当我们的话题转为Python 3 时，会讨论另一个MySQL 适配器：MySQL
Connector/Python。在下面的代码中，我们会故意留下错误，从而让你能够自己想到创建
处理程序。

首先我们以管理员的身份登录数据库，创建数据库并赋予权限，然后再以普通用户的身
份重新登录数据库客户端，具体代码如下所示。

>>> import MySQLdb
>>> cxn = MySQLdb.connect(user='root')
>>> cxn.query('DROP DATABASE test')
Traceback (most recent call last):
File "<stdin>", line 1, in ?
_mysql_exceptions.OperationalError: (1008, "Can't drop, database
'test'; database doesn't exist")
>>> cxn.query('CREATE DATABASE test')
>>> cxn.query("GRANT ALL ON test.* to ''@'localhost'")
>>> cxn.commit()
>> > cxn.close()

在上面的代码中，并没有使用游标。一些适配器有Connection 对象，这些对象可以使用
query()方法执行SQL 查询，不过不是所有的适配器都能这样。因此，建议或者不要使用这个
方法，或者事先检查适配器中该方法是否可用。

commit()方法是可选的，因为MySQL 默认开启了自动提交。下面我们要重新作为普通用
户登录这个新数据库，创建表，然后通过Python 执行一些常用的SQL 查询和命令。这次会
使用游标以及execute()方法。

下面的代码展示了创建表的方法。重复创建表（在没有事先删除表的情况下）会导致一
个错误发生。

>>> cxn = MySQLdb.connect(db='test')
>>> cur = cxn.cursor()
>>> cur.execute('CREATE TABLE users(login VARCHAR(8), userid INT)')
0L

现在，向数据库中添加一些行，并对其进行查询。

>>> cur.execute("INSERT INTO users VALUES('john', 7000)")
1L
>>> cur.execute("INSERT INTO users VALUES('jane', 7001)")
1L
>>> cur.execute("INSERT INTO users VALUES('bob', 7200)")
1L
>>> cur.execute("SELECT * FROM users WHERE login LIKE 'j%'")
2L
>>> for data in cur.fetchall():
... print '%s\t%s' % data
...
john 7000
ja ne 7001

最后一个功能是更新表，包括更新和删除行。

>>> cur.execute("UPDATE users SET userid=7100 WHERE userid=7001")
1L
>>> cur.execute("SELECT * FROM users")
3L
>>> for data in cur.fetchall():
... print '%s\t%s' % data
...
john 7000
jane 7100
bob 7200
>>> cur.execute('DELETE FROM users WHERE login="bob"')
1L
>>> cur.execute('DROP TABLE users')
0L
>>> cur.close()
>>> cxn.commit()
>> > cxn.close()

MySQL 是目前最流行的开源数据库之一，因此存在可用的Python 适配器并不令人意外。

PostgreSQL
另一个流行的开源数据库是PostgreSQL。与MySQL 不同，Postgres 至少包含3 种Python
适配器：psycopg、PyPgSQL 和PyGreSQL。还有一种适配器，叫PoPy，目前已废弃，并且
在2003 年将其项目与PyGreSQL 进行了合并。目前剩下的这三种适配器都有其自己的特性和
优缺点，所以根据实践对其进行选择更加明智。

当我们介绍各种适配器的使用方法时，需要注意PyPgSQL 自2006 年起就不再开发了，
而PyGreSQL 则是在2009 年发布的最新版本（4.0）。这两种适配器不再活跃，使得psycopg
成为PostgreSQL 适配器的唯一引领者，因此本书的示例最终将使用该适配器。psycopg 目前
已进入到第二个版本，这意味着虽然相关的示例中使用了版本1 的psycopg 模块，但是当下
载它时，需要使用psycopg2 来代替。

庆幸的是，这几种适配器的接口都很相似，所以可以创建一个应用，并对比这三种适配器
的性能（如果性能对你来说很重要）。下面的代码是每种适配器创建Connection 对象的代码。

psycopg

>>> import psycopg
>> > cxn = psycopg.connect(user='pgsql')

PyPgSQL

>>> from pyPgSQL import PgSQL
>> > cxn = PgSQL.connect(user='pgsql')

PyGreSQL

>>> import pgdb
>> > cxn = pgdb.connect(user='pgsql')

下面是一些可以用于这三种适配器的通用代码。

>>> cur = cxn.cursor()
>>> cur.execute('SELECT * FROM pg_database')
>>> rows = cur.fetchall()
>>> for i in rows:
... print i
>>> cur.close()
>>> cxn.commit()
>> > cxn.close()

最后，你可以看到每种适配器的输出结果有些许不同。

数据库适配器示例应用

在下面的例子中，我们将演示如何使用Python 来访问数据库。为了能够尽可能多地演示
多样性和代码，这里添加了对3 种不同数据库系统的支持：Gadfly、SQLite 以及MySQL。为
了后续加入更多的东西，首先会给出Python 2.x 下的完整源码，但不会给出逐行解释。

应用的运行与之前小节中描述的要点非常相似，所以你应该可以在没有完整解释的情况
下理解其功能，只需要从底部的main()函数开始即可（为了保持简单，对于拥有服务器的完
整系统，如MySQL，我们将直接作为root 用户进行登录，当然这种做法在生产环境中不提
倡）。下面是应用的源码，其文件名为ushuffle_db.py。

#!/usr/bin/env python3
"""
Database Operations Module (2025-08-26)
Optimized version supporting SQLite, MySQL and Gadfly
"""

import os
import logging
from random import randrange as rand
from typing import Tuple, Optional, Dict, Any, Iterator


# Constants and Configuration
class Config:
    COLSIZ = 10
    FIELDS = ('login', 'userid', 'projid')
    RDBMS_MAP = {'s': 'sqlite', 'm': 'mysql', 'g': 'gadfly'}
    DB_NAME = 'test'
    DB_USER = 'root'
    NAME_LEN = 16


# Logging Setup
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('db_operations.log'),
        logging.StreamHandler()
    ]
)

# Formatting Utilities
tformat = lambda s: str(s).title().ljust(Config.COLSIZ)
cformat = lambda s: s.upper().ljust(Config.COLSIZ)

# Sample Data
NAMES = (
    ('aaron', 8312), ('angela', 7603), ('dave', 7306),
    ('davina', 7902), ('elliot', 7911), ('ernie', 7410),
    ('jess', 7912), ('jim', 7512), ('larry', 7311),
    ('leslie', 7808), ('melissa', 8602), ('pat', 7711),
    ('serena', 7003), ('stan', 7607), ('faye', 6812),
    ('amy', 7209), ('mona', 7404), ('jennifer', 7608),
)


class DatabaseManager:
    def __init__(self):
        self.connection = None
        self.cursor = None
        self.db_type = None
        self.db_exc = None

    def setup(self) -> str:
        """Setup and return selected database type"""
        while True:
            try:
                choice = input('''
Choose a database system:
(M)ySQL 
(G)adfly
(S)QLite 
Enter choice: ''').strip().lower()[0]
                if choice in Config.RDBMS_MAP:
                    self.db_type = Config.RDBMS_MAP[choice]
                    return self.db_type
                raise ValueError
            except (IndexError, ValueError):
                print("Invalid choice. Please enter M, G or S.")

    def connect(self) -> bool:
        """Establish database connection"""
        db_dir = f"{self.db_type}_{Config.DB_NAME}"

        try:
            if self.db_type == 'sqlite':
                import sqlite3
                self.db_exc = sqlite3
                os.makedirs(db_dir, exist_ok=True)
                self.connection = sqlite3.connect(
                    os.path.join(db_dir, Config.DB_NAME)
                )

            elif self.db_type == 'mysql':
                import mysql.connector
                from mysql.connector import errorcode
                self.db_exc = mysql.connector

                try:
                    self.connection = mysql.connector.connect(
                        user=Config.DB_USER,
                        database=Config.DB_NAME,
                        autocommit=True
                    )
                except self.db_exc.Error as err:
                    if err.errno == errorcode.ER_BAD_DB_ERROR:
                        # Create database if not exists
                        conn = mysql.connector.connect(user=Config.DB_USER)
                        cursor = conn.cursor()
                        cursor.execute(f"CREATE  DATABASE {Config.DB_NAME}")
                        conn.close()
                        self.connection = mysql.connector.connect(
                            user=Config.DB_USER,
                            database=Config.DB_NAME
                        )
                    else:
                        raise

            elif self.db_type == 'gadfly':
                from gadfly import gadfly
                self.db_exc = gadfly
                try:
                    self.connection = gadfly(Config.DB_NAME, db_dir)
                except IOError:
                    os.makedirs(db_dir, exist_ok=True)
                    self.connection = gadfly()
                    self.connection.startup(Config.DB_NAME, db_dir)

            self.cursor = self.connection.cursor()
            return True

        except ImportError as e:
            logging.error(f"Database  driver not installed: {e}")
        except Exception as e:
            logging.error(f"Connection  failed: {e}")
        return False

    def execute_sql(self, sql: str, params=None) -> bool:
        """Execute SQL with error handling"""
        try:
            if params:
                self.cursor.execute(sql, params)
            else:
                self.cursor.execute(sql)
            return True
        except self.db_exc.Error as e:
            logging.error(f"SQL  execution failed: {e}\nSQL: {sql}\nParams: {params}")
            return False

    def create_table(self) -> bool:
        """Create users table with indexes"""
        try:
            sql = f"""
            CREATE TABLE IF NOT EXISTS users (
                login VARCHAR({Config.NAME_LEN}),
                userid INTEGER,
                projid INTEGER,
                INDEX idx_projid (projid),
                INDEX idx_userid (userid)
            )"""
            return self.execute_sql(sql)
        except Exception as e:
            logging.error(f"Table  creation failed: {e}")
            return False

    def drop_table(self) -> bool:
        """Drop users table"""
        return self.execute_sql("DROP  TABLE IF EXISTS users")

    def rand_name_generator(self) -> Iterator[Tuple[str, int]]:
        """Generate random names from sample data"""
        pick = set(NAMES)
        while pick:
            yield pick.pop()

    def insert_data(self, batch_size: int = 5) -> int:
        """Insert data in batches"""
        count = 0
        generator = self.rand_name_generator()

        if self.db_type in ('sqlite', 'gadfly'):
            query = "INSERT INTO users VALUES(?, ?, ?)"
        else:  # mysql
            query = "INSERT INTO users VALUES(%s, %s, %s)"

        while True:
            batch = [
                (*next(generator), rand(1, 5))
                for _ in range(batch_size)
            ]
            if not batch:
                break
            if self.execute_sql(query, batch):
                count += len(batch)

        return count

    def update_groups(self) -> Tuple[int, int, int]:
        """Randomly update user groups"""
        fr = rand(1, 5)
        to = rand(1, 5)

        if self.db_type in ('sqlite', 'gadfly'):
            query = "UPDATE users SET projid=? WHERE projid=?"
        else:
            query = "UPDATE users SET projid=%s WHERE projid=%s"

        self.execute_sql(query, (to, fr))
        return fr, to, self.cursor.rowcount

    def delete_group(self) -> Tuple[int, int]:
        """Randomly delete a user group"""
        rm = rand(1, 5)

        if self.db_type in ('sqlite', 'gadfly'):
            query = "DELETE FROM users WHERE projid=?"
        else:
            query = "DELETE FROM users WHERE projid=%s"

        self.execute_sql(query, (rm,))
        return rm, self.cursor.rowcount

    def display_data(self):
        """Display formatted table data"""
        if not self.execute_sql("SELECT  * FROM users"):
            return

        headers = ''.join([f"{field.upper():<{Config.COLSIZ}}"
                           for field in Config.FIELDS])
        print(f"\n{headers}")

        for data in self.cursor.fetchall():
            row = ''.join([f"{str(item).title():<{Config.COLSIZ}}"
                           for item in data])
            print(row)

    def close(self):
        """Close database connection"""
        if self.cursor:
            self.cursor.close()
        if self.connection:
            if self.db_type == 'mysql':
                self.connection.commit()
            self.connection.close()


def main():
    """Main execution function"""
    db_manager = DatabaseManager()

    print(f"\n{' Database Operations ':=^40}")
    print(f"Date: 2025-08-26 (Tuesday)")

    # Step 1: Setup database type
    db_type = db_manager.setup()
    print(f"\n[1/6] Connecting to {db_type.upper()}  database...")

    # Step 2: Connect to database
    if not db_manager.connect():
        print("ERROR: Database not supported or unreachable")
        return

    # Step 3: Create table
    print("\n[2/6] Creating users table...")
    if not db_manager.create_table():
        print("ERROR: Failed to create table")
        return

        # Step 4: Insert data
    print("\n[3/6] Inserting sample data...")
    count = db_manager.insert_data()
    print(f"Inserted {count} records")
    db_manager.display_data()

    # Step 5: Update groups
    print("\n[4/6] Updating user groups...")
    fr, to, num = db_manager.update_groups()
    print(f"Moved {num} users from group {fr} to {to}")
    db_manager.display_data()

    # Step 6: Delete group
    print("\n[5/6] Deleting random group...")
    rm, num = db_manager.delete_group()
    print(f"Deleted {num} users from group {rm}")
    db_manager.display_data()

    # Cleanup
    print("\n[6/6] Cleaning up...")
    db_manager.drop_table()
    db_manager.close()
    print("\nOperation completed successfully")


if __name__ == '__main__':
    main()