# -*- coding: utf-8 -*- ############################ 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 # # # # 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 . # # # ################################################################################ from __future__ import absolute_import import datetime import pickle import time import requests import jwt import urllib3 from .Requester import Requester from . import AuthenticatedUser from . import NamedUser from . import Gist import github.PaginatedList from . import Repository from . import Installation from . import License from . import Topic import github.GithubObject from . import HookDescription from . import GitignoreTemplate from . import RateLimit from . import InstallationAuthorization from . import GithubException from . import Invitation from . import Consts import six 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(object): """ 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, client_id=None, client_secret=None, user_agent='PyGithub/Python', per_page=DEFAULT_PER_PAGE, api_preview=False, verify=True, retry=None): """ :param login_or_token: string :param password: string :param base_url: string :param timeout: integer :param client_id: string :param client_secret: string :param user_agent: string :param per_page: int :param verify: boolean or string :param retry: int or urllib3.util.retry.Retry object """ assert login_or_token is None or isinstance(login_or_token, (str, six.text_type)), login_or_token assert password is None or isinstance(password, (str, six.text_type)), password assert jwt is None or isinstance(jwt, (str, six.text_type)), jwt assert isinstance(base_url, (str, six.text_type)), base_url assert isinstance(timeout, six.integer_types), timeout assert client_id is None or isinstance(client_id, (str, six.text_type)), client_id assert client_secret is None or isinstance(client_secret, (str, six.text_type)), client_secret assert user_agent is None or isinstance(user_agent, (str, six.text_type)), user_agent assert isinstance(api_preview, (bool)) assert retry is None or isinstance(retry, (int)) or isinstance(retry, (urllib3.util.Retry)) self.__requester = Requester(login_or_token, password, jwt, base_url, timeout, client_id, client_secret, user_agent, per_page, api_preview, verify, retry) def __get_FIX_REPO_GET_GIT_REF(self): """ :type: bool """ return self.__requester.FIX_REPO_GET_GIT_REF def __set_FIX_REPO_GET_GIT_REF(self, value): self.__requester.FIX_REPO_GET_GIT_REF = value FIX_REPO_GET_GIT_REF = property(__get_FIX_REPO_GET_GIT_REF, __set_FIX_REPO_GET_GIT_REF) def __get_per_page(self): """ :type: int """ return self.__requester.per_page def __set_per_page(self, value): self.__requester.per_page = value # v2: Remove this property? Why should it be necessary to read/modify it after construction per_page = property(__get_per_page, __set_per_page) # 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, six.text_type)), key headers, data = self.__requester.requestJsonAndCheck( "GET", "/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_user(self, login=github.GithubObject.NotSet): """ :calls: `GET /users/:user `_ or `GET /user `_ :param login: string :rtype: :class:`github.NamedUser.NamedUser` """ assert login is github.GithubObject.NotSet or isinstance(login, (str, six.text_type)), login if login is github.GithubObject.NotSet: return AuthenticatedUser.AuthenticatedUser(self.__requester, {}, {"url": "/user"}, completed=False) else: headers, data = self.__requester.requestJsonAndCheck( "GET", "/users/" + login ) 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, six.integer_types), 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, six.text_type)), login headers, data = self.__requester.requestJsonAndCheck( "GET", "/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, six.integer_types), 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, six.text_type, int, int)), full_name_or_id url_base = "/repositories/" if isinstance(full_name_or_id, int) or isinstance(full_name_or_id, int) else "/repos/" url = "%s%s" % (url_base, full_name_or_id) if lazy: return Repository.Repository(self.__requester, {}, {"url": url}, completed=False) headers, data = self.__requester.requestJsonAndCheck( "GET", "%s%s" % (url_base, full_name_or_id) ) 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, six.integer_types), 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_gist(self, id): """ :calls: `GET /gists/:id `_ :param id: string :rtype: :class:`github.Gist.Gist` """ assert isinstance(id, (str, six.text_type)), id headers, data = self.__requester.requestJsonAndCheck( "GET", "/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, six.text_type)), 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("%s:%s" % (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, six.text_type)), 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("%s:%s" % (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, six.text_type)), 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("%s:%s" % (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, six.text_type)), 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("%s:%s" % (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, six.text_type)), query url_parameters = dict() if sort is not github.GithubObject.NotSet: # pragma no branch (Should be covered) assert sort in ('author-date', 'committer-date'), 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("%s:%s" % (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, six.text_type)), 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("%s:%s" % (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, six.text_type)), 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, six.text_type)), name headers, attributes = self.__requester.requestJsonAndCheck( "GET", "/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, six.text_type)), name headers, attributes = self.__requester.requestJsonAndCheck( "GET", "/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 optionaly 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 informations 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_installation(self, id): """ :param id: :return: """ return Installation.Installation(self.__requester, headers={}, attributes={"id": id}, completed=True) class GithubIntegration(object): """ 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, six.text_type)), 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: """ 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( "{}/app/installations/{}/access_tokens".format(self.base_url, installation_id), headers={ "Authorization": "Bearer {}".format(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": "Bearer {}".format(self.create_jwt()), "Accept": Consts.mediaTypeIntegrationPreview, "User-Agent": "PyGithub/Python", } response = requests.get( "{}/repos/{}/{}/installation".format(DEFAULT_BASE_URL, owner, repo), headers=headers ) response_dict = response.json() return Installation.Installation(None, headers, response_dict, True)