基于PyMySQL实现应用层字段加密：保护敏感数据的Python实战方案

1. 项目概述：为什么我们需要在应用层做字段加密？

最近在做一个涉及用户敏感信息的项目，比如身份证号、手机号、家庭住址这些，数据最终要存到MySQL里。甲方爸爸和合规部门的要求很明确：这些敏感字段在数据库里不能是明文。一开始，我理所当然地想到了MySQL自带的加密函数，比如AES_ENCRYPT，直接在SQL里搞定，感觉挺省事。但实际踩坑后发现，这条路问题不少。首先，密钥管理麻烦，得在数据库配置或SQL里写死，权限控制不好做；其次，一旦数据导出来，或者有数据库备份，密文和密钥可能一起泄露，风险并没降低；最重要的是，如果运维同学或者有数据库直接访问权限的人想查，他们还是能看到明文，这达不到“库中无明文”的真正要求。

所以，我们把目光投向了应用层加密，也就是在数据进入数据库之前，在Python代码里就完成加密。PyMySQL作为Python连接MySQL的主流驱动之一，自然成了我们的操作入口。这个方案的核心思想是“端到端加密”：数据在离开应用的那一刻就已经是密文，数据库只是作为一个“盲存储”的容器，即使数据库被拖库，或者DBA直接查询，没有应用层的密钥也无法解密。这才能真正满足对敏感数据的保护需求。今天要聊的，就是基于PyMySQL，如何实现一套稳健、易用的字段级加密方案，而不是简单调用几个加密函数。

2. 整体设计与核心思路拆解

2.1 方案选型：应用层加密 vs 数据库层加密

为什么坚定地选择应用层加密？这背后是一系列权衡。数据库层加密（如TDE、列加密）对应用透明，但通常保护的是“静止数据”，即存储在磁盘上的数据文件。对于拥有数据库权限的用户，数据在内存中或查询结果里依然是明文。而字段级加密，尤其是在应用层实现，可以做到“使用中数据”也是密文。我们的目标是：让敏感数据在数据库的整个生命周期（存储、传输、备份）中，都以密文形式存在，只有经过授权的应用服务，在内存中处理时才会短暂解密。

基于PyMySQL，我们有几种实现路径：

1. SQL拼接法：在构造INSERT或UPDATE语句时，手动调用Python的加密库（如cryptography）对值进行加密，然后将密文的字节串或Base64编码后的字符串拼接到SQL中。这种方法最直接，但代码侵入性强，容易遗漏，且加解密逻辑散落在各处，难以维护。
2. ORM拦截法：如果使用SQLAlchemy等ORM框架，可以利用其事件监听机制（如before_flush,before_update）在数据提交前自动加密，查询后自动解密。这是比较优雅的方式，但对项目技术栈有要求。
3. 驱动层包装法：在PyMySQL的游标（Cursor）层面进行拦截。通过继承或包装PyMySQL的Cursor类，覆写execute和fetch*系列方法，在SQL执行前对参数中的特定字段进行加密，在获取结果后对特定列进行解密。这种方法对业务代码几乎零侵入，只需要更换一下连接或游标的使用方式，是本次重点讨论的方案。

我们选择了驱动层包装法。它的优势在于将加解密逻辑集中在一个地方，与业务逻辑解耦，无论是使用原生SQL还是简单ORM（如自己封装的DB类），都能无缝接入。关键在于如何精准地识别哪些字段需要加解密。

2.2 核心组件与加密策略定义

一个完整的字段级加密方案，需要明确以下几个核心部分：

1. 加密算法与模式选择：

   • 算法：AES（高级加密标准）是目前对称加密的主流选择，兼顾安全与性能。
   • 密钥长度：使用AES-256-GCM或AES-256-CBC。AES-256提供更强的安全性。GCM模式是认证加密模式，能同时提供保密性和完整性校验，且支持关联数据（AAD），更为现代和安全。CBC模式则需要单独处理填充和初始化向量（IV），并需结合HMAC来保证完整性，稍显繁琐但兼容性极广。
   • 初始化向量（IV）：为了确保同样的明文每次加密后产生不同的密文，必须使用随机且唯一的IV。对于CBC模式，IV需要随密文一起存储；对于GCM模式，除了IV（在GCM中常称为nonce），还有一个认证标签（Tag）也需要存储。

