| 12345678910111213141516171819202122232425262728293031323334 |
- from sqlalchemy import create_engine, Column, Integer, String, TypeDecorator
- from sqlalchemy.ext.declarative import declarative_base
- from sqlalchemy.orm import sessionmaker
- from cryptography.fernet import Fernet
- import base64
- # 生成加密密钥 (实际应用中应该安全地存储和管理这个密钥)
- key = Fernet.generate_key()
- cipher_suite = Fernet(key)
- Base = declarative_base()
- class EncryptedString(TypeDecorator):
- """自动加密/解密的字符串类型"""
- impl = String # 底层数据库类型
- def process_bind_param(self, value, dialect):
- """在写入数据库前加密数据"""
- if value is not None:
- # 如果是字符串则编码为bytes
- if isinstance(value, str):
- value = value.encode()
- # 加密并返回base64编码的字符串(便于存储)
- return base64.b64encode(cipher_suite.encrypt(value)).decode('utf-8')
- return value
- def process_result_value(self, value, dialect):
- """从数据库读取后解密数据"""
- if value is not None:
- # 解密base64编码的字符串
- return cipher_suite.decrypt(base64.b64decode(value)).decode('utf-8')
- return value
|
