(self, db: Database, flow_client: FlowClient)
| 16 | |
| 17 | class TokenManager: |
| 18 | """Token lifecycle manager with AT auto-refresh""" |
| 19 | |
| 20 | def __init__(self, db: Database, flow_client: FlowClient): |
| 21 | self.db = db |
| 22 | self.flow_client = flow_client |
| 23 | self._refresh_lock_guard = asyncio.Lock() |
| 24 | self._project_lock_guard = asyncio.Lock() |
| 25 | self._refresh_locks: dict[int, asyncio.Lock] = {} |
| 26 | self._project_locks: dict[int, asyncio.Lock] = {} |
| 27 | self._refresh_futures: dict[int, asyncio.Task] = {} |
| 28 | self._at_validation_cache: dict[int, float] = {} |
| 29 | self._last_at_refresh_error: dict[int, str] = {} |
| 30 | self._protocol_refresher_task: Optional[asyncio.Task] = None |
nothing calls this directly
no outgoing calls
no test coverage detected