Add auth token to request, refreshing if needed.
(
self,
request: ClientRequest,
handler: ClientHandlerType,
)
| 86 | return self.access_token |
| 87 | |
| 88 | async def __call__( |
| 89 | self, |
| 90 | request: ClientRequest, |
| 91 | handler: ClientHandlerType, |
| 92 | ) -> ClientResponse: |
| 93 | """Add auth token to request, refreshing if needed.""" |
| 94 | # Skip token for refresh endpoint to avoid recursion |
| 95 | if str(request.url).endswith("/token/refresh"): |
| 96 | return await handler(request) |
| 97 | |
| 98 | # Refresh token if needed |
| 99 | if not self.access_token or ( |
| 100 | self.token_expires_at and time.time() >= self.token_expires_at |
| 101 | ): |
| 102 | await self._refresh_access_token(request.session) |
| 103 | |
| 104 | # Add token to request |
| 105 | request.headers[hdrs.AUTHORIZATION] = f"Bearer {self.access_token}" |
| 106 | _LOGGER.debug("Added Bearer token to request") |
| 107 | |
| 108 | # Execute request |
| 109 | response = await handler(request) |
| 110 | |
| 111 | # If we get 401, try refreshing token once |
| 112 | if response.status == HTTPStatus.UNAUTHORIZED: |
| 113 | _LOGGER.info("Got 401, attempting token refresh...") |
| 114 | await self._refresh_access_token(request.session) |
| 115 | request.headers[hdrs.AUTHORIZATION] = f"Bearer {self.access_token}" |
| 116 | response = await handler(request) |
| 117 | |
| 118 | return response |
| 119 | |
| 120 | |
| 121 | class TestServer: |
nothing calls this directly
no test coverage detected