from datetime import datetime from sqlalchemy import Boolean, DateTime, Float, Index, Integer, String, Text, text from sqlalchemy.orm import Mapped, mapped_column from app.models.base import Base, TimestampMixin, SoftDeleteMixin from app.utils.security import encrypt_text, decrypt_text class ApiKey(Base, TimestampMixin, SoftDeleteMixin): """对外开放 API 的密钥管理表。 每个 api-key 对应一个外部调用方(公司/组织), 可配置可调用模型、配额、有效期、并发限制。 """ __tablename__ = "api_keys" __table_args__ = ( Index("idx_api_keys_active", "is_active", postgresql_where=text("deleted_at IS NULL")), Index("idx_api_keys_company", "company_name"), ) id: Mapped[str] = mapped_column(String(32), primary_key=True) company_name: Mapped[str] = mapped_column(String(128), nullable=False) api_key_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, index=True) api_key_prefix: Mapped[str] = mapped_column(String(16), nullable=False) api_key_encrypted: Mapped[str] = mapped_column(Text, nullable=False, comment="AES-256-GCM 加密的完整 API Key") description: Mapped[str | None] = mapped_column(Text, nullable=True) def decrypt_api_key(self) -> str | None: """解密并返回完整 API Key。""" return decrypt_text(self.api_key_encrypted) def set_plaintext_key(self, plaintext: str) -> None: """设置明文 API Key(自动加密存储)。""" self.api_key_encrypted = encrypt_text(plaintext) # === 可调用模型配置 === callable_models: Mapped[str] = mapped_column(Text, nullable=False, server_default="[]", comment='JSON数组: [{"engine_type":"video","engine_id":"xxx","model_name":"doubao-seedance-2-0-260128"}]') # === 配额配置(不设置=无限制) === quota_limit: Mapped[float | None] = mapped_column(Float, nullable=True, comment="配额总量,NULL=无限") quota_cycle: Mapped[str | None] = mapped_column(String(16), nullable=True, comment="daily|monthly|one_time|NULL=无限") quota_used: Mapped[float] = mapped_column(Float, nullable=False, default=0.0, server_default="0.0", comment="当前周期已使用量") # === 有效期(不设置=永不过期) === valid_from: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) valid_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) # === 并发限制(不设置=无限制) === max_concurrent_video_tasks: Mapped[int | None] = mapped_column(Integer, nullable=True, comment="最大并发视频任务数,NULL=无限") # === 状态 === is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="true") last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)