wiki/FormalSciences/ComputerScience/ProgrammingLanguage/Python/2.核心进阶/2.7-数据库操作.md

title	description	keywords	tags	author	date
数据库操作	Python DB-API 2.0 规范是一个重要的标准，它为 Python 程序与数据库间的交互提供了一致的接口。这个规范的目标是简化数据库编程，同时保持足够的灵活性来支持不同的数据库系统。	Python DB-API 2.0 数据库	Python/进阶 技术/程序语言	仲平	2024-04-26

Python DB-API 2.0 规范概述

**Python DB-API 2.0 规范是一个重要的标准，它为 Python 程序与数据库间的交互提供了一致的接口。**这个规范的目标是简化数据库编程，同时保持足够的灵活性来支持不同的数据库系统。

DB-API 2.0 规范定义在 Python 标准库的 PEP 249 文档中。它规定了数据库驱动（或数据库模块）应遵循的接口标准，以便程序员可以使用一致的编程风格来访问不同的数据库系统。

核心组件

组件	描述	关键方法或属性
连接对象（Connection）	代表数据库的连接。	- connect(): 连接数据库。 - commit(): 提交当前事务。 - rollback(): 回滚当前事务。 - close(): 关闭连接。
游标对象（Cursor）	用于执行查询和获取结果。	- execute(sql, [parameters]): 执行 SQL 语句。 - executemany(sql, seq_of_parameters): 执行相同的 SQL 语句多次。 - fetchone(): 获取结果集的下一行。 - fetchmany(size): 获取结果集的下几行。 - fetchall(): 获取结果集中的所有行。 - close(): 关闭游标对象。

标准方法

方法类型	方法名	描述
连接方法	connect()	连接到数据库。
	commit()	提交当前的事务。
	rollback()	回滚当前的事务。
	close()	关闭数据库连接。
游标方法	execute(sql, [parameters])	执行一个 SQL 语句。
	executemany(sql, seq_of_parameters)	执行相同的 SQL 语句多次。
	fetchone()	从结果集中获取下一行。
	fetchmany(size)	从结果集中获取多行。
	fetchall()	获取结果集中的所有行。
	close()	关闭游标对象。

标准异常

异常类别	描述
Warning	警告类的基类，用于非致命性问题的提示。
Error	与数据库相关错误的基类。如果未指定具体的异常，则抛出这个错误。
InterfaceError	与数据库接口（而非数据库本身）相关的错误。
DatabaseError	数据库操作过程中发生的错误的基类。
DataError	数据异常，比如数值溢出、数据类型不匹配等。
OperationalError	数据库操作中的内部错误，如连接问题、内存分配问题等。
IntegrityError	数据完整性相关的错误，如外键违反等。
InternalError	数据库内部错误，如游标无效、事务同步失效等。
ProgrammingError	程序错误，如表找不到、SQL 语句语法错误等。
NotSupportedError	当尝试使用数据库不支持的功能或 API 时抛出的错误。

Python 数据库操作

首先，需要安装对应数据库的 Python 库，例如，如果是使用 MySQL，可以安装 mysql-connector-python。然后，可以使用如下代码连接到数据库并执行简单的 SQL 语句：

import sqlite3

