如何实现仅允许0或1条记录的SQLAlchemy加密密钥存储表？

下面提供几种可靠的实现方式，从数据库层面到代码层都有覆盖，你可以根据自己的数据库环境和需求选择：

1. 数据库层面：唯一约束+固定标识列 #

这是最可靠的方案，直接在数据库表上添加约束，从根源上限制只能存在一条记录。

实现代码 #

from sqlalchemy import Column, String, Integer, UniqueConstraint
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class EncryptedKey(Base):
    __tablename__ = 'encrypted_keys'
    
    id = Column(Integer, primary_key=True)
    encrypted_key = Column(String, nullable=False)
    # 固定值列，用于唯一约束控制单条记录
    singleton_guard = Column(Integer, default=1, nullable=False)
    
    # 给固定列添加唯一约束，确保整个表只能有一个值为1的记录
    __table_args__ = (
        UniqueConstraint('singleton_guard', name='_singleton_guard_unique'),
    )

原理说明 #

插入第二条记录时，数据库会因为singleton_guard列的唯一约束抛出冲突错误，直接阻止重复插入。你可以在代码中捕获这个错误，提示用户“密钥已存在”或者执行更新操作。

2. 代码层面：单例模式封装操作 #

如果需要更灵活的逻辑控制（比如自动更新现有密钥），可以在模型类中封装一个类方法，确保每次操作都只会维护一条记录。

实现代码 #

from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

class EncryptedKey(Base):
    # 复用上面的表定义（可以去掉singleton_guard列，也可以保留双重保险）
    __tablename__ = 'encrypted_keys'
    
    id = Column(Integer, primary_key=True)
    encrypted_key = Column(String, nullable=False)

    @classmethod
    def get_or_update(cls, db: Session, new_encrypted_key: str):
        # 先查询是否已有记录
        existing_key = db.query(cls).first()
        if existing_key:
            # 更新现有密钥
            existing_key.encrypted_key = new_encrypted_key
            db.commit()
            db.refresh(existing_key)
            return existing_key
        
        # 无记录则创建新的
        new_key = cls(encrypted_key=new_encrypted_key)
        db.add(new_key)
        try:
            db.commit()
            db.refresh(new_key)
            return new_key
        except IntegrityError:
            # 处理并发插入场景：回滚后重新查询
            db.rollback()
            return db.query(cls).first()

原理说明 #

通过先查询再操作的逻辑，确保不会重复创建记录；并发场景下通过捕获IntegrityError（如果同时有两个请求尝试创建），回滚后重新查询现有记录，避免出现多条数据。

3. 数据库层面：CHECK约束（仅部分数据库支持） #

部分数据库（如PostgreSQL）支持表级CHECK约束，直接限制表的行数不超过1。

实现代码 #

from sqlalchemy import Column, String, Integer, CheckConstraint
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class EncryptedKey(Base):
    __tablename__ = 'encrypted_keys'
    
    id = Column(Integer, primary_key=True)
    encrypted_key = Column(String, nullable=False)
    
    # 约束表的总行数不超过1
    __table_args__ = (
        CheckConstraint('(SELECT COUNT(*) FROM encrypted_keys) <= 1', name='max_one_row'),
    )

注意事项 #

• MySQL不支持这种引用整个表的CHECK约束，仅适合PostgreSQL等兼容的数据库；
• 这种约束在插入时会触发全表计数，性能略低于第一种方案。

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