当前位置：首页 > 文章列表 > 文章 > python教程 > Python数据库操作教程：CRUD实战详解

Python数据库操作教程：CRUD实战详解

2025-09-02 20:36:59 0浏览收藏

掌握Python数据库操作是提升开发效率的关键。本文详细讲解了如何使用Python进行数据库的CRUD操作，并着重介绍了使用`sqlite3`模块连接数据库、创建游标、执行SQL语句以及提交事务的完整流程。同时，强调了**参数化查询**在防止SQL注入攻击中的重要作用，并提供了实际代码示例。此外，还深入探讨了事务管理，确保数据的一致性和完整性，避免数据丢失或损坏。学习本文，你将能够安全高效地使用Python操作数据库，为你的项目保驾护航。

Python操作数据库需通过驱动建立连接并执行SQL，遵循连接、创建游标、执行SQL、提交事务、关闭连接的流程，使用参数化查询防SQL注入，结合try-except-finally管理事务确保数据一致性。

Python怎样操作数据库_Python数据库CRUD步骤解析

Python操作数据库的核心在于通过特定的数据库驱动（如sqlite3、psycopg2、mysql-connector-python等）建立连接，然后利用这个连接来执行SQL语句，完成数据的增、删、改、查（CRUD）操作。这过程通常涉及连接数据库、创建游标、执行SQL、提交事务（对数据进行修改时）以及关闭连接几个基本步骤。

解决方案

要使用Python操作数据库，我们通常会遵循一套相对标准的流程，这套流程基于Python的DB-API 2.0规范，虽然具体实现会因数据库类型和所选驱动而异，但核心逻辑是相通的。这里以Python内置的sqlite3模块为例，演示如何进行CRUD操作。

首先，你需要一个数据库文件。如果文件不存在，sqlite3会自动创建一个。

import sqlite3