2. 密钥管理：

   • 这是安全的核心。绝对禁止将密钥硬编码在代码或配置文件中。
   • 推荐使用环境变量、或专业的密钥管理服务（KMS）来获取加密主密钥。应用启动时从安全的位置读取。
   • 在实际操作中，我们可以使用一个固定的主密钥，或者通过KMS生成一个数据加密密钥（DEK），再用主密钥加密DEK（形成加密的DEK，即EDEK）存储。每次加密时使用DEK。这里为了简化，我们先演示使用一个从环境变量获取的固定密钥。

3. 字段识别规则：

   • 我们需要一种方式来告诉包装器，哪张表的哪个字段需要加密。可以通过配置字典来实现，例如：{‘users’: [‘id_card’, ‘mobile’, ‘address’], ‘bank_cards’: [‘card_number’, ‘cvv’]}。
   • 在游标中解析SQL，虽然复杂但可以做到精准匹配。一个更实用的简化方案是：我们约定，所有需要加密的字段，在传入参数时，其值是一个特殊的标记对象（如一个字典{‘__encrypted__’: True, ‘value’: raw_value}），或者在参数名上做约定（如以_enc后缀结尾）。在包装器里，我们检测到这个标记，就对raw_value进行加密，并用密文替换原值。查询解密时，则根据结果集的元数据（列名）判断是否需要解密。

为了平衡安全性与实现的复杂度，我们下面的实操将采用AES-256-GCM算法，密钥从环境变量获取，并通过一个字段映射配置来识别需要加解密的表字段。同时，我们会将IV和认证标签与密文一起，组合成一个字符串存储到数据库的TEXT或BLOB字段中。

3. 核心工具准备与加密模块实现

3.1 环境与依赖安装

首先，确保你的Python环境（建议3.7+）并安装必要的库。除了PyMySQL，我们主要依赖cryptography这个强大的密码学库。

pip install pymysql cryptography

cryptography库提供了工业级的、安全的密码学原语实现，比Python自带的hashlib和早期的Crypto库更受推荐。

3.2 实现安全的加解密工具类

我们先创建一个独立的模块（如crypto_util.py），实现核心的加密和解密功能。这里采用AES-256-GCM模式。

# crypto_util.py import os import base64 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives import padding from cryptography.hazmat.backends import default_backend from typing import Union class FieldCrypto: """ 字段级加密工具类，使用 AES-256-GCM 模式。 密文格式：base64(iv + ciphertext + tag)。IV长度为12字节，Tag长度为16字节。 """ def __init__(self, key: Union[str, bytes]): """ 初始化加密器。 :param key: 加密密钥。如果是字符串，将使用UTF-8编码并确保长度为32字节（256位）。 强烈建议从环境变量等安全位置获取。 """ if isinstance(key, str): key = key.encode('utf-8') # 确保密钥长度为32字节 (AES-256) if len(key) != 32: # 如果长度不对，可以使用HKDF或SHA256派生，但这里简单抛错提示 raise ValueError(f"密钥必须为32字节长度，当前为{len(key)}字节。请检查密钥来源。") self.key = key self.iv_length = 12 # GCM推荐的非重复随机数长度 self.tag_length = 16 # GCM认证标签长度 def encrypt(self, plaintext: str) -> str: """加密明文字符串，返回Base64编码的完整密文包（含IV和Tag）。""" if plaintext is None: return None # 生成随机IV iv = os.urandom(self.iv_length) # 构建加密器 encryptor = Cipher( algorithms.AES(self.key), modes.GCM(iv), backend=default_backend() ).encryptor() # 加密并生成Tag plaintext_bytes = plaintext.encode('utf-8') ciphertext = encryptor.update(plaintext_bytes) + encryptor.finalize() tag = encryptor.tag # 组合 IV + Ciphertext + Tag combined = iv + ciphertext + tag # 返回Base64字符串，便于存储到文本字段 return base64.b64encode(combined).decode('utf-8') def decrypt(self, encrypted_b64: str) -> str: """解密Base64编码的密文包，返回明文字符串。""" if encrypted_b64 is None: return None try: combined = base64.b64decode(encrypted_b64) except Exception: # 如果无法解码，可能不是加密字段，直接返回原值（或根据策略处理） # 这里为了安全，我们选择抛出一个明确的异常 raise ValueError("提供的字符串不是有效的Base64编码密文格式") # 拆分 IV, Ciphertext, Tag iv = combined[:self.iv_length] tag = combined[-self.tag_length:] ciphertext = combined[self.iv_length:-self.tag_length] # 构建解密器 decryptor = Cipher( algorithms.AES(self.key), modes.GCM(iv, tag), backend=default_backend() ).decryptor() # 解密 plaintext_bytes = decryptor.update(ciphertext) + decryptor.finalize() return plaintext_bytes.decode('utf-8') # 示例：从环境变量获取密钥（生产环境务必如此） import os SECRET_KEY = os.environ.get('DB_FIELD_ENCRYPTION_KEY') if not SECRET_KEY: # 仅为演示，生产环境必须设置环境变量 SECRET_KEY = 'this-is-a-32-byte-long-secret-key-here!' # 32字节 crypto_util = FieldCrypto(SECRET_KEY) # 简单测试 if __name__ == '__main__': test_text = "130123456781234567" encrypted = crypto_util.encrypt(test_text) print(f"密文: {encrypted}") decrypted = crypto_util.decrypt(encrypted) print(f"解密后: {decrypted}") assert test_text == decrypted

