(self)
| 35 | self._session = None |
| 36 | |
| 37 | async def get_token(self) -> str | None: |
| 38 | if self._token and time.time() < (self._token_expires_at - TOKEN_REFRESH_BUFFER): |
| 39 | return self._token |
| 40 | |
| 41 | logger.info("Fetching new GitHub App installation token...") |
| 42 | |
| 43 | try: |
| 44 | jwt_token = self._generate_jwt() |
| 45 | session = await self._get_session() |
| 46 | |
| 47 | url = f"https://api.github.com/app/installations/{self.installation_id}/access_tokens" |
| 48 | headers = { |
| 49 | "Authorization": f"Bearer {jwt_token}", |
| 50 | "Accept": "application/vnd.github+json", |
| 51 | "X-GitHub-Api-Version": "2022-11-28", |
| 52 | } |
| 53 | |
| 54 | async with session.post(url, headers=headers) as response: |
| 55 | if response.status == 201: |
| 56 | data = await response.json() |
| 57 | self._token = data["token"] |
| 58 | self._token_expires_at = time.time() + 3600 |
| 59 | logger.info("GitHub App token refreshed successfully") |
| 60 | return self._token |
| 61 | else: |
| 62 | error = await response.text() |
| 63 | logger.error(f"Failed to get installation token: {response.status} - {error}") |
| 64 | return None |
| 65 | |
| 66 | except Exception as e: |
| 67 | logger.error(f"Error getting GitHub App token: {e}") |
| 68 | return None |
| 69 | |
| 70 | |
| 71 | github_app_auth: GitHubAppAuth | None = None |
no test coverage detected