############################ Copyrights and license ############################ # # # Copyright 2013 AKFish # # Copyright 2013 Ed Jackson # # Copyright 2013 Jonathan J Hunt # # Copyright 2013 Peter Golm # # Copyright 2013 Steve Brown # # Copyright 2013 Vincent Jacques # # Copyright 2014 C. R. Oldham # # Copyright 2014 Thialfihar # # Copyright 2014 Tyler Treat # # Copyright 2014 Vincent Jacques # # Copyright 2015 Daniel Pocock # # Copyright 2015 Joseph Rawson # # Copyright 2015 Uriel Corfa # # Copyright 2015 edhollandAL # # Copyright 2016 Jannis Gebauer # # Copyright 2016 Peter Buckley # # Copyright 2017 Colin Hoglund # # Copyright 2017 Jannis Gebauer # # Copyright 2018 Agor Maxime # # Copyright 2018 Joshua Hoblitt # # Copyright 2018 Maarten Fonville # # Copyright 2018 Mike Miller # # Copyright 2018 Svend Sorensen # # Copyright 2018 Wan Liuyang # # Copyright 2018 sfdye # # Copyright 2018 itsbruce # # Copyright 2019 Tomas Tomecek # # Copyright 2019 Rigas Papathanasopoulos # # # # This file is part of PyGithub. # # http://pygithub.readthedocs.io/ # # # # PyGithub is free software: you can redistribute it and/or modify it under # # the terms of the GNU Lesser General Public License as published by the Free # # Software Foundation, either version 3 of the License, or (at your option) # # any later version. # # # # PyGithub is distributed in the hope that it will be useful, but WITHOUT ANY # # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS # # FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more # # details. # # # # You should have received a copy of the GNU Lesser General Public License # # along with PyGithub. If not, see . # # # ################################################################################ import datetime import pickle import time import jwt import requests import urllib3 import github.ApplicationOAuth import github.Event import github.Gist import github.GithubObject import github.License import github.NamedUser import github.PaginatedList import github.Topic from . import ( AuthenticatedUser, Consts, GithubApp, GithubException, GitignoreTemplate, HookDescription, Installation, InstallationAuthorization, RateLimit, Repository, ) from .Requester import Requester DEFAULT_BASE_URL = "https://api.github.com" DEFAULT_STATUS_URL = "https://status.github.com" # As of 2018-05-17, Github imposes a 10s limit for completion of API requests. # Thus, the timeout should be slightly > 10s to account for network/front-end # latency. DEFAULT_TIMEOUT = 15 DEFAULT_PER_PAGE = 30 class Github: """ This is the main class you instantiate to access the Github API v3. Optional parameters allow different authentication methods. """ def __init__( self, login_or_token=None, password=None, jwt=None, base_url=DEFAULT_BASE_URL, timeout=DEFAULT_TIMEOUT, user_agent="PyGithub/Python", per_page=DEFAULT_PER_PAGE, verify=True, retry=None, pool_size=None, ): """ :param login_or_token: string :param password: string :param base_url: string :param timeout: integer :param user_agent: string :param per_page: int :param verify: boolean or string :param retry: int or urllib3.util.retry.Retry object :param pool_size: int """ assert login_or_token is None or isinstance(login_or_token, str), login_or_token assert password is None or isinstance(password, str), password assert jwt is None or isinstance(jwt, str), jwt assert isinstance(base_url, str), base_url assert isinstance(timeout, int), timeout assert user_agent is None or isinstance(user_agent, str), user_agent assert ( retry is None or isinstance(retry, (int)) or isinstance(retry, (urllib3.util.Retry)) ) assert pool_size is None or isinstance(pool_size, (int)), pool_size self.__requester = Requester( login_or_token, password, jwt, base_url, timeout, user_agent, per_page, verify, retry, pool_size, ) @property def FIX_REPO_GET_GIT_REF(self): """ :type: bool """ return self.__requester.FIX_REPO_GET_GIT_REF @FIX_REPO_GET_GIT_REF.setter def FIX_REPO_GET_GIT_REF(self, value): self.__requester.FIX_REPO_GET_GIT_REF = value # v2: Remove this property? Why should it be necessary to read/modify it after construction @property def per_page(self): """ :type: int """ return self.__requester.per_page @per_page.setter def per_page(self, value): self.__requester.per_page = value # v2: Provide a unified way to access values of headers of last response # v2: (and add/keep ad hoc properties for specific useful headers like rate limiting, oauth scopes, etc.) # v2: Return an instance of a class: using a tuple did not allow to add a field "resettime" @property def rate_limiting(self): """ First value is requests remaining, second value is request limit. :type: (int, int) """ remaining, limit = self.__requester.rate_limiting if limit < 0: self.get_rate_limit() return self.__requester.rate_limiting @property def rate_limiting_resettime(self): """ Unix timestamp indicating when rate limiting will reset. :type: int """ if self.__requester.rate_limiting_resettime == 0: self.get_rate_limit() return self.__requester.rate_limiting_resettime def get_rate_limit(self): """ Rate limit status for different resources (core/search/graphql). :calls: `GET /rate_limit `_ :rtype: :class:`github.RateLimit.RateLimit` """ headers, data = self.__requester.requestJsonAndCheck("GET", "/rate_limit") return RateLimit.RateLimit(self.__requester, headers, data["resources"], True) @property def oauth_scopes(self): """ :type: list of string """ return self.__requester.oauth_scopes def get_license(self, key=github.GithubObject.NotSet): """ :calls: `GET /license/{license} `_ :param key: string :rtype: :class:`github.License.License` """ assert isinstance(key, str), key headers, data = self.__requester.requestJsonAndCheck("GET", f"/licenses/{key}") return github.License.License(self.__requester, headers, data, completed=True) def get_licenses(self): """ :calls: `GET /licenses `_ :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.License.License` """ url_parameters = dict() return github.PaginatedList.PaginatedList( github.License.License, self.__requester, "/licenses", url_parameters ) def get_events(self): """ :calls: `GET /events `_ :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Event.Event` """ return github.PaginatedList.PaginatedList( github.Event.Event, self.__requester, "/events", None ) def get_user(self, login=github.GithubObject.NotSet): """ :calls: `GET /users/{user} `_ or `GET /user `_ :param login: string :rtype: :class:`github.NamedUser.NamedUser` or :class:`github.AuthenticatedUser.AuthenticatedUser` """ assert login is github.GithubObject.NotSet or isinstance(login, str), login if login is github.GithubObject.NotSet: return AuthenticatedUser.AuthenticatedUser( self.__requester, {}, {"url": "/user"}, completed=False ) else: headers, data = self.__requester.requestJsonAndCheck( "GET", f"/users/{login}" ) return github.NamedUser.NamedUser( self.__requester, headers, data, completed=True ) def get_user_by_id(self, user_id): """ :calls: `GET /user/{id} `_ :param user_id: int :rtype: :class:`github.NamedUser.NamedUser` """ assert isinstance(user_id, int), user_id headers, data = self.__requester.requestJsonAndCheck("GET", f"/user/{user_id}") return github.NamedUser.NamedUser( self.__requester, headers, data, completed=True ) def get_users(self, since=github.GithubObject.NotSet): """ :calls: `GET /users `_ :param since: integer :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.NamedUser.NamedUser` """ assert since is github.GithubObject.NotSet or isinstance(since, int), since url_parameters = dict() if since is not github.GithubObject.NotSet: url_parameters["since"] = since return github.PaginatedList.PaginatedList( github.NamedUser.NamedUser, self.__requester, "/users", url_parameters ) def get_organization(self, login): """ :calls: `GET /orgs/{org} `_ :param login: string :rtype: :class:`github.Organization.Organization` """ assert isinstance(login, str), login headers, data = self.__requester.requestJsonAndCheck("GET", f"/orgs/{login}") return github.Organization.Organization( self.__requester, headers, data, completed=True ) def get_organizations(self, since=github.GithubObject.NotSet): """ :calls: `GET /organizations `_ :param since: integer :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Organization.Organization` """ assert since is github.GithubObject.NotSet or isinstance(since, int), since url_parameters = dict() if since is not github.GithubObject.NotSet: url_parameters["since"] = since return github.PaginatedList.PaginatedList( github.Organization.Organization, self.__requester, "/organizations", url_parameters, ) def get_repo(self, full_name_or_id, lazy=False): """ :calls: `GET /repos/{owner}/{repo} `_ or `GET /repositories/{id} `_ :rtype: :class:`github.Repository.Repository` """ assert isinstance(full_name_or_id, (str, int)), full_name_or_id url_base = "/repositories/" if isinstance(full_name_or_id, int) else "/repos/" url = f"{url_base}{full_name_or_id}" if lazy: return Repository.Repository( self.__requester, {}, {"url": url}, completed=False ) headers, data = self.__requester.requestJsonAndCheck("GET", url) return Repository.Repository(self.__requester, headers, data, completed=True) def get_repos( self, since=github.GithubObject.NotSet, visibility=github.GithubObject.NotSet ): """ :calls: `GET /repositories `_ :param since: integer :param visibility: string ('all','public') :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Repository.Repository` """ assert since is github.GithubObject.NotSet or isinstance(since, int), since url_parameters = dict() if since is not github.GithubObject.NotSet: url_parameters["since"] = since if visibility is not github.GithubObject.NotSet: assert visibility in ("public", "all"), visibility url_parameters["visibility"] = visibility return github.PaginatedList.PaginatedList( github.Repository.Repository, self.__requester, "/repositories", url_parameters, ) def get_project(self, id): """ :calls: `GET /projects/{project_id} `_ :rtype: :class:`github.Project.Project` :param id: integer """ headers, data = self.__requester.requestJsonAndCheck( "GET", "/projects/%d" % (id), headers={"Accept": Consts.mediaTypeProjectsPreview}, ) return github.Project.Project(self.__requester, headers, data, completed=True) def get_project_column(self, id): """ :calls: `GET /projects/columns/{column_id} `_ :rtype: :class:`github.ProjectColumn.ProjectColumn` :param id: integer """ headers, data = self.__requester.requestJsonAndCheck( "GET", "/projects/columns/%d" % id, headers={"Accept": Consts.mediaTypeProjectsPreview}, ) return github.ProjectColumn.ProjectColumn( self.__requester, headers, data, completed=True ) def get_gist(self, id): """ :calls: `GET /gists/{id} `_ :param id: string :rtype: :class:`github.Gist.Gist` """ assert isinstance(id, str), id headers, data = self.__requester.requestJsonAndCheck("GET", f"/gists/{id}") return github.Gist.Gist(self.__requester, headers, data, completed=True) def get_gists(self, since=github.GithubObject.NotSet): """ :calls: `GET /gists/public `_ :param since: datetime.datetime format YYYY-MM-DDTHH:MM:SSZ :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Gist.Gist` """ assert since is github.GithubObject.NotSet or isinstance( since, datetime.datetime ), since url_parameters = dict() if since is not github.GithubObject.NotSet: url_parameters["since"] = since.strftime("%Y-%m-%dT%H:%M:%SZ") return github.PaginatedList.PaginatedList( github.Gist.Gist, self.__requester, "/gists/public", url_parameters ) def search_repositories( self, query, sort=github.GithubObject.NotSet, order=github.GithubObject.NotSet, **qualifiers, ): """ :calls: `GET /search/repositories `_ :param query: string :param sort: string ('stars', 'forks', 'updated') :param order: string ('asc', 'desc') :param qualifiers: keyword dict query qualifiers :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Repository.Repository` """ assert isinstance(query, str), query url_parameters = dict() if ( sort is not github.GithubObject.NotSet ): # pragma no branch (Should be covered) assert sort in ("stars", "forks", "updated"), sort url_parameters["sort"] = sort if ( order is not github.GithubObject.NotSet ): # pragma no branch (Should be covered) assert order in ("asc", "desc"), order url_parameters["order"] = order query_chunks = [] if query: # pragma no branch (Should be covered) query_chunks.append(query) for qualifier, value in qualifiers.items(): query_chunks.append(f"{qualifier}:{value}") url_parameters["q"] = " ".join(query_chunks) assert url_parameters["q"], "need at least one qualifier" return github.PaginatedList.PaginatedList( github.Repository.Repository, self.__requester, "/search/repositories", url_parameters, ) def search_users( self, query, sort=github.GithubObject.NotSet, order=github.GithubObject.NotSet, **qualifiers, ): """ :calls: `GET /search/users `_ :param query: string :param sort: string ('followers', 'repositories', 'joined') :param order: string ('asc', 'desc') :param qualifiers: keyword dict query qualifiers :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.NamedUser.NamedUser` """ assert isinstance(query, str), query url_parameters = dict() if sort is not github.GithubObject.NotSet: assert sort in ("followers", "repositories", "joined"), sort url_parameters["sort"] = sort if order is not github.GithubObject.NotSet: assert order in ("asc", "desc"), order url_parameters["order"] = order query_chunks = [] if query: query_chunks.append(query) for qualifier, value in qualifiers.items(): query_chunks.append(f"{qualifier}:{value}") url_parameters["q"] = " ".join(query_chunks) assert url_parameters["q"], "need at least one qualifier" return github.PaginatedList.PaginatedList( github.NamedUser.NamedUser, self.__requester, "/search/users", url_parameters, ) def search_issues( self, query, sort=github.GithubObject.NotSet, order=github.GithubObject.NotSet, **qualifiers, ): """ :calls: `GET /search/issues `_ :param query: string :param sort: string ('comments', 'created', 'updated') :param order: string ('asc', 'desc') :param qualifiers: keyword dict query qualifiers :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Issue.Issue` """ assert isinstance(query, str), query url_parameters = dict() if sort is not github.GithubObject.NotSet: assert sort in ("comments", "created", "updated"), sort url_parameters["sort"] = sort if order is not github.GithubObject.NotSet: assert order in ("asc", "desc"), order url_parameters["order"] = order query_chunks = [] if query: # pragma no branch (Should be covered) query_chunks.append(query) for qualifier, value in qualifiers.items(): query_chunks.append(f"{qualifier}:{value}") url_parameters["q"] = " ".join(query_chunks) assert url_parameters["q"], "need at least one qualifier" return github.PaginatedList.PaginatedList( github.Issue.Issue, self.__requester, "/search/issues", url_parameters ) def search_code( self, query, sort=github.GithubObject.NotSet, order=github.GithubObject.NotSet, highlight=False, **qualifiers, ): """ :calls: `GET /search/code `_ :param query: string :param sort: string ('indexed') :param order: string ('asc', 'desc') :param highlight: boolean (True, False) :param qualifiers: keyword dict query qualifiers :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.ContentFile.ContentFile` """ assert isinstance(query, str), query url_parameters = dict() if ( sort is not github.GithubObject.NotSet ): # pragma no branch (Should be covered) assert sort in ("indexed",), sort url_parameters["sort"] = sort if ( order is not github.GithubObject.NotSet ): # pragma no branch (Should be covered) assert order in ("asc", "desc"), order url_parameters["order"] = order query_chunks = [] if query: # pragma no branch (Should be covered) query_chunks.append(query) for qualifier, value in qualifiers.items(): query_chunks.append(f"{qualifier}:{value}") url_parameters["q"] = " ".join(query_chunks) assert url_parameters["q"], "need at least one qualifier" headers = {"Accept": Consts.highLightSearchPreview} if highlight else None return github.PaginatedList.PaginatedList( github.ContentFile.ContentFile, self.__requester, "/search/code", url_parameters, headers=headers, ) def search_commits( self, query, sort=github.GithubObject.NotSet, order=github.GithubObject.NotSet, **qualifiers, ): """ :calls: `GET /search/commits `_ :param query: string :param sort: string ('author-date', 'committer-date') :param order: string ('asc', 'desc') :param qualifiers: keyword dict query qualifiers :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Commit.Commit` """ assert isinstance(query, str), query url_parameters = dict() if sort is not github.GithubObject.NotSet: assert sort in ("author-date", "committer-date"), sort url_parameters["sort"] = sort if order is not github.GithubObject.NotSet: assert order in ("asc", "desc"), order url_parameters["order"] = order query_chunks = [] if query: query_chunks.append(query) for qualifier, value in qualifiers.items(): query_chunks.append(f"{qualifier}:{value}") url_parameters["q"] = " ".join(query_chunks) assert url_parameters["q"], "need at least one qualifier" return github.PaginatedList.PaginatedList( github.Commit.Commit, self.__requester, "/search/commits", url_parameters, headers={"Accept": Consts.mediaTypeCommitSearchPreview}, ) def search_topics(self, query, **qualifiers): """ :calls: `GET /search/topics `_ :param query: string :param qualifiers: keyword dict query qualifiers :rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Topic.Topic` """ assert isinstance(query, str), query url_parameters = dict() query_chunks = [] if query: # pragma no branch (Should be covered) query_chunks.append(query) for qualifier, value in qualifiers.items(): query_chunks.append(f"{qualifier}:{value}") url_parameters["q"] = " ".join(query_chunks) assert url_parameters["q"], "need at least one qualifier" return github.PaginatedList.PaginatedList( github.Topic.Topic, self.__requester, "/search/topics", url_parameters, headers={"Accept": Consts.mediaTypeTopicsPreview}, ) def render_markdown(self, text, context=github.GithubObject.NotSet): """ :calls: `POST /markdown `_ :param text: string :param context: :class:`github.Repository.Repository` :rtype: string """ assert isinstance(text, str), text assert context is github.GithubObject.NotSet or isinstance( context, github.Repository.Repository ), context post_parameters = {"text": text} if context is not github.GithubObject.NotSet: post_parameters["mode"] = "gfm" post_parameters["context"] = context._identity status, headers, data = self.__requester.requestJson( "POST", "/markdown", input=post_parameters ) return data def get_hook(self, name): """ :calls: `GET /hooks/{name} `_ :param name: string :rtype: :class:`github.HookDescription.HookDescription` """ assert isinstance(name, str), name headers, attributes = self.__requester.requestJsonAndCheck( "GET", f"/hooks/{name}" ) return HookDescription.HookDescription( self.__requester, headers, attributes, completed=True ) def get_hooks(self): """ :calls: `GET /hooks `_ :rtype: list of :class:`github.HookDescription.HookDescription` """ headers, data = self.__requester.requestJsonAndCheck("GET", "/hooks") return [ HookDescription.HookDescription( self.__requester, headers, attributes, completed=True ) for attributes in data ] def get_gitignore_templates(self): """ :calls: `GET /gitignore/templates `_ :rtype: list of string """ headers, data = self.__requester.requestJsonAndCheck( "GET", "/gitignore/templates" ) return data def get_gitignore_template(self, name): """ :calls: `GET /gitignore/templates/{name} `_ :rtype: :class:`github.GitignoreTemplate.GitignoreTemplate` """ assert isinstance(name, str), name headers, attributes = self.__requester.requestJsonAndCheck( "GET", f"/gitignore/templates/{name}" ) return GitignoreTemplate.GitignoreTemplate( self.__requester, headers, attributes, completed=True ) def get_emojis(self): """ :calls: `GET /emojis `_ :rtype: dictionary of type => url for emoji` """ headers, attributes = self.__requester.requestJsonAndCheck("GET", "/emojis") return attributes def create_from_raw_data(self, klass, raw_data, headers={}): """ Creates an object from raw_data previously obtained by :attr:`github.GithubObject.GithubObject.raw_data`, and optionally headers previously obtained by :attr:`github.GithubObject.GithubObject.raw_headers`. :param klass: the class of the object to create :param raw_data: dict :param headers: dict :rtype: instance of class ``klass`` """ return klass(self.__requester, headers, raw_data, completed=True) def dump(self, obj, file, protocol=0): """ Dumps (pickles) a PyGithub object to a file-like object. Some effort is made to not pickle sensitive information like the Github credentials used in the :class:`Github` instance. But NO EFFORT is made to remove sensitive information from the object's attributes. :param obj: the object to pickle :param file: the file-like object to pickle to :param protocol: the `pickling protocol `_ """ pickle.dump((obj.__class__, obj.raw_data, obj.raw_headers), file, protocol) def load(self, f): """ Loads (unpickles) a PyGithub object from a file-like object. :param f: the file-like object to unpickle from :return: the unpickled object """ return self.create_from_raw_data(*pickle.load(f)) def get_oauth_application(self, client_id, client_secret): return github.ApplicationOAuth.ApplicationOAuth( self.__requester, headers={}, attributes={"client_id": client_id, "client_secret": client_secret}, completed=False, ) def get_app(self, slug=github.GithubObject.NotSet): """ :calls: `GET /apps/{slug} `_ or `GET /app `_ :param slug: string :rtype: :class:`github.GithubApp.GithubApp` """ assert slug is github.GithubObject.NotSet or isinstance(slug, str), slug if slug is github.GithubObject.NotSet: return GithubApp.GithubApp( self.__requester, {}, {"url": "/app"}, completed=False ) else: headers, data = self.__requester.requestJsonAndCheck("GET", f"/apps/{slug}") return GithubApp.GithubApp(self.__requester, headers, data, completed=True) class GithubIntegration: """ Main class to obtain tokens for a GitHub integration. """ def __init__(self, integration_id, private_key, base_url=DEFAULT_BASE_URL): """ :param base_url: string :param integration_id: int :param private_key: string """ self.base_url = base_url self.integration_id = integration_id self.private_key = private_key assert isinstance(base_url, str), base_url def create_jwt(self, expiration=60): """ Creates a signed JWT, valid for 60 seconds by default. The expiration can be extended beyond this, to a maximum of 600 seconds. :param expiration: int :return string: """ now = int(time.time()) payload = {"iat": now, "exp": now + expiration, "iss": self.integration_id} encrypted = jwt.encode(payload, key=self.private_key, algorithm="RS256") if isinstance(encrypted, bytes): encrypted = encrypted.decode("utf-8") return encrypted def get_access_token(self, installation_id, user_id=None): """ Get an access token for the given installation id. POSTs https://api.github.com/app/installations//access_tokens :param user_id: int :param installation_id: int :return: :class:`github.InstallationAuthorization.InstallationAuthorization` """ body = {} if user_id: body = {"user_id": user_id} response = requests.post( f"{self.base_url}/app/installations/{installation_id}/access_tokens", headers={ "Authorization": f"Bearer {self.create_jwt()}", "Accept": Consts.mediaTypeIntegrationPreview, "User-Agent": "PyGithub/Python", }, json=body, ) if response.status_code == 201: return InstallationAuthorization.InstallationAuthorization( requester=None, # not required, this is a NonCompletableGithubObject headers={}, # not required, this is a NonCompletableGithubObject attributes=response.json(), completed=True, ) elif response.status_code == 403: raise GithubException.BadCredentialsException( status=response.status_code, data=response.text ) elif response.status_code == 404: raise GithubException.UnknownObjectException( status=response.status_code, data=response.text ) raise GithubException.GithubException( status=response.status_code, data=response.text ) def get_installation(self, owner, repo): """ :calls: `GET /repos/{owner}/{repo}/installation `_ :param owner: str :param repo: str :rtype: :class:`github.Installation.Installation` """ headers = { "Authorization": f"Bearer {self.create_jwt()}", "Accept": Consts.mediaTypeIntegrationPreview, "User-Agent": "PyGithub/Python", } response = requests.get( f"{self.base_url}/repos/{owner}/{repo}/installation", headers=headers, ) response_dict = response.json() return Installation.Installation(None, headers, response_dict, True)