SQLAlchemy实现列加密且检索时不自动解密的方案咨询

Hey there! Let’s tackle this problem step by step. First, let’s be clear: the default EncryptedType from sqlalchemy_utils is built to automatically decrypt data when you pull it from the database—so it doesn’t natively support your new requirement of keeping data encrypted during retrieval. But don’t worry, we’ve got straightforward workarounds to make this happen.

Option 1: Customize EncryptedType to Skip Decryption on Retrieval #

The simplest approach is to create a subclass of EncryptedType and override the method that handles result values. By default, process_result_value decrypts stored data; we can modify it to return the raw encrypted string instead.

Here’s a quick implementation:

from sqlalchemy_utils.types.encrypted import EncryptedType

class EncryptedReadOnlyType(EncryptedType):
    def process_result_value(self, value, dialect):
        # Bypass default decryption logic—return the raw encrypted value from the database
        return value

How to Use It #

In your SQLAlchemy model, just swap the original EncryptedType with this custom type:

from sqlalchemy import Column, Integer, String
from your_module import EncryptedReadOnlyType

class YourModel(Base):
    __tablename__ = "your_table"
    id = Column(Integer, primary_key=True)
    sensitive_data = Column(EncryptedReadOnlyType(type_in=String, key="your-predefined-key"))

Now, data still gets encrypted automatically on insert, but when you query sensitive_data, you’ll get the encrypted string directly—no decryption applied.

If you ever need to toggle between encrypted and decrypted retrieval (for debugging or specific use cases), add a configurable flag to the custom type:

class ToggleableEncryptedType(EncryptedType):
    def __init__(self, *args, decrypt_on_retrieve=True, **kwargs):
        super().__init__(*args, **kwargs)
        self.decrypt_on_retrieve = decrypt_on_retrieve

    def process_result_value(self, value, dialect):
        if self.decrypt_on_retrieve and value is not None:
            # Use default decryption if enabled
            return super().process_result_value(value, dialect)
        # Return encrypted value otherwise
        return value

Use it like this when you want encrypted retrieval:

sensitive_data = Column(ToggleableEncryptedType(type_in=String, key="your-key", decrypt_on_retrieve=False))

Option 2: Build Your Own Custom SQLAlchemy Type #

If you’d rather not depend on sqlalchemy_utils, create a custom TypeDecorator from scratch using a library like cryptography. This gives you full control over the encryption flow.

Here’s an example using Fernet (the same symmetric encryption method EncryptedType uses under the hood):

from sqlalchemy import TypeDecorator, String
from cryptography.fernet import Fernet

# Note: In production, never hardcode your key—use environment variables or a secure secrets manager
ENCRYPTION_KEY = b"your-32-url-safe-base64-key-here"
fernet = Fernet(ENCRYPTION_KEY)

class EncryptedStorageType(TypeDecorator):
    # Use String as the underlying database type
    impl = String

    def process_bind_param(self, value, dialect):
        if value is not None:
            # Encrypt the value before storing it
            encrypted_bytes = fernet.encrypt(value.encode("utf-8"))
            return encrypted_bytes.decode("utf-8")
        return value

    def process_result_value(self, value, dialect):
        # Return the raw encrypted value without decrypting
        return value

Usage in Model #

class YourModel(Base):
    __tablename__ = "your_table"
    id = Column(Integer, primary_key=True)
    sensitive_data = Column(EncryptedStorageType)

This works exactly as needed: data is encrypted on insert, and stays encrypted when you retrieve it.

Key Notes to Remember #

• Key Management: Never hardcode encryption keys in your codebase. Use environment variables, a secrets manager (like AWS Secrets Manager or HashiCorp Vault), or secure configuration files.
• Compatibility: If migrating from the original EncryptedType, ensure your custom type uses the same encryption algorithm and key—this keeps existing encrypted data compatible.
• Testing: Always test both insertion and retrieval to confirm encrypted values are stored and returned as expected.

内容的提问来源于stack exchange，提问作者Jfach