# 1. 连接数据库
# 如果数据库文件不存在，会自动创建
try:
    conn = sqlite3.connect('my_database.db')
    print("数据库连接成功。")

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

    # 3. 创建表（如果不存在）- Create Schema
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL,
            age INTEGER
        )
    ''')
    conn.commit() # 提交事务，保存表结构更改
    print("表 'users' 创建或已存在。")

    # --- CRUD 操作示例 ---

    # C - Create (插入数据)
    # 使用参数化查询防止SQL注入
    user_data = [
        ('Alice', 'alice@example.com', 30),
        ('Bob', 'bob@example.com', 24),
        ('Charlie', 'charlie@example.com', 35)
    ]
    for name, email, age in user_data:
        try:
            cursor.execute("INSERT INTO users (name, email, age) VALUES (?, ?, ?)", (name, email, age))
            print(f"插入用户: {name}")
        except sqlite3.IntegrityError:
            print(f"用户 {name} ({email}) 已存在，跳过插入。")
    conn.commit() # 提交事务，保存数据更改

    # R - Read (查询数据)
    print("\n--- 查询所有用户 ---")
    cursor.execute("SELECT id, name, email, age FROM users")
    users = cursor.fetchall() # 获取所有结果
    for user in users:
        print(f"ID: {user[0]}, 姓名: {user[1]}, 邮箱: {user[2]}, 年龄: {user[3]}")

    print("\n--- 查询特定用户（年龄大于25）---")
    cursor.execute("SELECT name, email FROM users WHERE age > ?", (25,))
    older_users = cursor.fetchall()
    for user in older_users:
        print(f"姓名: {user[0]}, 邮箱: {user[1]}")

    # U - Update (更新数据)
    print("\n--- 更新用户数据 ---")
    new_email = 'alice_new@example.com'
    old_email = 'alice@example.com'
    cursor.execute("UPDATE users SET email = ? WHERE email = ?", (new_email, old_email))
    if cursor.rowcount > 0:
        print(f"用户 {old_email} 的邮箱已更新为 {new_email}。")
    else:
        print(f"未找到邮箱为 {old_email} 的用户进行更新。")
    conn.commit() # 提交事务

    # 再次查询确认更新
    print("\n--- 确认更新后的用户数据 ---")
    cursor.execute("SELECT name, email FROM users WHERE email = ?", (new_email,))
    updated_alice = cursor.fetchone()
    if updated_alice:
        print(f"更新后的Alice: 姓名: {updated_alice[0]}, 邮箱: {updated_alice[1]}")

    # D - Delete (删除数据)
    print("\n--- 删除用户数据 ---")
    user_to_delete = 'Bob'
    cursor.execute("DELETE FROM users WHERE name = ?", (user_to_delete,))
    if cursor.rowcount > 0:
        print(f"用户 {user_to_delete} 已被删除。")
    else:
        print(f"未找到用户 {user_to_delete} 进行删除。")
    conn.commit() # 提交事务

    # 再次查询确认删除
    print("\n--- 确认删除后的所有用户 ---")
    cursor.execute("SELECT name FROM users")
    remaining_users = cursor.fetchall()
    if remaining_users:
        print("当前剩余用户:", [user[0] for user in remaining_users])
    else:
        print("数据库中已无用户。")

except sqlite3.Error as e:
    print(f"数据库操作发生错误: {e}")
finally:
    # 4. 关闭连接
    if conn:
        conn.close()
        print("\n数据库连接已关闭。")

这段代码展示了一个完整的CRUD流程。需要注意的是，每次对数据库进行修改（INSERT, UPDATE, DELETE）后，都需要调用conn.commit()来保存更改。如果发生错误或需要撤销操作，可以使用conn.rollback()。

Python连接主流数据库有哪些常用库？如何选择合适的数据库驱动？

在Python的世界里，连接各种数据库的库非常丰富，这得益于Python DB-API 2.0规范的存在，它为数据库驱动提供了一套统一的接口。这使得我们在切换不同数据库时，核心的CRUD逻辑可以保持相对一致。

常用数据库连接库（驱动）：

SQLite: Python标准库内置了sqlite3模块。这是最方便的选项，无需额外安装，适用于小型应用、本地数据存储或开发测试。
PostgreSQL: psycopg2是PostgreSQL最流行、功能最强大的驱动。它支持大多数PostgreSQL特性，性能也很好。另一个选择是asyncpg，专为异步操作设计，性能极高。
MySQL: mysql-connector-python是Oracle官方提供的Python驱动。PyMySQL是另一个纯Python实现的MySQL客户端库，兼容DB-API 2.0，且在某些场景下可能更易于安装和使用。
SQL Server: pyodbc是一个通用的ODBC驱动，可以连接到任何支持ODBC的数据库，包括SQL Server。微软也提供了mssql-tools和相关的Python驱动。
Oracle: cx_Oracle（现在改名为python-oracledb）是Oracle官方的Python接口，提供了对Oracle数据库的全面支持。

如何选择合适的数据库驱动？

选择合适的数据库驱动，不仅仅是看它能连上数据库，更要考虑项目的实际需求和未来的发展。

目标数据库类型： 这是最直接的决定因素。如果你用的是PostgreSQL，那自然要看psycopg2；MySQL则可能在mysql-connector-python和PyMySQL之间选择。
性能要求： 对于高并发、大数据量的应用，驱动的性能至关重要。例如，异步驱动（如asyncpg）在某些场景下能提供显著的性能优势。
特性支持： 某些数据库有其独特的高级特性（如JSONB类型、地理空间数据、存储过程等），你需要确保所选驱动能够良好地支持这些特性。
社区活跃度和维护状态： 一个活跃的社区意味着遇到问题时更容易找到解决方案，库也会持续得到更新和维护，修复bug，增加新功能。
ORM框架的兼容性： 如果你打算使用ORM（如SQLAlchemy、Django ORM），那么你需要选择一个与ORM兼容且推荐的驱动。通常，ORM会指定或推荐其最佳配合的驱动。
易用性和文档： 良好的文档和清晰的API设计能大大降低开发难度和学习成本。
许可协议： 检查驱动的许可协议是否与你的项目兼容。大多数开源驱动都是MIT、BSD或Apache等宽松许可。

我个人在选择时，倾向于优先选择官方或社区广泛推荐的驱动，因为它们通常更稳定、功能更全面、社区支持更好。如果项目对性能有极致要求，我才会深入研究异步驱动或更底层的实现。

在Python中执行SQL查询时，如何有效防止SQL注入攻击？

SQL注入是数据库安全领域一个老生常谈但又极其重要的议题。它发生在应用程序将用户输入直接拼接进SQL查询字符串，导致恶意用户可以通过输入特定的SQL片段来改变查询的意图，从而窃取、篡改数据甚至完全控制数据库。防止SQL注入，在Python中，核心且最有效的方法是使用参数化查询（Parameterized Queries），也称为预处理语句（Prepared Statements）。

参数化查询的原理：

参数化查询的原理是SQL代码和查询参数是分开传递给数据库的。数据库会先解析SQL语句的结构，确定要执行的操作，然后将参数值绑定到相应的位置。这样，即使参数中包含恶意的SQL片段，数据库也只会将其视为普通的数据值，而不是可执行的SQL代码。

Python中的实现：

几乎所有的Python数据库驱动都支持参数化查询。它们通常使用占位符（placeholder）来表示参数。占位符的样式会因驱动而异：

sqlite3和psycopg2 (PostgreSQL): 使用问号 ? 作为位置占位符。

import sqlite3

conn = sqlite3.connect('my_database.db')
cursor = conn.cursor()

user_input_name = "Robert'); DROP TABLE users;--" # 恶意输入
user_input_email = "robert@example.com"

# 错误示例：直接拼接字符串，易受SQL注入攻击
# cursor.execute(f"INSERT INTO users (name, email) VALUES ('{user_input_name}', '{user_input_email}')")

# 正确示例：使用参数化查询
cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", (user_input_name, user_input_email))
conn.commit()
print("使用参数化查询，恶意输入被当作普通数据处理。")
conn.close()

在这个例子中，即使user_input_name包含DROP TABLE users，数据库也只会把它当成一个长字符串，而不是执行删除表的命令。

mysql-connector-python和PyMySQL (MySQL): 通常使用%s作为位置占位符，或者%(name)s作为命名占位符。

# 假设你已经连接了MySQL数据库
# import mysql.connector
# conn = mysql.connector.connect(...)
# cursor = conn.cursor()

# user_input_name = "Robert'); DROP TABLE users;--"
# user_input_email = "robert@example.com"

# 使用位置占位符
# cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)", (user_input_name, user_input_email))

# 使用命名占位符（更具可读性）
# cursor.execute("INSERT INTO users (name, email) VALUES (%(name)s, %(email)s)",
#                {'name': user_input_name, 'email': user_input_email})
# conn.commit()

注意事项：

永远不要直接用f-string或字符串格式化来构建SQL查询。 这是导致SQL注入的罪魁祸首。
不仅仅是INSERT语句，SELECT、UPDATE、DELETE等所有涉及用户输入的查询都必须使用参数化。
LIKE子句的参数化： 如果在LIKE子句中使用%或_通配符，你需要将这些通配符添加到参数值中，而不是添加到SQL字符串中。
```
search_term = "%Alice%" # 通配符是参数值的一部分
cursor.execute("SELECT name FROM users WHERE name LIKE ?", (search_term,))
```
表名或列名的动态化： 如果需要动态指定表名或列名，参数化查询是无能为力的，因为它只对值进行参数化。在这种情况下，你需要对表名/列名进行严格的白名单校验，确保它们只包含预期的、安全的字符串，或者使用ORM的API来处理。

通过始终坚持使用参数化查询，可以极大地提高应用程序的数据库安全性，有效抵御SQL注入攻击。

使用Python操作数据库时，如何管理事务以确保数据一致性？

事务（Transaction）是数据库管理系统（DBMS）执行过程中的一个逻辑工作单元，它包含一系列操作，这些操作要么全部成功提交，要么全部失败回滚。事务的核心目的是确保数据的一致性（Consistency）、原子性（Atomicity）、隔离性（Isolation）和持久性（Durability），通常称为ACID特性。在Python中操作数据库时，正确管理事务对于维护数据的完整性和可靠性至关重要。

Python中的事务管理：

大多数Python数据库驱动都遵循DB-API 2.0规范，提供了明确的事务管理机制：

conn.commit()：提交事务 当一系列数据库操作（如多个INSERT、UPDATE、DELETE）都成功执行，并且你希望将这些更改永久保存到数据库时，需要调用conn.commit()。这会结束当前的事务，并使所有更改可见。
conn.rollback()：回滚事务 如果在事务执行过程中发生了错误，或者某个操作失败了，你可能不希望之前的所有更改生效。这时可以调用conn.rollback()，它会将数据库恢复到事务开始前的状态，撤销所有未提交的更改。
默认行为：
- 许多数据库（如PostgreSQL、MySQL）在执行单个INSERT/UPDATE/DELETE语句后，如果没有明确的BEGIN TRANSACTION，可能会默认开启一个隐式事务。
- sqlite3模块默认是自动提交（autocommit）模式，这意味着每个execute()调用都会立即提交。为了进行显式事务管理，你需要禁用自动提交或确保在连接时就进入事务模式。不过，更常见和推荐的做法是，在进行修改操作后，统一调用conn.commit()来提交事务。如果你没有调用conn.commit()，那么对数据库的修改将不会被保存。

事务管理的典型场景和实践：

假设我们有一个转账操作，需要从一个账户扣钱，同时给另一个账户加钱。这两个操作必须作为一个整体成功或失败。

import sqlite3

conn = sqlite3.connect('bank.db')
cursor = conn.cursor()

# 创建账户表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS accounts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        balance REAL NOT NULL DEFAULT 0.0
    )
''')
conn.commit()