注意：上面的SECRET_KEY在示例中是硬编码的，这只是为了演示。在实际生产环境中，你必须通过环境变量、或从安全的密钥管理服务中动态获取密钥。密钥的泄露意味着所有加密数据都可能被解密。

3.3 定义字段映射配置

我们需要一个配置来告知系统哪些字段需要被处理。创建一个配置模块（如config.py）或直接在主逻辑中定义。

# encryption_config.py ENCRYPTION_CONFIG = { # ‘表名’: [‘字段名1‘， ’字段名2‘, ...] 'users': ['id_card', 'mobile', 'email', 'real_name'], 'orders': ['recipient_phone', 'recipient_address'], 'medical_records': ['patient_id_number', 'diagnosis_details'] }

这个配置将用于指导我们的PyMySQL游标包装器，在执行SQL时识别和转换目标字段。

4. 实现PyMySQL游标包装器

这是整个方案的核心。我们将创建一个自定义的游标类，继承自pymysql.cursors.Cursor，并覆写关键方法。

4.1 包装器类结构设计

我们的EncryptedCursor需要做两件事：

1. 在执行execute前：分析传入的参数（可能是字典或元组），根据ENCRYPTION_CONFIG和当前执行的SQL（通过解析表名），对参数中对应字段的值进行加密。
2. 在fetchone或fetchall后：根据返回结果集的列名，判断哪些列是加密字段，并对这些列的值进行解密。

难点在于准确地将参数与SQL中的占位符匹配，并识别出表名。一个相对稳健的简化策略是：我们假设使用命名占位符（%(name)s）并且参数是字典形式。这样我们可以直接通过字典键名（即字段名）来判断是否需要加密。