try:
    # 连接到 SQLite 数据库
    # 如果文件不存在，会自动创建
    conn = sqlite3.connect('example.db')

    # 创建一个游标对象，用于执行 SQL 命令
    cursor = conn.cursor()

    # 创建一个新表
    # IF NOT EXISTS 用于避免在表已存在时产生错误
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS stocks
        (date text, trans text, symbol text, qty real, price real)
    ''')

    # 插入一条记录
    cursor.execute("INSERT INTO stocks VALUES ('2023-11-17','BUY','AAPL',100,35.14)")

    # 提交当前事务
    conn.commit()

    # 查询数据
    cursor.execute("SELECT * FROM stocks WHERE symbol = 'AAPL'")
    print(cursor.fetchall())

# 捕获任何可能发生的异常
except sqlite3.Error as e:
    print(f"数据库错误: {e}")
except Exception as e:
    print(f"非数据库错误: {e}")
finally:
    # 关闭游标
    if cursor:
        cursor.close()
    # 关闭数据库连接
    if conn:
        conn.close()

连接（Connection）

连接对象代表了 Python 应用程序和数据库之间的连接。通过这个连接，程序能够执行 SQL 命令、处理事务。

创建连接

• 连接是通过调用特定数据库模块的 connect() 函数创建的。
• 这个函数通常需要数据库特定的参数，如主机名、数据库名、用户名和密码。

import sqlite3

# 创建连接到 SQLite 数据库文件 'example.db'
conn = sqlite3.connect('example.db')

关键特性和方法

1. 事务处理：
   • commit(): 提交当前事务。在执行如 INSERT、UPDATE、DELETE 等操作后，需要调用此方法以确保更改被保存。
   • rollback(): 回滚当前事务。在遇到错误或需要撤销更改时使用。
2. 连接管理：
   • close(): 关闭连接。关闭后，连接对象和其下所有的游标将不可用。
3. 错误处理：
   • 连接操作可能会抛出 DB-API 定义的异常，如 InterfaceError 或 DatabaseError。

try:
    # 执行数据库操作
    conn.commit()  # 提交事务
except Exception as e:
    conn.rollback()  # 回滚事务
    raise e
finally:
    conn.close()  # 关闭连接

Python 进行数据库操作务必管理好数据库连接的生命周期。使用完毕后应及时关闭，避免资源泄露。建议使用 Python 的上下文管理器（with 语句）可以自动管理连接的开启和关闭。

游标（Cursor）

游标是通过连接对象创建的，用于执行 SQL 命令和处理查询结果。

创建游标

• 通过连接对象的 cursor() 方法创建。
• 可以创建多个游标，用于执行不同的数据库操作。

cursor = conn.cursor()  # 创建游标

关键特性和方法

1. 执行 SQL 命令：
   • execute(sql, [parameters]): 执行单个 SQL 命令。
   • executemany(sql, seq_of_parameters): 执行相同 SQL 命令多次，但使用不同的参数。
2. 结果处理：
   • fetchone(): 返回结果集的下一行。
   • fetchmany(size): 返回结果集的下几行。
   • fetchall(): 返回结果集中的所有行。
3. 元数据获取：
   • 游标对象提供了诸如 description 属性，可以用来获取查询结果的列信息。
4. 游标管理：
   • close(): 关闭游标。关闭后，游标将不可用。

# 执行 SQL 查询
cursor.execute('SELECT * FROM some_table')

# 获取查询结果
rows = cursor.fetchall()  # 获取所有行
for row in rows:
    print(row)

# 关闭游标
cursor.close()

为避免资源占用，执行完毕后应及时关闭游标。使用参数化查询而不是字符串拼接，以防止 SQL 注入攻击。

执行 SQL 语句（execute）

在 Python 中，使用游标对象执行 SQL 语句并处理结果。

• 使用游标对象的 execute() 方法执行 SQL 语句。
• 对于查询操作（如 SELECT），结果可以通过游标提供的方法获取。
• 对于非查询操作（如 INSERT、UPDATE、DELETE），结果通常是影响的行数。

# 执行一个查询
cursor.execute('SELECT * FROM some_table')

# 或者执行一个更新
cursor.execute('UPDATE some_table SET column = value WHERE condition')

处理结果

• 使用 fetchone()、fetchmany(size) 或 fetchall() 方法获取查询结果。
• fetchone() 返回单行，fetchmany(size) 返回指定数量的行，fetchall() 返回所有行。

# 执行查询
cursor.execute('SELECT * FROM some_table')

# 获取单行
row = cursor.fetchone()
print(row)

# 获取多行
rows = cursor.fetchmany(5)
for row in rows:
    print(row)

# 获取所有行
all_rows = cursor.fetchall()
for row in all_rows:
    print(row)

错误处理

在 Python 数据库编程中，错误处理是必不可少的部分。常见的数据库错误包括连接错误、SQL 语法错误、数据类型不匹配等。合理的错误处理可以提高程序的健壮性和用户体验。

• 使用 try-except 块捕获并处理数据库异常。
• 利用 Python 的 DB-API 定义的异常类（如 DatabaseError，IntegrityError 等）来识别和响应特定的错误情况。

try:
    conn.execute('SELECT * FROM non_existent_table')
except sqlite3.DatabaseError as e:
    print("Database error occurred:", e)

调试数据库应用

• 使用日志记录重要的操作和异常信息。
• 在开发阶段，打印或记录 SQL 语句和其参数，以便检查和调整。
• 使用交互式 Python 环境（如 IPython 或 Jupyter Notebook）进行逐步执行和测试。

参数化查询

参数化查询是一种编写 SQL 查询的方法，可以提高安全性和灵活性。

编写安全的 SQL 查询以避免 SQL 注入

通过使用占位符而非直接拼接字符串来构建 SQL 语句，可以有效防止 SQL 注入攻击。

# 错误的做法：直接字符串拼接
unsafe_sql = "SELECT * FROM users WHERE name = '" + user_input + "'"

# 正确的做法：使用参数化查询
cursor.execute("SELECT * FROM users WHERE name = %s", (user_input,))

使用参数而非字符串拼接

在构建 SQL 语句时，应优先考虑使用参数化查询。

user_id = 5
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))

事务管理

基本概念（ACID 特性）

事务是数据库操作的基本单位，它应满足 ACID 特性，即原子性（Atomicity）、一致性（Consistency）、隔离性（Isolation）、持久性（Durability）。

• 原子性：确保所有操作要么全部完成，要么全部不执行。
• 一致性：事务完成时，数据必须处于一致状态。
• 隔离性：多个事务同时进行时，一个事务的操作不应影响其他事务。
• 持久性：一旦事务提交，其结果就永久保存在数据库中。

在 Python 中进行事务控制

在 Python 中，事务控制通常由连接对象（Connection）来管理。使用 commit() 和 rollback() 方法来控制事务的提交和回滚。

conn = sqlite3.connect('example.db')
try:
    # 开始一个事务
    conn.execute('INSERT INTO table_name VALUES (...)')

    # 更多数据库操作...

    # 提交事务
    conn.commit()
except Exception as e:
    # 出现错误，回滚事务
    conn.rollback()
    raise e
finally:
    # 关闭连接
    conn.close()

数据库连接池

理解数据库连接池的概念

数据库连接池是一种创建和管理数据库连接的技术，用于减少建立和关闭连接的开销。在高并发情况下，连接池可以显著提高性能。

使用连接池提高性能

在 Python 中，可以使用第三方库（如 sqlalchemy.pool）来实现数据库连接池。连接池管理着一组连接，当需要时分配给请求者，使用完毕后返回池中。

from sqlalchemy.pool import QueuePool
from sqlalchemy import create_engine

# 创建一个带连接池的引擎
engine = create_engine('postgresql://username:password@localhost/dbname', poolclass=QueuePool)

# 使用连接
with engine.connect() as connection:
    result = connection.execute("SELECT * FROM table_name")
    for row in result:
        print(row)

ORM 工具

在 Python 中，对象关系映射（Object-Relational Mapping, ORM）是一种流行的技术，它允许开发者以对象的形式操作数据库，而无需编写 SQL 语句。ORM（Object-Relational Mapping）是一种将数据库表转换为程序中对象的技术，使开发者能够以面向对象的方式处理数据库。

ORM 库	简介	特点	适用场景
Django ORM	Django 框架的一部分	无需编写 SQL，支持多数据库，丰富的查询构造器	Django 框架的 web 应用
SQLAlchemy	流行的独立 ORM 工具	灵活模型定义，强大查询，支持多数据库	适用于各种规模应用，特别是复杂数据库应用
Peewee	小巧但功能完整的 ORM 库	简单易用，学习曲线低	小型项目或简单 ORM 需求
Tortoise ORM	异步 ORM 库	支持异步操作，适合异步编程环境	需要高并发处理的现代 web 应用
Pony ORM	提供直观查询语言的 ORM 库	独特的查询语法，易于理解	寻求直观查询方式的开发者

基本的 ORM 操作

使用 ORM 工具时，开发者定义模型（即类），这些模型映射到数据库的表。在 Python 中，常见的 ORM 工具包括 SQLAlchemy 和 Django ORM。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()
engine = create_engine('sqlite:///example.db')
Session = sessionmaker(bind=engine)

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)

# 创建表
Base.metadata.create_all(engine)

# 使用会话进行数据库操作
session = Session()
new_user = User(name='Alice')
session.add(new_user)
session.commit()
session.close()

项目实战

为了将所学知识应用于实际，可以从开发一个简单的数据库应用开始。例如，创建一个图书管理系统，包括图书的添加、查询、更新和删除功能。

import sqlite3
from sqlite3 import Error

def create_connection(db_file):
    """ 创建一个数据库连接到 SQLite 数据库 """
    conn = None
    try:
        conn = sqlite3.connect(db_file)
        return conn
    except Error as e:
        print(e)

    return conn

def create_table(conn, create_table_sql):
    """ 创建表 """
    try:
        c = conn.cursor()
        c.execute(create_table_sql)
    except Error as e:
        print(e)

def main():
    database = r"pythonsqlite.db"

    sql_create_projects_table = """ CREATE TABLE IF NOT EXISTS projects (
                                        id integer PRIMARY KEY,
                                        name text NOT NULL,
                                        begin_date text,
                                        end_date text
                                    ); """

    sql_create_tasks_table = """CREATE TABLE IF NOT EXISTS tasks (
                                    id integer PRIMARY KEY,
                                    name text NOT NULL,
                                    priority integer,
                                    status_id integer NOT NULL,
                                    project_id integer NOT NULL,
                                    begin_date text NOT NULL,
                                    end_date text NOT NULL,
                                    FOREIGN KEY (project_id) REFERENCES projects (id)
                                );"""

    # 创建一个数据库连接
    conn = create_connection(database)

    # 创建表
    if conn is not None:
        create_table(conn, sql_create_projects_table)
        create_table(conn, sql_create_tasks_table)
    else:
        print("无法创建数据库连接。")

    # 插入数据
    project = ('Cool App with SQLite & Python', '2023-01-01', '2023-12-31')
    project_id = insert_project(conn, project)

    task_1 = ('Analyze the requirements of the app', 1, 1, project_id, '2023-01-01', '2023-01-02')
    task_2 = ('Confirm with user about the top priorities', 1, 1, project_id, '2023-01-03', '2023-01-05')

    # 插入任务
    insert_task(conn, task_1)
    insert_task(conn, task_2)

    # 查询数据
    print("1. 查询项目:")
    select_all_projects(conn)

    print("2. 查询任务:")
    select_all_tasks(conn)

    # 更新数据
    update_task(conn, (2, '2023-01-04', 2))

    # 删除数据
    delete_task(conn, 2)

    # 关闭连接
    conn.close()

def insert_project(conn, project):
    """
    创建一个新的项目
    :param conn:
    :param project:
    :return: project id
    """
    sql = ''' INSERT INTO projects(name,begin_date,end_date)
              VALUES(?,?,?) '''
    cur = conn.cursor()
    cur.execute(sql, project)
    conn.commit()
    return cur.lastrowid

def insert_task(conn, task):
    """
    创建一个新的任务
    :param conn:
    :param task:
    :return:
    """
    sql = ''' INSERT INTO tasks(name,priority,status_id,project_id,begin_date,end_date)
              VALUES(?,?,?,?,?,?) '''
    cur = conn.cursor()
    cur.execute(sql, task)
    conn.commit()
    return cur.lastrowid

def select_all_projects(conn):
    """
    查询所有项目
    :param conn: the Connection object
    :return:
    """
    cur = conn.cursor()
    cur.execute("SELECT * FROM projects")

    rows = cur.fetchall()

    for row in rows:
        print(row)

def select_all_tasks(conn):
    """
    查询所有任务
    :param conn: the Connection object
    :return:
    """
    cur = conn.cursor()
    cur.execute("SELECT * FROM tasks")

    rows = cur.fetchall()

    for row in rows:
        print(row)

def update_task(conn, task):
    """
    更新任务
    :param conn:
    :param task:
    :return: project id
    """
    sql = ''' UPDATE tasks
              SET priority = ? ,
                  end_date = ?
              WHERE id = ?'''
    cur = conn.cursor()
    cur.execute(sql, task)
    conn.commit()

def delete_task(conn, id):
    """
    删除一个任务通过任务 id
    :param conn:  Connection to the SQLite database
    :param id: id of the task
    :return:
    """
    sql = 'DELETE FROM tasks WHERE id=?'
    cur = conn.cursor()
    cur.execute(sql, (id,))
    conn.commit()

if __name__ == '__main__':
    main()