# 初始化账户（如果不存在）
def ensure_account(name, initial_balance):
    cursor.execute("SELECT id FROM accounts WHERE name = ?", (name,))
    if not cursor.fetchone():
        cursor.execute("INSERT INTO accounts (name, balance) VALUES (?, ?)", (name, initial_balance))
        conn.commit()
        print(f"创建账户: {name}, 余额: {initial_balance}")

ensure_account('Alice', 1000.0)
ensure_account('Bob', 500.0)

def transfer_money(sender_name, receiver_name, amount):
    if amount <= 0:
        print("转账金额必须大于零。")
        return False

    try:
        # 开始事务
        # 在sqlite3中，只要不调用commit，所有的修改都会累积在一个事务中
        # 显式地，你也可以用 BEGIN TRANSACTION

        # 1. 检查发送方余额
        cursor.execute("SELECT balance FROM accounts WHERE name = ?", (sender_name,))
        sender_balance = cursor.fetchone()[0]

        if sender_balance < amount:
            print(f"账户 {sender_name} 余额不足。")
            conn.rollback() # 余额不足，回滚事务
            return False

        # 2. 从发送方账户扣款
        cursor.execute("UPDATE accounts SET balance = balance - ? WHERE name = ?", (amount, sender_name))

        # 模拟一个可能发生的错误，例如网络中断、数据库死锁等
        # if sender_name == 'Alice' and amount == 200:
        #    raise ValueError("模拟转账失败！")

        # 3. 给接收方账户加款
        cursor.execute("UPDATE accounts SET balance = balance + ? WHERE name = ?", (amount, receiver_name))

        # 所有操作成功，提交事务
        conn.commit()
        print(f"成功从 {sender_name} 转账 {amount} 到 {receiver_name}。")
        return True

    except Exception as e:
        print(f"转账过程中发生错误: {e}")
        conn.rollback() # 发生任何错误，回滚事务
        return False