# encrypted_cursor.py import pymysql.cursors import re from crypto_util import crypto_util # 导入之前实现的加密工具 from encryption_config import ENCRYPTION_CONFIG # 导入加密配置 class EncryptedCursor(pymysql.cursors.Cursor): """ 支持字段级加密的PyMySQL游标包装器。 假设SQL使用命名占位符（%(name)s），且参数为字典。 """ def _get_table_name_from_sql(self, sql: str) -> str: """ 粗糙地从SQL语句中提取表名。 仅适用于简单的 INSERT INTO table_name, UPDATE table_name, SELECT ... FROM table_name 语句。 生产环境可能需要更复杂的SQL解析器。 """ sql_upper = sql.strip().upper() # 移除注释等（简单处理） sql_upper = re.sub(r'--.*$', '', sql_upper, flags=re.MULTILINE) sql_upper = re.sub(r'/\*.*?\*/', '', sql_upper, flags=re.DOTALL) table_name = None # 匹配 INSERT INTO `table` / INSERT INTO table insert_match = re.search(r'INSERT\s+INTO\s+[`"]?(\w+)[`"]?', sql_upper, re.IGNORECASE) if insert_match: table_name = insert_match.group(1).lower() # 匹配 UPDATE `table` / UPDATE table update_match = re.search(r'UPDATE\s+[`"]?(\w+)[`"]?', sql_upper, re.IGNORECASE) if update_match: table_name = update_match.group(1).lower() # 匹配 SELECT ... FROM `table` / FROM table # 这里非常简陋，仅处理单表简单查询 from_match = re.search(r'FROM\s+[`"]?(\w+)[`"]?', sql_upper, re.IGNORECASE) if from_match and not table_name: table_name = from_match.group(1).lower() return table_name def _encrypt_parameters(self, sql: str, params: dict) -> dict: """根据SQL和配置，加密参数字典中需要加密的字段值。""" if not params or not isinstance(params, dict): return params table_name = self._get_table_name_from_sql(sql) if not table_name or table_name not in ENCRYPTION_CONFIG: # 如果没提取到表名或该表不在加密配置中，直接返回原参数 return params encrypted_fields = ENCRYPTION_CONFIG[table_name] encrypted_params = params.copy() for field in encrypted_fields: if field in encrypted_params and encrypted_params[field] is not None: raw_value = encrypted_params[field] # 只加密字符串类型，如果是其他类型需要先转字符串（根据业务决定） if isinstance(raw_value, str): encrypted_params[field] = crypto_util.encrypt(raw_value) # 也可以处理数字类型，但需统一转换为字符串格式 # elif isinstance(raw_value, (int, float)): # encrypted_params[field] = crypto_util.encrypt(str(raw_value)) else: # 非字符串类型，可以选择不加密、记录日志或抛异常 # 这里我们选择原样保留，但记录警告（实际项目应用日志库） print(f"Warning: Field '{field}' in table '{table_name}' has non-string value, skipped encryption.") return encrypted_params def _decrypt_result_row(self, row: tuple, description: tuple) -> tuple: """ 解密查询结果的一行数据。 :param row: 原始行数据（元组） :param description: cursor.description，包含列信息 :return: 解密后的行数据（元组） """ if not row or not description: return row # 获取当前查询涉及的表名（这里需要额外处理，因为游标可能不知道表名） # 一个变通方法：在_execute方法中，将表名暂存到游标属性中，但多表查询会复杂化。 # 更实用的方法：根据列名判断。我们约定加密字段的列名在配置中唯一，或者携带特殊前缀/后缀。 # 这里采用一个简化方案：假设配置中的字段名在所有表中是唯一的，或者我们通过上下文知道表名。 # 由于这是一个复杂点，我们在下一个版本优化。此处先实现一个基础版：如果列名在全局加密字段集合中，则尝试解密。 all_encrypted_fields = set() for fields in ENCRYPTION_CONFIG.values(): all_encrypted_fields.update(fields) decrypted_values = list(row) for idx, col_desc in enumerate(description): col_name = col_desc[0] # 列名 if col_name in all_encrypted_fields and decrypted_values[idx] is not None: # 尝试解密 try: decrypted_values[idx] = crypto_util.decrypt(decrypted_values[idx]) except (ValueError, Exception) as e: # 解密失败，可能该列不是加密数据，或数据已损坏。保留原值并记录错误。 print(f"Warning: Failed to decrypt column '{col_name}': {e}") # 生产环境应使用日志记录器 return tuple(decrypted_values) def execute(self, query, args=None): """重写execute方法，在执行前加密参数。""" encrypted_args = args if isinstance(args, dict): # 只有参数是字典时才进行加密处理 encrypted_args = self._encrypt_parameters(query, args) # 调用父类方法执行SQL return super().execute(query, encrypted_args) def executemany(self, query, args): """重写executemany方法，用于批量插入/更新。""" if args and isinstance(args[0], dict): encrypted_args = [] for arg_set in args: encrypted_args.append(self._encrypt_parameters(query, arg_set)) args = encrypted_args return super().executemany(query, args) def fetchone(self): """重写fetchone，获取数据后解密。""" row = super().fetchone() if row: return self._decrypt_result_row(row, self.description) return row def fetchall(self): """重写fetchall，获取所有数据后逐行解密。""" rows = super().fetchall() if rows: return [self._decrypt_result_row(row, self.description) for row in rows] return rows def fetchmany(self, size=None): """重写fetchmany。""" rows = super().fetchmany(size) if rows: return [self._decrypt_result_row(row, self.description) for row in rows] return rows

4.2 创建加密连接类

为了方便使用，我们还可以创建一个返回加密游标的连接类。

# encrypted_connection.py import pymysql from encrypted_cursor import EncryptedCursor class EncryptedConnection(pymysql.Connection): """ 使用EncryptedCursor作为默认游标的连接类。 """ def cursor(self, cursor=None): """ 重写cursor方法，默认返回我们的EncryptedCursor。 """ if cursor: return super().cursor(cursor) return super().cursor(EncryptedCursor) # 使用示例 def get_encrypted_connection(): conn = EncryptedConnection( host='localhost', user='your_username', password='your_password', database='your_database', charset='utf8mb4', cursorclass=EncryptedCursor # 这里指定游标类，或在EncryptedConnection中已重写 ) return conn

