""" API Key 模型 — 用于第三方应用/脚本通过 API 调用平台服务 """ import hashlib import logging from datetime import datetime from typing import Optional from sqlalchemy import Column, String, DateTime, JSON, func, ForeignKey from sqlalchemy.dialects.mysql import CHAR from sqlalchemy.orm import relationship, Session from app.core.database import Base import uuid logger = logging.getLogger(__name__) KEY_PREFIX = "sk-tg-" class ApiKey(Base): """API Key 表""" __tablename__ = "api_keys" id = Column(CHAR(36), primary_key=True, default=lambda: str(uuid.uuid4()), comment="密钥ID") user_id = Column(CHAR(36), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, comment="所属用户ID") workspace_id = Column(CHAR(36), ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False, comment="所属工作区ID") name = Column(String(100), nullable=False, comment="密钥名称(用于标识)") key_prefix = Column(String(12), nullable=False, comment="密钥前缀(如 sk-tg-)") key_hash = Column(String(255), nullable=False, unique=True, comment="SHA256 哈希存储") permissions = Column(JSON, nullable=True, comment="权限范围(null=全部权限)") last_used_at = Column(DateTime, nullable=True, comment="最后使用时间") expires_at = Column(DateTime, nullable=True, comment="过期时间(null=永不过期)") status = Column(String(20), default="active", comment="状态: active/revoked") created_at = Column(DateTime, default=func.now(), comment="创建时间") # 关系 user = relationship("User", backref="api_keys") def __repr__(self): return f"" def verify_api_key(api_key_str: str, db: Session) -> Optional["ApiKey"]: """验证 API Key 字符串,返回 ApiKey 对象或 None。""" if not api_key_str or not api_key_str.startswith(KEY_PREFIX): return None key_hash = hashlib.sha256(api_key_str.encode()).hexdigest() api_key = db.query(ApiKey).filter( ApiKey.key_hash == key_hash, ApiKey.status == "active", ).first() if not api_key: return None if api_key.expires_at and api_key.expires_at < datetime.utcnow(): logger.warning("API Key %s 已过期", api_key.id) return None return api_key