GithubIntegration.py
1 from __future__ import annotations 2 3 import warnings 4 from typing import Any 5 6 import deprecated 7 import urllib3 8 from urllib3 import Retry 9 10 import github 11 from github import Consts 12 from github.Auth import AppAuth 13 from github.GithubApp import GithubApp 14 from github.GithubException import GithubException 15 from github.Installation import Installation 16 from github.InstallationAuthorization import InstallationAuthorization 17 from github.PaginatedList import PaginatedList 18 from github.Requester import Requester 19 20 21 class GithubIntegration: 22 """ 23 Main class to obtain tokens for a GitHub integration. 24 """ 25 26 # keep non-deprecated arguments in-sync with Requester 27 # v3: remove integration_id, private_key, jwt_expiry, jwt_issued_at and jwt_algorithm 28 # v3: move auth to the front of arguments 29 # v3: move * before first argument so all arguments must be named, 30 # allows to reorder / add new arguments / remove deprecated arguments without breaking user code 31 # added here to force named parameters because new parameters have been added 32 auth: AppAuth 33 base_url: str 34 __requester: Requester 35 36 def __init__( 37 self, 38 integration_id: int | str | None = None, 39 private_key: str | None = None, 40 base_url: str = Consts.DEFAULT_BASE_URL, 41 *, 42 timeout: int = Consts.DEFAULT_TIMEOUT, 43 user_agent: str = Consts.DEFAULT_USER_AGENT, 44 per_page: int = Consts.DEFAULT_PER_PAGE, 45 verify: bool | str = True, 46 retry: int | Retry | None = None, 47 pool_size: int | None = None, 48 seconds_between_requests: float | None = Consts.DEFAULT_SECONDS_BETWEEN_REQUESTS, 49 seconds_between_writes: float | None = Consts.DEFAULT_SECONDS_BETWEEN_WRITES, 50 jwt_expiry: int = Consts.DEFAULT_JWT_EXPIRY, 51 jwt_issued_at: int = Consts.DEFAULT_JWT_ISSUED_AT, 52 jwt_algorithm: str = Consts.DEFAULT_JWT_ALGORITHM, 53 auth: AppAuth | None = None, 54 ) -> None: 55 """ 56 :param integration_id: int deprecated, use auth=github.Auth.AppAuth(...) instead 57 :param private_key: string deprecated, use auth=github.Auth.AppAuth(...) instead 58 :param base_url: string 59 :param timeout: integer 60 :param user_agent: string 61 :param per_page: int 62 :param verify: boolean or string 63 :param retry: int or urllib3.util.retry.Retry object 64 :param pool_size: int 65 :param seconds_between_requests: float 66 :param seconds_between_writes: float 67 :param jwt_expiry: int deprecated, use auth=github.Auth.AppAuth(...) instead 68 :param jwt_issued_at: int deprecated, use auth=github.Auth.AppAuth(...) instead 69 :param jwt_algorithm: string deprecated, use auth=github.Auth.AppAuth(...) instead 70 :param auth: authentication method 71 """ 72 if integration_id is not None: 73 assert isinstance(integration_id, (int, str)), integration_id 74 if private_key is not None: 75 assert isinstance(private_key, str), "supplied private key should be a string" 76 assert isinstance(base_url, str), base_url 77 assert isinstance(timeout, int), timeout 78 assert user_agent is None or isinstance(user_agent, str), user_agent 79 assert isinstance(per_page, int), per_page 80 assert isinstance(verify, (bool, str)), verify 81 assert retry is None or isinstance(retry, int) or isinstance(retry, urllib3.util.Retry), retry 82 assert pool_size is None or isinstance(pool_size, int), pool_size 83 assert seconds_between_requests is None or seconds_between_requests >= 0 84 assert seconds_between_writes is None or seconds_between_writes >= 0 85 assert isinstance(jwt_expiry, int), jwt_expiry 86 assert Consts.MIN_JWT_EXPIRY <= jwt_expiry <= Consts.MAX_JWT_EXPIRY, jwt_expiry 87 assert isinstance(jwt_issued_at, int) 88 89 self.base_url = base_url 90 91 if ( 92 integration_id is not None 93 or private_key is not None 94 or jwt_expiry != Consts.DEFAULT_JWT_EXPIRY 95 or jwt_issued_at != Consts.DEFAULT_JWT_ISSUED_AT 96 or jwt_algorithm != Consts.DEFAULT_JWT_ALGORITHM 97 ): 98 warnings.warn( 99 "Arguments integration_id, private_key, jwt_expiry, jwt_issued_at and jwt_algorithm are deprecated, " 100 "please use auth=github.Auth.AppAuth(...) instead", 101 category=DeprecationWarning, 102 ) 103 auth = AppAuth( 104 integration_id, # type: ignore 105 private_key, # type: ignore 106 jwt_expiry=jwt_expiry, 107 jwt_issued_at=jwt_issued_at, 108 jwt_algorithm=jwt_algorithm, 109 ) 110 111 assert isinstance( 112 auth, AppAuth 113 ), f"GithubIntegration requires github.Auth.AppAuth authentication, not {type(auth)}" 114 115 self.auth = auth 116 117 self.__requester = Requester( 118 auth=auth, 119 base_url=self.base_url, 120 timeout=timeout, 121 user_agent=user_agent, 122 per_page=per_page, 123 verify=verify, 124 retry=retry, 125 pool_size=pool_size, 126 seconds_between_requests=seconds_between_requests, 127 seconds_between_writes=seconds_between_writes, 128 ) 129 130 def close(self) -> None: 131 """ 132 Close connections to the server. Alternatively, use the GithubIntegration object as a context manager: 133 134 .. code-block:: python 135 136 with github.GithubIntegration(...) as gi: 137 # do something 138 """ 139 self.__requester.close() 140 141 def __enter__(self) -> GithubIntegration: 142 return self 143 144 def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: 145 self.close() 146 147 def get_github_for_installation( 148 self, installation_id: int, token_permissions: dict[str, str] | None = None 149 ) -> github.Github: 150 # The installation has to authenticate as an installation, not an app 151 auth = self.auth.get_installation_auth(installation_id, token_permissions, self.__requester) 152 return github.Github(**self.__requester.withAuth(auth).kwargs) 153 154 def _get_headers(self) -> dict[str, str]: 155 """ 156 Get headers for the requests. 157 """ 158 return { 159 "Accept": Consts.mediaTypeIntegrationPreview, 160 } 161 162 def _get_installed_app(self, url: str) -> Installation: 163 """ 164 Get installation for the given URL. 165 """ 166 headers, response = self.__requester.requestJsonAndCheck("GET", url, headers=self._get_headers()) 167 168 return Installation( 169 requester=self.__requester, 170 headers=headers, 171 attributes=response, 172 completed=True, 173 ) 174 175 @deprecated.deprecated( 176 "Use github.Github(auth=github.Auth.AppAuth), github.Auth.AppAuth.token or github.Auth.AppAuth.create_jwt(expiration) instead" 177 ) 178 def create_jwt(self, expiration: int | None = None) -> str: 179 """ 180 Create a signed JWT 181 https://docs.github.com/en/developers/apps/building-github-apps/authenticating-with-github-apps#authenticating-as-a-github-app 182 """ 183 return self.auth.create_jwt(expiration) 184 185 def get_access_token( 186 self, installation_id: int, permissions: dict[str, str] | None = None 187 ) -> InstallationAuthorization: 188 """ 189 :calls: `POST /app/installations/{installation_id}/access_tokens <https://docs.github.com/en/rest/apps/apps#create-an-installation-access-token-for-an-app>` 190 """ 191 if permissions is None: 192 permissions = {} 193 194 if not isinstance(permissions, dict): 195 raise GithubException(status=400, data={"message": "Invalid permissions"}, headers=None) 196 197 body = {"permissions": permissions} 198 headers, response = self.__requester.requestJsonAndCheck( 199 "POST", 200 f"/app/installations/{installation_id}/access_tokens", 201 headers=self._get_headers(), 202 input=body, 203 ) 204 205 return InstallationAuthorization( 206 requester=self.__requester, 207 headers=headers, 208 attributes=response, 209 completed=True, 210 ) 211 212 @deprecated.deprecated("Use get_repo_installation") 213 def get_installation(self, owner: str, repo: str) -> Installation: 214 """ 215 Deprecated by get_repo_installation 216 217 :calls: `GET /repos/{owner}/{repo}/installation <https://docs.github.com/en/rest/reference/apps#get-a-repository-installation-for-the-authenticated-app>` 218 """ 219 return self._get_installed_app(url=f"/repos/{owner}/{repo}/installation") 220 221 def get_installations(self) -> PaginatedList[Installation]: 222 """ 223 :calls: GET /app/installations <https://docs.github.com/en/rest/reference/apps#list-installations-for-the-authenticated-app> 224 """ 225 return PaginatedList( 226 contentClass=Installation, 227 requester=self.__requester, 228 firstUrl="/app/installations", 229 firstParams=None, 230 headers=self._get_headers(), 231 list_item="installations", 232 ) 233 234 def get_org_installation(self, org: str) -> Installation: 235 """ 236 :calls: `GET /orgs/{org}/installation <https://docs.github.com/en/rest/apps/apps#get-an-organization-installation-for-the-authenticated-app>` 237 """ 238 return self._get_installed_app(url=f"/orgs/{org}/installation") 239 240 def get_repo_installation(self, owner: str, repo: str) -> Installation: 241 """ 242 :calls: `GET /repos/{owner}/{repo}/installation <https://docs.github.com/en/rest/reference/apps#get-a-repository-installation-for-the-authenticated-app>` 243 """ 244 return self._get_installed_app(url=f"/repos/{owner}/{repo}/installation") 245 246 def get_user_installation(self, username: str) -> Installation: 247 """ 248 :calls: `GET /users/{username}/installation <https://docs.github.com/en/rest/apps/apps#get-a-user-installation-for-the-authenticated-app>` 249 """ 250 return self._get_installed_app(url=f"/users/{username}/installation") 251 252 def get_app_installation(self, installation_id: int) -> Installation: 253 """ 254 :calls: `GET /app/installations/{installation_id} <https://docs.github.com/en/rest/apps/apps#get-an-installation-for-the-authenticated-app>` 255 """ 256 return self._get_installed_app(url=f"/app/installations/{installation_id}") 257 258 def get_app(self) -> GithubApp: 259 """ 260 :calls: `GET /app <https://docs.github.com/en/rest/reference/apps#get-the-authenticated-app>`_ 261 """ 262 263 headers, data = self.__requester.requestJsonAndCheck("GET", "/app", headers=self._get_headers()) 264 return GithubApp(requester=self.__requester, headers=headers, attributes=data, completed=True)