5. 完整实操流程与测试

现在，我们将上述模块组合起来，进行一个从建表到增删改查的全流程测试。

5.1 数据库表结构准备

假设我们有一张users表，其中包含需要加密的字段。

CREATE TABLE `users` ( `id` INT PRIMARY KEY AUTO_INCREMENT, `username` VARCHAR(50) NOT NULL UNIQUE, `id_card` TEXT COMMENT '加密存储的身份证号', `mobile` TEXT COMMENT '加密存储的手机号', `email` TEXT COMMENT '加密存储的邮箱', `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

注意，加密后的数据是二进制经过Base64编码的字符串，长度会显著增加（大约为原明文长度的4/3倍，再加上IV和Tag的固定开销），因此字段类型需要设置为TEXT或BLOB，VARCHAR可能不够用。

5.2 应用层代码集成与测试

编写一个测试脚本，演示如何使用我们的加密连接和游标。

# test_encrypted_db.py import sys import os # 将项目目录加入路径，确保能导入自定义模块 sys.path.append(os.path.dirname(os.path.abspath(__file__))) from encrypted_connection import get_encrypted_connection from crypto_util import crypto_util def test_encryption_flow(): # 1. 获取加密连接 conn = get_encrypted_connection() try: with conn.cursor() as cursor: # 2. 插入加密数据 insert_sql = """ INSERT INTO users (username, id_card, mobile, email) VALUES (%(username)s, %(id_card)s, %(mobile)s, %(email)s) """ user_data = { 'username': 'zhangsan', 'id_card': '110101199001011234', # 明文 'mobile': '13800138000', # 明文 'email': 'zhangsan@example.com' # 明文 } # 注意：cursor.execute内部会自动加密id_card, mobile, email字段 cursor.execute(insert_sql, user_data) user_id = conn.insert_id() # 获取自增ID print(f"插入用户成功，ID: {user_id}") conn.commit() # 3. 查询数据（应自动解密） select_sql = "SELECT id, username, id_card, mobile, email FROM users WHERE id = %s" cursor.execute(select_sql, (user_id,)) result = cursor.fetchone() print("查询结果（已自动解密）:") print(f" ID: {result[0]}") print(f" Username: {result[1]}") print(f" ID Card: {result[2]}") # 这里应该是明文 print(f" Mobile: {result[3]}") # 这里应该是明文 print(f" Email: {result[4]}") # 这里应该是明文 # 4. 验证数据库中的存储确实是密文（使用普通游标查询） with conn.cursor(pymysql.cursors.Cursor) as raw_cursor: raw_cursor.execute("SELECT id_card, mobile FROM users WHERE id = %s", (user_id,)) raw_result = raw_cursor.fetchone() print("\n数据库原始存储内容（密文）:") print(f" ID Card (密文): {raw_result[0]}") print(f" Mobile (密文): {raw_result[1]}") # 尝试用工具解密，验证一致性 decrypted_id_card = crypto_util.decrypt(raw_result[0]) print(f" 工具解密ID Card: {decrypted_id_card}") assert decrypted_id_card == user_data['id_card'] # 5. 更新加密字段 update_sql = "UPDATE users SET mobile = %(mobile)s WHERE id = %(id)s" update_data = {'mobile': '13900139000', 'id': user_id} cursor.execute(update_sql, update_data) conn.commit() print(f"\n已更新用户 {user_id} 的手机号。") # 再次查询确认更新和解密 cursor.execute(select_sql, (user_id,)) updated_result = cursor.fetchone() print(f"更新后手机号: {updated_result[3]}") finally: conn.close() if __name__ == '__main__': # 确保环境变量已设置，或修改crypto_util中的密钥 # os.environ['DB_FIELD_ENCRYPTION_KEY'] = 'your-32-byte-secret-key-here!!!' test_encryption_flow()

运行这个脚本，你会看到：

1. 数据插入时，id_card,mobile,email在代码中是明文，但通过EncryptedCursor执行后，存入数据库的是Base64编码的密文。
2. 通过同一个EncryptedCursor查询时，取出的数据自动被解密为明文。
3. 用普通游标查询，可以看到数据库中存储的确实是无法直接识别的密文。
4. 更新操作同样会自动加密新值。

6. 高级议题、常见问题与排查技巧

6.1 模糊查询与索引失效问题

这是字段级加密最大的挑战之一。一旦数据被加密，数据库的LIKE查询、范围查询（BETWEEN,>,<）以及基于该字段的索引都将完全失效，因为数据库操作的是无意义的密文。

解决方案：

1. 取舍与业务设计：首先评估该字段是否真的需要模糊查询。例如，手机号、身份证号通常用于精确匹配，加密不影响=查询。如果需要模糊搜索，考虑业务上是否可以用其他非敏感字段替代（如用户昵称）。
2. 保留明文哈希：对于需要模糊查询的字段（如姓名），可以额外存储一个单向哈希值（如SHA256）或盲索引。查询时，对搜索词同样计算哈希，然后在哈希列上进行精确匹配。但这只能用于等值查询，不能用于LIKE ‘%张%’这种部分匹配。
3. 确定性加密：使用确定性加密算法（如AES-SIV，或在固定IV下的ECB模式），相同的明文总是产生相同的密文。这样可以在密文上做等值查询和建立索引。但这种方法会泄露明文模式，安全性降低，需谨慎评估。
4. 应用层过滤：如果数据量不大，可以将所有数据取回应用层，解密后在内存中过滤。这显然不适用于大数据集。

实操心得：在项目初期就必须和产品、合规部门明确哪些字段需要加密，以及这些字段的查询需求。通常，身份证、银行卡号等用于核验的字段只做精确匹配，加密不影响。而像“诊断详情”这类长文本，本身就不适合数据库索引和模糊查询，加密存储是合理的。

6.2 密钥轮换与数据重加密

密钥不能永远不变。出于安全最佳实践，需要定期轮换加密密钥。这意味着需要用新密钥重新加密数据库中所有已有的密文数据。

操作步骤：

1. 生成新密钥：安全地生成一个新的加密密钥（KEY_NEW）。
2. 创建新加密工具实例：使用KEY_NEW创建一个新的FieldCrypto实例。
3. 分批读取-解密-再加密：
   • 使用旧密钥（KEY_OLD）的解密功能读取数据。
   • 使用KEY_NEW的加密功能重新加密数据。
   • 更新数据库记录。
4. 更新应用配置：在所有应用实例中，将使用的密钥从KEY_OLD切换到KEY_NEW。
5. 安全废弃旧密钥：确认所有数据重加密完成且应用运行稳定后，安全地销毁KEY_OLD。

这个过程需要在维护窗口进行，并确保数据一致性。务必先备份数据！

6.3 性能考量与优化

应用层加密解密会带来额外的CPU开销。

• 影响：主要影响批量插入/更新和数据量大的查询。单条操作的开销通常可忽略。
• 优化：
  • 对于批量操作，确保使用executemany，我们的包装器已支持。
  • 考虑使用更快的加密库实现，如pyca/cryptography本身已高度优化。
  • 对于绝对性能敏感且数据不敏感的场景，可以评估是否真的需要加密。

6.4 常见问题排查表

6.5 实现一个EncryptedDictCursor

很多项目喜欢使用DictCursor来获取字典形式的结果。我们可以类似地实现一个加密版本。

# encrypted_dict_cursor.py import pymysql.cursors from encrypted_cursor import EncryptedCursor # 继承我们已有的EncryptedCursor class EncryptedDictCursor(EncryptedCursor, pymysql.cursors.DictCursorMixin): """ 结合字段加密和字典返回格式的游标。 继承顺序很重要，确保方法解析顺序正确。 """ def __init__(self, connection): super().__init__(connection) # DictCursorMixin需要的初始化 self._query = None self._rows = None self._lastrowid = None # fetch* 方法在EncryptedCursor中已重写并解密，DictCursorMixin会处理字典化。 # 我们只需要确保解密发生在字典化之前。由于Python的多继承方法解析顺序（MRO）， # 调用super().fetchone()会先走到EncryptedCursor的fetchone，完成解密。 # 因此这里不需要再重写fetch方法。

然后在EncryptedConnection中也支持返回这个游标。

这套方案实施下来，业务代码几乎无需改动，只需要换一个连接类，并在配置文件中声明哪些字段需要加密，就能实现数据库字段级的透明加密与解密。它把安全性集中在应用层，密钥管理更灵活，也真正做到了“数据库看不到明文”，是满足许多数据安全合规要求的有效实践。当然，每个项目都需要根据自身的具体需求，在安全性、性能和查询灵活性之间找到合适的平衡点。