# app/database.py import pymysql from flask import current_app from contextlib import contextmanager class Database: """数据库连接管理类""" def __init__(self, app=None): self.app = app if app is not None: self.init_app(app) def init_app(self, app): """初始化数据库连接""" app.teardown_appcontext(self.teardown) @contextmanager def connection(self): """获取数据库连接上下文管理器""" conn = pymysql.connect( host=current_app.config['DB_HOST'], port=current_app.config['DB_PORT'], user=current_app.config['DB_USER'], password=current_app.config['DB_PASSWORD'], db=current_app.config['DB_NAME'], charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) try: yield conn finally: conn.close() def teardown(self, exception): """应用上下文销毁时执行""" pass # 创建数据库实例 db = Database() 讲解代码

<think>我们有一个数据库管理类 `Database`，它用于管理数据库连接。这个类被设计成可以与Flask应用集成。下面我将详细解释每一部分代码。 ### 1. 导入模块 ```python import pymysql from flask import current_app from contextlib import contextmanager ``` - `pymysql`: 用于连接MySQL数据库。 - `current_app`: 从Flask中导入，代表当前的应用实例，用于访问应用配置。 - `contextmanager`: 一个上下文管理器装饰器，用于创建支持`with`语句的上下文。 ### 2. 类定义 ```python class Database: """数据库连接管理类""" def __init__(self, app=None): self.app = app if app is not None: self.init_app(app) ``` - 初始化方法：如果传入`app`参数，则调用`init_app`方法进行初始化。 ### 3. 初始化应用 ```python def init_app(self, app): """初始化数据库连接""" app.teardown_appcontext(self.teardown) ``` - 将`teardown`方法注册为应用上下文销毁时的回调函数。当应用上下文结束时，会调用`teardown`方法（虽然目前`teardown`方法为空）。 ### 4. 连接上下文管理器 ```python @contextmanager def connection(self): """获取数据库连接上下文管理器""" conn = pymysql.connect( host=current_app.config['DB_HOST'], port=current_app.config['DB_PORT'], user=current_app.config['DB_USER'], password=current_app.config['DB_PASSWORD'], db=current_app.config['DB_NAME'], charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) try: yield conn finally: conn.close() ``` - 使用`@contextmanager`装饰器定义了一个上下文管理器。这样可以使用`with`语句来管理数据库连接。 - 在上下文管理器中，使用`current_app`的配置来创建数据库连接。注意，这里使用了`current_app`，所以这个方法必须在应用上下文中调用（例如在视图函数中）。 - 连接参数包括主机、端口、用户名、密码、数据库名，以及字符集（utf8mb4）和游标类型（返回字典格式的游标）。 - 在`try`块中，使用`yield conn`将连接对象提供给`with`语句块使用。当`with`语句块执行完毕或发生异常时，会执行`finally`块，关闭连接。 ### 5. 销毁方法 ```python def teardown(self, exception): """应用上下文销毁时执行""" pass ``` - 目前这个方法为空。通常，我们可能会在这里处理一些清理工作，但在这个例子中，因为每个连接都在使用后立即关闭，所以不需要额外的清理。 ### 6. 创建数据库实例 ```python db = Database() ``` - 创建一个`Database`的全局实例`db`，可以在应用的其他地方导入并使用。 ### 使用示例 在Flask应用中使用这个数据库连接管理器的示例： ```python from flask import Flask from app.database import db app = Flask(__name__) app.config.update( DB_HOST='localhost', DB_PORT=3306, DB_USER='root', DB_PASSWORD='password', DB_NAME='test' ) db.init_app(app) @app.route('/') def index(): with db.connection() as conn: with conn.cursor() as cursor: cursor.execute("SELECT * FROM users") result = cursor.fetchall() return str(result) ``` ### 注意事项 1. 这个`Database`类使用了应用上下文，因此不能在应用上下文之外使用`connection`方法（例如在应用启动之前）。 2. 每次调用`connection()`都会创建一个新的数据库连接，并在使用后关闭。对于频繁的数据库操作，可以考虑使用连接池来优化性能，但这里没有使用连接池。 ### 改进建议 1. 可以考虑使用连接池来管理数据库连接，避免频繁创建和关闭连接带来的开销。 2. 可以在`teardown`方法中增加一些日志记录或资源清理。 3. 可以考虑将连接作为应用上下文的扩展，使得在同一个应用上下文中可以复用连接（但要注意线程安全）。 ### 相关问题 §§相关问题§§: 1. 如何在Flask中实现数据库连接池？ 2. 使用上下文管理器管理数据库连接有什么好处？ 3. 如何确保数据库操作的事务性？ 4. 在多线程环境下，这个数据库连接管理类是否安全？</think>这段代码实现了一个 Flask 应用的数据库连接管理类。下面我将详细讲解代码结构和功能： ### 代码详解 #### 1. 导入依赖 ```python import pymysql from flask import current_app from contextlib import contextmanager ``` - `pymysql`：用于连接 MySQL 数据库 - `current_app`：访问当前 Flask 应用实例 - `contextmanager`：创建上下文管理器装饰器 #### 2. Database 类定义 ```python class Database: """数据库连接管理类""" def __init__(self, app=None): self.app = app if app is not None: self.init_app(app) ``` - 初始化方法：支持直接传入 Flask 应用实例初始化 - 如果提供了 app 参数，自动调用 `init_app()` 方法 #### 3. 应用初始化方法 ```python def init_app(self, app): """初始化数据库连接""" app.teardown_appcontext(self.teardown) ``` - 注册应用上下文销毁时的回调函数 - 使用 `teardown_appcontext` 确保资源正确释放 #### 4. 数据库连接上下文管理器 ```python @contextmanager def connection(self): """获取数据库连接上下文管理器""" conn = pymysql.connect( host=current_app.config['DB_HOST'], port=current_app.config['DB_PORT'], user=current_app.config['DB_USER'], password=current_app.config['DB_PASSWORD'], db=current_app.config['DB_NAME'], charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) try: yield conn finally: conn.close() ``` - 使用 `@contextmanager` 创建上下文管理器 - 从 Flask 应用配置中读取数据库连接参数 - `yield conn` 提供连接对象给 with 代码块使用 - `finally` 确保连接总是被关闭 - 使用 `DictCursor` 使查询结果以字典形式返回 #### 5. 上下文销毁处理 ```python def teardown(self, exception): """应用上下文销毁时执行""" pass ``` - 目前为空实现，可扩展添加清理逻辑 #### 6. 创建全局实例 ```python db = Database() ``` - 创建全局可用的数据库实例 - 可在应用中导入使用：`from app.database import db` ### 使用示例 在 Flask 视图函数中使用： ```python from .database import db @app.route('/users') def get_users(): with db.connection() as conn: with conn.cursor() as cursor: cursor.execute("SELECT * FROM users") users = cursor.fetchall() return jsonify(users) ``` ### 关键设计点 1. **上下文管理器模式**：确保数据库连接正确关闭 2. **延迟初始化**：在需要时才创建连接 3. **配置驱动**：从 Flask 应用配置获取连接参数 4. **线程安全**：每个请求使用独立连接 5. **资源管理**：使用 `finally` 保证连接关闭 6. **字典游标**：`DictCursor` 使结果更易处理 ### 改进建议 1. 添加连接池支持 2. 实现连接重试机制 3. 添加查询超时设置 4. 增加日志记录 5. 支持连接健康检查