github.py

"""This module contains the main interfaces to the API."""

import json
import re
import typing as t

import uritemplate  # type: ignore

from . import apps
from . import auths
from . import decorators
from . import events
from . import gists
from . import issues
from . import licenses
from . import models
from . import notifications
from . import orgs
from . import projects
from . import pulls
from . import search
from . import session
from . import structs
from . import users
from . import utils
from .decorators import requires_app_credentials
from .decorators import requires_auth
from .decorators import requires_basic_auth
from .repos import invitation
from .repos import repo

_pubsub_re = re.compile(
    r"https?://[\w\d\-\.\:]+/\w[\w-]+\w/[\w\._-]+/events/\w+"
)


class GitHub(models.GitHubCore):
    """Stores all the session information.

    There are two ways to log into the GitHub API

    .. code-block:: python

        from github3 import login
        g = login(user, password)
        g = login(token=token)
        g = login(user, token=token)

    or

    .. code-block:: python

        from github3 import GitHub
        g = GitHub(user, password)
        g = GitHub(token=token)
        g = GitHub(user, token=token)

    This is simple backward compatibility since originally there was no way to
    call the GitHub object with authentication parameters.
    """

    def __init__(
        self, username="", password="", token="", session=None, api_version=""
    ):
        """Create a new GitHub instance to talk to the API.

        :param str api_version:
            API version to send with X-GitHub-Api-Version header.
            See https://docs.github.com/en/rest/overview/api-versions
            for details about API versions.
        """
        super().__init__({}, session or self.new_session())

        if api_version:
            self.session.headers.update({"X-GitHub-Api-Version": api_version})

        if token:
            self.login(username, token=token)
        elif username and password:
            self.login(username, password)

    def _repr(self):
        if self.session.auth:
            return f"<GitHub [{self.session.auth!r}]>"
        return f"<Anonymous GitHub at 0x{id(self):x}>"

    @requires_auth
    def activate_membership(self, organization):
        """Activate the membership to an organization.

        :param organization:
            the organization or organization login for which to activate the
            membership
        :type organization:
            str
        :type organization:
            :class:`~github3.orgs.Organization`
        :returns:
            the activated membership
        :rtype:
            :class:`~github3.orgs.Membership`
        """
        organization_name = getattr(organization, "login", organization)
        url = self._build_url(
            "user", "memberships", "orgs", organization_name
        )
        data = {"state": "active"}
        _json = self._json(self._patch(url, data=json.dumps(data)), 200)
        return self._instance_or_null(orgs.Membership, _json)

    @requires_auth
    def add_email_addresses(self, addresses=[]):
        """Add the addresses to the authenticated user's account.

        :param list addresses:
            (optional), email addresses to be added
        :returns:
            list of email objects
        :rtype:
            [:class:`~github3.users.Email`]
        """
        json = []
        if addresses:
            url = self._build_url("user", "emails")
            json = self._json(self._post(url, data=addresses), 201)
        return [users.Email(email, self) for email in json] if json else []

    def all_events(self, number=-1, etag=None):
        """Iterate over public events.

        :param int number:
            (optional), number of events to return. Default: -1
            returns all available events
        :param str etag:
            (optional), ETag from a previous request to the same endpoint
        :returns:
            generator of events
        :rtype:
            :class:`~github3.events.Event`
        """
        url = self._build_url("events")
        return self._iter(int(number), url, events.Event, etag=etag)

    def all_organizations(
        self, number=-1, since=None, etag=None, per_page=None
    ):
        """Iterate over every organization in the order they were created.

        :param int number:
            (optional), number of organizations to return.
            Default: -1, returns all of them
        :param int since:
            (optional), last organization id seen (allows restarting an
            iteration)
        :param str etag:
            (optional), ETag from a previous request to the same endpoint
        :param int per_page:
            (optional), number of organizations to list per request
        :returns:
            generator of organizations
        :rtype:
            :class:`~github3.orgs.ShortOrganization`
        """
        url = self._build_url("organizations")
        return self._iter(
            int(number),
            url,
            orgs.ShortOrganization,
            params={"since": since, "per_page": per_page},
            etag=etag,
        )

    def all_repositories(
        self, number=-1, since=None, etag=None, per_page=None
    ):
        """Iterate over every repository in the order they were created.

        :param int number:
            (optional), number of repositories to return.
            Default: -1, returns all of them
        :param int since:
            (optional), last repository id seen (allows restarting an
            iteration)
        :param str etag:
            (optional), ETag from a previous request to the same endpoint
        :param int per_page:
            (optional), number of repositories to list per request
        :returns:
            generator of repositories
        :rtype:
            :class:`~github3.repos.repo.ShortRepository`
        """
        url = self._build_url("repositories")
        return self._iter(
            int(number),
            url,
            repo.ShortRepository,
            params={"since": since, "per_page": per_page},
            etag=etag,
        )

    def all_users(self, number=-1, etag=None, per_page=None, since=None):
        """Iterate over every user in the order they signed up for GitHub.

        .. versionchanged:: 1.0.0

            Inserted the ``since`` parameter after the ``number`` parameter.

        :param int number:
            (optional), number of users to return. Default: -1, returns all of
            them
        :param int since:
            (optional), ID of the last user that you've seen.
        :param str etag:
            (optional), ETag from a previous request to the same endpoint
        :param int per_page:
            (optional), number of users to list per request
        :returns:
            generator of users
        :rtype:
            :class:`~github3.users.ShortUser`
        """
        url = self._build_url("users")
        return self._iter(
            int(number),
            url,
            users.ShortUser,
            etag=etag,
            params={"per_page": per_page, "since": since},
        )

    def app(self, app_slug):
        """Retrieve information about a specific app using its "slug".

        .. versionadded:: 1.2.0

        .. seealso::

            `Get a single GitHub App`_
                API Documentation

        :param app_slug:
            The identifier for the specific slug, e.g.,
            ``test-github3-py-apps``.
        :returns:
            The app if and only if it is public.
        :rtype:
            :class:`~github3.apps.App`

        .. _Get a single GitHub App:
            https://developer.github.com/v3/apps/#get-a-single-github-app
        """
        headers = apps.APP_PREVIEW_HEADERS
        url = self._build_url("apps", str(app_slug))
        json = self._json(self._get(url, headers=headers), 200)
        return self._instance_or_null(apps.App, json)

    @decorators.requires_app_bearer_auth
    def app_installation(self, installation_id):
        """Retrieve a specific App installation by its ID.

        .. versionadded: 1.2.0

        .. seealso::

            `Get a single installation`_
                API Documentation

        :param int installation_id:
            The ID of the specific installation.
        :returns:
            The installation.
        :rtype:
            :class:`~github3.apps.Installation`

        .. _Get a single installation:
            https://developer.github.com/v3/apps/#get-a-single-installation
        """
        url = self._build_url("app", "installations", str(installation_id))
        json = self._json(
            self._get(url, headers=apps.APP_PREVIEW_HEADERS), 200
        )
        return self._instance_or_null(apps.Installation, json)

    @decorators.requires_app_bearer_auth
    def app_installations(self, number=-1):
        """Retrieve the list of installations for the authenticated app.

        .. versionadded:: 1.2.0

        .. seealso::

            `Find installations`_
                API Documentation

        :returns:
            The installations of the authenticated App.
        :rtype:
            :class:`~github3.apps.Installation`

        .. _Find installations:
            https://developer.github.com/v3/apps/#find-installations
        """
        url = self._build_url("app", "installations")
        return self._iter(
            int(number),
            url,
            apps.Installation,
            headers=apps.APP_PREVIEW_HEADERS,
        )

    @decorators.requires_app_bearer_auth
    def app_installation_for_organization(self, organization):
        """Retrieve an App installation for a specific organization.

        .. versionadded:: 1.2.0

        .. seealso::

            `Find organization installation`_
                API Documentation

        :param str organization:
            The name of the organization.
        :returns:
            The installation
        :rtype:
            :class:`~github3.apps.Installation`

        .. _Find organization installation:
            https://developer.github.com/v3/apps/#find-organization-installation
        """
        url = self._build_url("orgs", organization, "installation")
        json = self._json(
            self._get(url, headers=apps.APP_PREVIEW_HEADERS), 200
        )
        return self._instance_or_null(apps.Installation, json)

    @decorators.requires_app_bearer_auth
    def app_installation_for_repository(self, owner, repository):
        """Retrieve an App installation for a specific repository.

        .. versionadded:: 1.2.0

        .. seealso::

            `Find repository installation`_
                API Documentation

        :param str owner:
            The name of the owner.
        :param str repostory:
            The name of the repository.
        :returns:
            The installation
        :rtype:
            :class:`~github3.apps.Installation`

        .. _Find repository installation:
            https://developer.github.com/v3/apps/#find-repository-installation
        """
        owner = getattr(owner, "login", str(owner))
        repository = getattr(repository, "name", repository)
        url = self._build_url("repos", owner, repository, "installation")
        json = self._json(
            self._get(url, headers=apps.APP_PREVIEW_HEADERS), 200
        )
        return self._instance_or_null(apps.Installation, json)

    @decorators.requires_app_bearer_auth
    def app_installation_for_user(self, user):
        """Retrieve an App installation for a specific repository.

        .. versionadded:: 1.2.0

        .. seealso::

            `Find user installation`_
                API Documentation

        :param str user:
            The name of the user.
        :returns:
            The installation
        :rtype:
            :class:`~github3.apps.Installation`

        .. _Find user installation:
            https://developer.github.com/v3/apps/#find-user-installation
        """
        user = getattr(user, "login", str(user))
        url = self._build_url("users", user, "installation")
        json = self._json(
            self._get(url, headers=apps.APP_PREVIEW_HEADERS), 200
        )
        return self._instance_or_null(apps.Installation, json)

    @decorators.requires_app_installation_auth
    def app_installation_repos(self, number=-1, etag=None):
        """Retrieve repositories accessible by app installation.

        .. versionadded:: 3.2.1

        .. seealso::

            `List repositories accessible to the app installation`_
                API Documentation

        :returns:
            The repositories accessible to the app installation
        :rtype:
            :class:`~github3.repos.repo.ShortRepository`

        .. _List repositories accessible to the app installation:
            https://docs.github.com/en/rest/apps/installations#list-repositories-accessible-to-the-app-installation
        """
        url = self._build_url("installation", "repositories")
        return self._iter(
            count=int(number),
            url=url,
            cls=repo.ShortRepository,
            params=None,
            etag=etag,
            list_key="repositories",
        )

    @decorators.requires_app_bearer_auth
    def authenticated_app(self):
        """Retrieve information about the current app.

        .. versionadded:: 1.2.0

        .. seealso::

            `Get the authenticated GitHub App`_
                API Documentation

        :returns:
            Metadata about the application
        :rtype:
            :class:`~github3.apps.App`

        .. _Get the authenticated GitHub App:
            https://developer.github.com/v3/apps/#get-the-authenticated-github-app
        """
        headers = apps.APP_PREVIEW_HEADERS
        json = self._json(
            self._get(self._build_url("app"), headers=headers), 200
        )
        return self._instance_or_null(apps.App, json)

    @requires_basic_auth
    def authorization(self, id_num):
        """Get information about authorization ``id``.

        :param int id_num:
            (required), unique id of the authorization
        :returns:
            :class:`~github3.auths.Authorization`
        """
        json = None
        if int(id_num) > 0:
            url = self._build_url("authorizations", str(id_num))
            json = self._json(self._get(url), 200)
        return self._instance_or_null(auths.Authorization, json)

    @requires_basic_auth
    def authorizations(self, number=-1, etag=None):
        """Iterate over authorizations for the authenticated user.

        .. note::

            This will return a 404 if you are using a token for
            authentication.

        :param int number:
            (optional), number of authorizations to return.
            Default: -1 returns all available authorizations
        :param str etag:
            (optional), ETag from a previous request to the same endpoint
        :returns:
            generator of authorizations
        :rtype:
            :class:`~github3.auths.Authorization`
        """
        url = self._build_url("authorizations")
        return self._iter(int(number), url, auths.Authorization, etag=etag)

    def authorize(
        self,
        username,
        password,
        scopes=None,
        note="",
        note_url="",
        client_id="",
        client_secret="",
    ):
        """Obtain an authorization token.

        The retrieved token will allow future consumers to use the API without
        a username and password.

        :param str username:
            (required)
        :param str password:
            (required)
        :param list scopes:
            (optional), areas you want this token to apply to, i.e., 'gist',
            'user'
        :param str note:
            (optional), note about the authorization
        :param str note_url:
            (optional), url for the application
        :param str client_id:
            (optional), 20 character OAuth client key for which to create a
            token
        :param str client_secret:
            (optional), 40 character OAuth client secret for which to create
            the token
        :returns:
            created authorization
        :rtype:
            :class:`~github3.auths.Authorization`
        """
        json = None

        if username and password:
            url = self._build_url("authorizations")
            data = {
                "note": note,
                "note_url": note_url,
                "client_id": client_id,
                "client_secret": client_secret,
            }
            if scopes:
                data["scopes"] = scopes

            with self.session.temporary_basic_auth(username, password):
                json = self._json(self._post(url, data=data), 201)

        return self._instance_or_null(auths.Authorization, json)

    @requires_auth
    def blocked_users(
        self, number: int = -1, etag: t.Optional[str] = None
    ) -> t.Iterator[users.ShortUser]:
        """Iterate over the users blocked by this organization.

        .. versionadded:: 2.1.0

        :param int number:
            (optional), number of users to iterate over.  Default: -1 iterates
            over all values
        :param str etag:
            (optional), ETag from a previous request to the same endpoint
        :returns:
            generator of the members of this team
        :rtype:
            :class:`~github3.users.ShortUser`
        """
        url = self._build_url("user", "blocks")
        return self._iter(int(number), url, users.ShortUser, etag=etag)

    @requires_auth
    def block(self, username: users.UserLike) -> bool:
        """Block a specific user from an organization.

        .. versionadded:: 2.1.0

        :parameter str username:
            Name (or user-like instance) of the user to block.
        :returns:
            True if successful, Fales otherwise
        :rtype:
            bool
        """
        url = self._build_url("user", "blocks", str(username))
        return self._boolean(self._put(url), 204, 404)

    @requires_auth
    def unblock(self, username: users.UserLike) -> bool:
        """Unblock a specific user from an organization.

        .. versionadded:: 2.1.0

        :parameter str username:
            Name (or user-like instance) of the user to unblock.
        :returns:
            True if successful, Fales otherwise
        :rtype:
            bool
        """
        url = self._build_url("user", "blocks", str(username))
        return self._boolean(self._delete(url), 204, 404)

    @requires_auth
    def is_blocking(self, username: users.UserLike) -> bool:
        """Check if this organization is blocking a specific user.

        .. versionadded:: 2.1.0

        :parameter str username:
            Name (or user-like instance) of the user to unblock.
        :returns:
            True if successful, Fales otherwise
        :rtype:
            bool
        """
        url = self._build_url("user", "blocks", str(username))
        return self._boolean(self._get(url), 204, 404)

    def check_authorization(self, access_token):
        """Check an authorization created by a registered application.

        OAuth applications can use this method to check token validity
        without hitting normal rate limits because of failed login attempts.
        If the token is valid, it will return True, otherwise it will return
        False.

        :returns:
            True if token is valid, False otherwise
        :rtype:
            bool
        """
        p = self.session.params
        auth = (p.get("client_id"), p.get("client_secret"))
        if access_token and auth:
            url = self._build_url(
                "applications", str(auth[0]), "tokens", str(access_token)
            )
            resp = self._get(
                url,
                auth=auth,
                params={"client_id": None, "client_secret": None},
            )
            return self._boolean(resp, 200, 404)
        return False

    @requires_auth
    def create_gist(self, description, files, public=True):
        """Create a new gist.

        .. versionchanged:: 1.1.0

            Per `GitHub's recent announcement`_ authentication is now required
            for creating gists.

        .. _GitHub's recent announcement:
            https://blog.github.com/2018-02-18-deprecation-notice-removing-anonymous-gist-creation/

        :param str description:
            (required), description of gist
        :param dict files:
            (required), file names with associated dictionaries for content,
            e.g. ``{'spam.txt': {'content': 'File contents ...'}}``
        :param bool public:
            (optional), make the gist public if True
        :returns:
            the created gist if successful, otherwise ``None``
        :rtype:
            created gist
        :rtype:
            :class:`~github3.gists.gist.Gist`
        """
        new_gist = {
            "description": description,
            "public": public,
            "files": files,
        }
        url = self._build_url("gists")
        json = self._json(self._post(url, data=new_gist), 201)
        return self._instance_or_null(gists.Gist, json)

    @requires_auth
    def create_gpg_key(self, armored_public_key):
        """Create a new GPG key.

        .. versionadded:: 1.2.0

        :param str armored_public_key:
            (required), your GPG key, generated in ASCII-armored format
        :returns:
            the created GPG key if successful, otherwise ``None``
        :rtype:
            :class:`~github3.users.GPGKey`
        """
        url = self._build_url("user", "gpg_keys")
        data = {"armored_public_key": armored_public_key}
        json = self._json(self._post(url, data=data), 201)
        return self._instance_or_null(users.GPGKey, json)

    @requires_auth
    def create_issue(
        self,
        owner,
        repository,
        title,
        body=None,
        assignee=None,
        milestone=None,
        labels=[],
        assignees=None,
    ):
        """Create an issue on the repository.

        .. note::

            ``body``, ``assignee``, ``assignees``, ``milestone``, ``labels``
            are all optional.

        .. warning::

            This method retrieves the repository first and then uses it to
            create an issue. If you're making several issues, you should use
            :py:meth:`repository <github3.github.GitHub.repository>` and then
            use :py:meth:`create_issue
            <github3.repos.repo.Repository.create_issue>`

        :param str owner:
            (required), login of the owner
        :param str repository:
            (required), repository name
        :param str title:
            (required), Title of issue to be created
        :param str body:
            (optional), The text of the issue, markdown formatted
        :param str assignee:
            (optional), Login of person to assign the issue to
        :param assignees:
            (optional), logins of the users to assign the issue to
        :param int milestone:
            (optional), id number of the milestone to attribute this issue to
            (e.g. if ``m`` is a :class:`~github3.issues.Milestone` object,
            ``m.number`` is what you pass here.)
        :param list labels:
            (optional), List of label names.
        :returns:
            created issue
        :rtype:
            :class:`~github3.issues.ShortIssue`
        """
        repo = None
        if owner and repository and title:
            repo = self.repository(owner, repository)

        if repo is not None:
            return repo.create_issue(
                title, body, assignee, milestone, labels, assignees
            )

        return self._instance_or_null(issues.ShortIssue, None)

    @requires_auth
    def create_key(self, title, key, read_only=False):
        """Create a new key for the authenticated user.

        :param str title:
            (required), key title
        :param str key:
            (required), actual key contents, accepts path
            as a string or file-like object
        :param bool read_only:
            (optional), restrict key access to read-only, default to False
        :returns:
            created key
        :rtype:
            :class:`~github3.users.Key`
        """
        json = None

        if title and key:
            data = {"title": title, "key": key, "read_only": read_only}
            url = self._build_url("user", "keys")
            req = self._post(url, data=data)
            json = self._json(req, 201)
        return self._instance_or_null(users.Key, json)

    @requires_auth
    def create_repository(
        self,
        name,
        description="",
        homepage="",
        private=False,
        has_issues=True,
        has_wiki=True,
        auto_init=False,
        gitignore_template="",
        has_projects=True,
    ):
        """Create a repository for the authenticated user.

        :param str name:
            (required), name of the repository

            .. warning: this be no longer than 100 characters
        :param str description:
            (optional)
        :param str homepage:
            (optional)
        :param str private:
            (optional), If ``True``, create a private repository. API default:
            ``False``
        :param bool has_issues:
            (optional), If ``True``, enable issues for this repository. API
            default: ``True``
        :param bool has_wiki:
            (optional), If ``True``, enable the wiki for this repository. API
            default: ``True``
        :param bool auto_init:
            (optional), auto initialize the repository
        :param str gitignore_template:
            (optional), name of the git template to use; ignored if auto_init =
            False.
        :param bool has_projects:
            (optional), If ``True``, enable projects for this repository. API
            default: ``True``
        :returns:
            created repository
        :rtype:
            :class:`~github3.repos.repo.Repository`
        """
        url = self._build_url("user", "repos")
        data = {
            "name": name,
            "description": description,
            "homepage": homepage,
            "private": private,
            "has_issues": has_issues,
            "has_wiki": has_wiki,
            "auto_init": auto_init,
            "gitignore_template": gitignore_template,
            "has_projects": has_projects,
        }
        json = self._json(self._post(url, data=data), 201)
        return self._instance_or_null(repo.Repository, json)

    @requires_auth
    def delete_email_addresses(self, addresses=[]):
        """Delete the specified addresses the authenticated user's account.

        :param list addresses:
            (optional), email addresses to be removed
        :returns:
            True if successful, False otherwise
        :rtype:
            bool
        """
        url = self._build_url("user", "emails")
        return self._boolean(
            self._delete(url, data=json.dumps(addresses)), 204, 404
        )

    @requires_auth
    def emails(self, number=-1, etag=None):
        """Iterate over email addresses for the authenticated user.

        :param int number:
            (optional), number of email addresses to return.
            Default: -1 returns all available email addresses
        :param str etag:
            (optional), ETag from a previous request to the same endpoint
        :returns:
            generator of emails
        :rtype:
            :class:`~github3.users.Email`
        """
        url = self._build_url("user", "emails")
        return self._iter(int(number), url, users.Email, etag=etag)

    def emojis(self):
        """Retrieve a dictionary of all of the emojis that GitHub supports.

        :returns:
            dictionary where the key is what would be in between the
            colons and the value is the URL to the image, e.g.,

            .. code-block:: python

                {
                    '+1': 'https://github.global.ssl.fastly.net/images/...',
                    # ...
                }
        """
        url = self._build_url("emojis")
        return self._json(self._get(url), 200, include_cache_info=False)

    @requires_basic_auth
    def feeds(self):
        """List GitHub's timeline resources in Atom format.

        :returns:
            dictionary parsed to include URITemplates
        :rtype:
            dict
        """

        def replace_href(feed_dict):
            if not feed_dict:
                return feed_dict
            ret_dict = {}
            # Let's pluck out what we're most interested in, the href value
            href = feed_dict.pop("href", None)
            # Then we update the return dictionary with the rest of the values
            ret_dict.update(feed_dict)
            if href is not None:
                # So long as there is something to template, let's template it
                ret_dict["href"] = uritemplate.URITemplate(href)
            return ret_dict

        url = self._build_url("feeds")
        json = self._json(self._get(url), 200, include_cache_info=False)
        if json is None:  # If something went wrong, get out early
            return None

        # We have a response body to parse
        feeds = {}

        # Let's pop out the old links so we don't have to skip them below
        old_links = json.pop("_links", {})
        _links = {}
        # If _links is in the response JSON, iterate over that and recreate it
        # so that any templates contained inside can be turned into
        # URITemplates
        for key, value in old_links.items():
            if isinstance(value, list):
                # If it's an array/list of links, let's replace that with a
                # new list of links
                _links[key] = [replace_href(d) for d in value]
            else:
                # Otherwise, just use the new value
                _links[key] = replace_href(value)

        # Start building up our return dictionary
        feeds["_links"] = _links

        for key, value in json.items():
            # This should roughly be the same logic as above.
            if isinstance(value, list):
                feeds[key] = [uritemplate.URITemplate(v) for v in value]
            else:
                feeds[key] = uritemplate.URITemplate(value)

        return feeds

    @requires_auth
    def follow(self, username):
        """Make the authenticated user follow the provided username.

        :param str username:
            (required), user to follow
        :returns:
            True if successful, False otherwise
        :rtype:
            bool
        """
        resp = False
        if username:
            url = self._build_url("user", "following", username)
            resp = self._boolean(self._put(url), 204, 404)
        return resp

    def followed_by(self, username, number=-1, etag=None):
        """Iterate over users being followed by ``username``.

        .. versionadded:: 1.0.0

            This replaces iter_following('sigmavirus24').

        :param str username:
            (required), login of the user to check
        :param int number:
            (optional), number of people to return. Default: -1 returns all
            people you follow
        :param str etag:
            (optional), ETag from a previous request to the same endpoint
        :returns:
            generator of users
        :rtype:
            :class:`~github3.users.ShortUser`
        """
        url = self._build_url("users", username, "following")
        return self._iter(int(number), url, users.ShortUser, etag=etag)

    @requires_auth
    def followers(self, number=-1, etag=None):
        """Iterate over followers of the authenticated user.

        .. versionadded:: 1.0.0

            This replaces iter_followers().

        :param int number:
            (optional), number of followers to return. Default: -1 returns all
            followers
        :param str etag:
            (optional), ETag from a previous request to the same endpoint
        :returns:
            generator of followers