from __future__ import annotations from typing import Literal from backend.database.connection import Database from backend.errors import BusinessError from backend.features.accounts.models import Principal from backend.features.accounts.repository import SystemCredentialRepository from backend.features.accounts.service import now_utc from backend.security import SecretCipher CredentialName = Literal["tushare_token", "ifind_refresh_token", "ifind_access_token"] ALLOWED_CREDENTIALS: tuple[CredentialName, ...] = ( "tushare_token", "ifind_refresh_token", "ifind_access_token", ) class SystemCredentialService: def __init__( self, database: Database, repository: SystemCredentialRepository, cipher: SecretCipher, ) -> None: self._database = database self._repository = repository self._cipher = cipher def list_status(self) -> tuple[dict[str, str | bool | None], ...]: with self._database.read() as connection: configured = self._repository.list_status(connection) return tuple( { "name": name, "configured": name in configured, "updated_at": configured.get(name), } for name in ALLOWED_CREDENTIALS ) def save(self, actor: Principal, name: str, value: str) -> None: if name not in ALLOWED_CREDENTIALS: raise BusinessError("invalid_credential", "系统凭据类型无效。") cleaned = value.strip() if not cleaned or len(cleaned) > 4096: raise BusinessError("invalid_credential", "系统凭据内容无效。") with self._database.transaction() as connection: self._repository.save( connection, name, self._cipher.encrypt(cleaned), actor.user.id, now_utc() ) def get(self, name: CredentialName) -> str | None: with self._database.read() as connection: encrypted = self._repository.get_encrypted(connection, name) return self._cipher.decrypt(encrypted) if encrypted else None