Python: 数据库操作
This commit is contained in:
parent
e533c28f9d
commit
e13a2c79cf
GPG Key ID:
6666822800008000
527
Tech/programming-language/Python/进阶/数据库操作.md
Normal file
527
Tech/programming-language/Python/进阶/数据库操作.md
Normal file
@ -0,0 +1,527 @@
|
||||
## Python DB-API 2.0 规范概述
|
||||
|
||||
**Python DB-API 2.0 规范是一个重要的标准，它为 Python 程序与数据库间的交互提供了一致的接口。**这个规范的目标是简化数据库编程，同时保持足够的灵活性来支持不同的数据库系统。
|
||||
|
||||
DB-API 2.0 规范定义在 Python 标准库的 [PEP 249](https://peps.python.org/pep-0249/) 文档中。它规定了数据库驱动（或数据库模块）应遵循的接口标准，以便程序员可以使用一致的编程风格来访问不同的数据库系统。
|
||||
|
||||
### 核心组件
|
||||
|
||||
| 组件 | 描述 | 关键方法或属性 |
|
||||
| -------------------------- | ------------------------ | ------------------------------------------------------------ |
|
||||
| **连接对象（Connection）** | 代表数据库的连接。 | - `connect()`: 连接数据库。<br>- `commit()`: 提交当前事务。<br>- `rollback()`: 回滚当前事务。<br>- `close()`: 关闭连接。 |
|
||||
| **游标对象（Cursor）** | 用于执行查询和获取结果。 | - `execute(sql, [parameters])`: 执行 SQL 语句。<br>- `executemany(sql, seq_of_parameters)`: 执行相同的 SQL 语句多次。<br>- `fetchone()`: 获取结果集的下一行。<br>- `fetchmany(size)`: 获取结果集的下几行。<br>- `fetchall()`: 获取结果集中的所有行。<br>- `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 语句：
|
||||
|
||||
```python
|
||||
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()` 函数创建的。
|
||||
- 这个函数通常需要数据库特定的参数，如主机名、数据库名、用户名和密码。
|
||||
|
||||
```python
|
||||
import sqlite3
|
||||
|
||||
# 创建连接到 SQLite 数据库文件 'example.db'
|
||||
conn = sqlite3.connect('example.db')
|
||||
```
|
||||
|
||||
#### 关键特性和方法
|
||||
|
||||
1. **事务处理**：
|
||||
- `commit()`: 提交当前事务。在执行如 INSERT、UPDATE、DELETE 等操作后，需要调用此方法以确保更改被保存。
|
||||
- `rollback()`: 回滚当前事务。在遇到错误或需要撤销更改时使用。
|
||||
2. **连接管理**：
|
||||
- `close()`: 关闭连接。关闭后，连接对象和其下所有的游标将不可用。
|
||||
3. **错误处理**：
|
||||
- 连接操作可能会抛出 DB-API 定义的异常，如 `InterfaceError` 或 `DatabaseError`。
|
||||
|
||||
```python
|
||||
try:
|
||||
# 执行数据库操作
|
||||
conn.commit() # 提交事务
|
||||
except Exception as e:
|
||||
conn.rollback() # 回滚事务
|
||||
raise e
|
||||
finally:
|
||||
conn.close() # 关闭连接
|
||||
```
|
||||
|
||||
Python 进行数据库操作务必管理好数据库连接的生命周期。使用完毕后应及时关闭，避免资源泄露。建议使用 Python 的上下文管理器（`with` 语句）可以自动管理连接的开启和关闭。
|
||||
|
||||
### 游标（Cursor）
|
||||
|
||||
游标是通过连接对象创建的，用于执行 SQL 命令和处理查询结果。
|
||||
|
||||
#### 创建游标
|
||||
|
||||
- 通过连接对象的 `cursor()` 方法创建。
|
||||
- 可以创建多个游标，用于执行不同的数据库操作。
|
||||
|
||||
```python
|
||||
cursor = conn.cursor() # 创建游标
|
||||
```
|
||||
|
||||
#### 关键特性和方法
|
||||
|
||||
1. **执行 SQL 命令**：
|
||||
- `execute(sql, [parameters])`: 执行单个 SQL 命令。
|
||||
- `executemany(sql, seq_of_parameters)`: 执行相同 SQL 命令多次，但使用不同的参数。
|
||||
2. **结果处理**：
|
||||
- `fetchone()`: 返回结果集的下一行。
|
||||
- `fetchmany(size)`: 返回结果集的下几行。
|
||||
- `fetchall()`: 返回结果集中的所有行。
|
||||
3. **元数据获取**：
|
||||
- 游标对象提供了诸如 `description` 属性，可以用来获取查询结果的列信息。
|
||||
4. **游标管理**：
|
||||
- `close()`: 关闭游标。关闭后，游标将不可用。
|
||||
|
||||
```python
|
||||
# 执行 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），结果通常是影响的行数。
|
||||
|
||||
```python
|
||||
# 执行一个查询
|
||||
cursor.execute('SELECT * FROM some_table')
|
||||
|
||||
# 或者执行一个更新
|
||||
cursor.execute('UPDATE some_table SET column = value WHERE condition')
|
||||
```
|
||||
|
||||
### 处理结果
|
||||
|
||||
- 使用 `fetchone()`、`fetchmany(size)` 或 `fetchall()` 方法获取查询结果。
|
||||
- `fetchone()` 返回单行，`fetchmany(size)` 返回指定数量的行，`fetchall()` 返回所有行。
|
||||
|
||||
```python
|
||||
# 执行查询
|
||||
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` 等）来识别和响应特定的错误情况。
|
||||
|
||||
```python
|
||||
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 注入攻击。
|
||||
|
||||
```python
|
||||
# 错误的做法：直接字符串拼接
|
||||
unsafe_sql = "SELECT * FROM users WHERE name = '" + user_input + "'"
|
||||
|
||||
# 正确的做法：使用参数化查询
|
||||
cursor.execute("SELECT * FROM users WHERE name = %s", (user_input,))
|
||||
```
|
||||
|
||||
#### 使用参数而非字符串拼接
|
||||
|
||||
在构建 SQL 语句时，应优先考虑使用参数化查询。
|
||||
|
||||
```python
|
||||
user_id = 5
|
||||
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
|
||||
```
|
||||
|
||||
### 事务管理
|
||||
|
||||
### 基本概念（ACID 特性）
|
||||
|
||||
事务是数据库操作的基本单位，它应满足 ACID 特性，即原子性（Atomicity）、一致性（Consistency）、隔离性（Isolation）、持久性（Durability）。
|
||||
|
||||
- **原子性**：确保所有操作要么全部完成，要么全部不执行。
|
||||
- **一致性**：事务完成时，数据必须处于一致状态。
|
||||
- **隔离性**：多个事务同时进行时，一个事务的操作不应影响其他事务。
|
||||
- **持久性**：一旦事务提交，其结果就永久保存在数据库中。
|
||||
|
||||
#### 在 Python 中进行事务控制
|
||||
|
||||
在 Python 中，事务控制通常由连接对象（Connection）来管理。使用 `commit()` 和 `rollback()` 方法来控制事务的提交和回滚。
|
||||
|
||||
```python
|
||||
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`）来实现数据库连接池。连接池管理着一组连接，当需要时分配给请求者，使用完毕后返回池中。
|
||||
|
||||
```python
|
||||
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。
|
||||
|
||||
```python
|
||||
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()
|
||||
```
|
||||
|
||||
## 项目实战
|
||||
|
||||
为了将所学知识应用于实际，可以从开发一个简单的数据库应用开始。例如，创建一个图书管理系统，包括图书的添加、查询、更新和删除功能。
|
||||
|
||||
```python
|
||||
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()
|
||||
```
|
||||
Loading…
Reference in New Issue
Block a user