MCPcopy Create free account
hub / github.com/TheSmallHanCat/flow2api / ensure_valid_token

Method ensure_valid_token

src/services/token_manager.py:522–547  ·  view source on GitHub ↗

确保 token 的 AT 可用,并在必要时返回刷新后的最新对象。

(self, token: Optional[Token])

Source from the content-addressed store, hash-verified

520 return self._should_refresh_at(token)
521
522 async def ensure_valid_token(self, token: Optional[Token]) -> Optional[Token]:
523 """确保 token 的 AT 可用,并在必要时返回刷新后的最新对象。"""
524 if not token:
525 return None
526
527 if not self._should_refresh_at(token):
528 if self._has_recent_at_validation(token.id):
529 return token
530 try:
531 credits_result = await self._get_credits_for_token(token, token.at)
532 await self.db.update_token(
533 token.id,
534 credits=credits_result.get("credits", 0),
535 user_paygate_tier=credits_result.get("userPaygateTier"),
536 )
537 self._mark_at_valid(token.id)
538 return await self.db.get_token(token.id) or token
539 except Exception as e:
540 debug_logger.log_warning(
541 f"[AT_CHECK] Token {token.id}: 本地判定未过期,但上游校验失败,准备刷新 AT/ST - {e}"
542 )
543
544 if not await self._refresh_at(token.id):
545 return None
546
547 return await self.db.get_token(token.id)
548
549 async def is_at_valid(self, token_id: int, token: Optional[Token] = None) -> bool:
550 """检查AT是否有效,如果无效或即将过期则自动刷新

Callers 4

is_at_validMethod · 0.95
refresh_creditsMethod · 0.95
handle_generationMethod · 0.80
select_tokenMethod · 0.80

Calls 8

_should_refresh_atMethod · 0.95
_mark_at_validMethod · 0.95
_refresh_atMethod · 0.95
log_warningMethod · 0.80
update_tokenMethod · 0.45
get_tokenMethod · 0.45

Tested by

no test coverage detected