# 尝试转账
print("\n--- 首次转账 ---")
transfer_money('Alice', 'Bob', 100.0)

# 查看账户余额
print("\n--- 账户余额 ---")
cursor.execute("SELECT name, balance FROM accounts")
for account in cursor.fetchall():
    print(f"账户: {account[0]}, 余额: {account[1]}")

# 尝试一个会失败的转账（余额不足）
print("\n--- 尝试失败转账（余额不足） ---")
transfer_money('Bob', 'Alice', 1000.0)

# 再次查看账户余额，确认未发生更改
print("\n--- 账户余额（失败转账后） ---")
cursor.execute("SELECT name, balance FROM accounts")
for account in cursor.fetchall():
    print(f"账户: {account[0]}, 余额: {account[1]}")

conn.close()

关键点：

try...except...finally 结构： 这是处理事务的黄金法则。在try块中执行所有数据库操作，如果一切顺利，在try块的末尾调用conn.commit()。如果try块中发生任何异常，在except块中捕获并调用conn.rollback()来撤销所有更改。finally块通常用于确保连接被关闭，无论事务成功与否。
上下文管理器（with语句）： 对于资源管理，Python的with语句是一个非常优雅的解决方案。它能确保在代码块执行完毕后，资源（如文件、数据库连接）被正确关闭。虽然DB-API 2.0本身没有强制要求连接对象支持with语句，但许多高级库和ORM框架（如SQLAlchemy）都提供了这样的功能，或者你可以自己实现一个上下文管理器。
```
# 示例：使用with语句管理连接（如果conn对象支持）
# 在sqlite3中，连接对象本身不支持with，但我们可以通过自定义上下文管理器或使用ORM来实现
# 对于sqlite3，更常见的模式是：
# with sqlite3.connect('my_database.db') as conn:
#     cursor = conn.cursor()
#     # ... 执行操作 ...
#     # conn.commit() 在with块结束时会自动提交，如果没发生
```