Function plugin_check_tokens

src/api/admin.py:2442–2487

Return token status for external syncers using the plugin connection token.

(request: Optional[dict] = None, authorization: Optional[str] = Header(None))

@router.post("/api/plugin/check-tokens")
async def plugin_check_tokens(request: Optional[dict] = None, authorization: Optional[str] = Header(None)):
    """Return token status for external syncers using the plugin connection token."""
    await _verify_plugin_connection_token(authorization)

    request = request or {}
    requested_emails = request.get("emails") if isinstance(request, dict) else None
    email_filter = set()
    if isinstance(requested_emails, list):
        email_filter = {
            str(email or "").strip().lower()
            for email in requested_emails
            if str(email or "").strip()
        }

    rows = await db.get_all_tokens_with_stats()
    tokens = []
    for row in rows:
        email = str(row.get("email") or "").strip()
        if email_filter and email.lower() not in email_filter:
            continue
        token_obj = None
        try:
            token_obj = Token(**row)
        except Exception:
            token_obj = None
        needs_refresh = token_manager.needs_at_refresh(token_obj) if token_obj else True
        tokens.append({
            "id": row.get("id"),
            "email": email,
            "is_active": bool(row.get("is_active")),
            "needs_refresh": needs_refresh,
            "at_expires": row.get("at_expires").isoformat() if hasattr(row.get("at_expires"), "isoformat") else row.get("at_expires"),
            "last_used_at": row.get("last_used_at").isoformat() if hasattr(row.get("last_used_at"), "isoformat") else row.get("last_used_at"),
            "protocol_mode": row.get("protocol_mode") or "session",
            "auto_refresh_enabled": bool(row.get("auto_refresh_enabled", True)),
            "refresh_interval_minutes": row.get("refresh_interval_minutes") or 120,
            "last_st_refresh_at": (
                row.get("last_st_refresh_at").isoformat()
                if hasattr(row.get("last_st_refresh_at"), "isoformat")
                else row.get("last_st_refresh_at")
            ),
            "last_st_refresh_result": row.get("last_st_refresh_result") or "",
            "credits": row.get("credits", 0),
        })

    return {"success": True, "tokens": tokens}