How to use the gidgethub.abc.GitHubAPI function in gidgethub

To help you get started, we’ve selected a few gidgethub examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github brettcannon / gidgethub / gidgethub / treq.py View on Github external
from typing import Any, Mapping, Tuple

from twisted.internet import defer
from twisted.web.http_headers import Headers

import treq

from . import abc as gh_abc


class GitHubAPI(gh_abc.GitHubAPI):
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        from twisted.internet import reactor
        self._reactor = reactor
        super().__init__(*args, **kwargs)

    async def _request(self, method: str, url: str, headers: Mapping[str, str],
                       body: bytes = b'') -> Tuple[int, Mapping[str, str], bytes]:
        # We need to encode the headers to a format that Twisted will like.
        # As a note: treq will set a content-length even if we do, so we need
        # to strip any content-length header.
        headers = Headers(
            {
                k.encode('utf-8'): [v.encode('utf-8')]
                for k, v in headers.items()
                if k.lower() != 'content-length'
            }
github brettcannon / gidgethub / gidgethub / tornado.py View on Github external
from typing import Mapping, Tuple, Union, List, Dict, Any

from tornado import gen
from tornado import httpclient

from . import abc as gh_abc


class GitHubAPI(gh_abc.GitHubAPI):

    async def _request(self, method: str, url: str, headers: Mapping[str, str],
                       body: bytes = b'') -> Tuple[int, Mapping[str, str], bytes]:
        """Make an HTTP request."""
        # Setting 'body' to None fails type checking, so only add a 'body' argument if necessary.
        args: List[Union[str, Dict[Any, Any], bytes]] = [url, method, dict(headers)]
        if method != "GET" and body:
            args.append(body)
        # The below line is skipped from mypy because Tornado's HTTPRequest signature
        # requires many types of arguments some of which are internal to it
        # adding all of them to the `args` would be impractical.
        request = httpclient.HTTPRequest(*args)  # type: ignore
        # Since Tornado has designed AsyncHTTPClient to be a singleton, there's
        # no reason not to simply instantiate it every time.
        client = httpclient.AsyncHTTPClient()
        response = await client.fetch(request, raise_error=False)
github bioconda / bioconda-utils / bioconda_utils / githubhandler.py View on Github external
self.token = token
        #: If set, no actual modifying actions are taken
        self.dry_run = dry_run
        #: The installation ID if this instance is connected to an App
        self.installation = installation
        #: Owner of the Repo
        self.user = to_user
        #: Name of the Repo
        self.repo = to_repo
        #: Default variables for API calls
        self.var_default = {'user': to_user,
                            'repo': to_repo}

        # filled in by login():
        #: Gidgethub API object
        self.api: gidgethub.abc.GitHubAPI = None
        #: Login username
        self.username: str = None
        #: User avatar URL
        self.avatar_url: str = None
github brettcannon / gidgethub / gidgethub / aiohttp.py View on Github external
import asyncio
from typing import Any, Mapping, Tuple

import aiohttp

from . import abc as gh_abc


class GitHubAPI(gh_abc.GitHubAPI):

    def __init__(self, session: aiohttp.ClientSession, *args: Any,
                 **kwargs: Any) -> None:
        self._session = session
        super().__init__(*args, **kwargs)

    async def _request(self, method: str, url: str, headers: Mapping[str, str],
                       body: bytes = b'') -> Tuple[int, Mapping[str, str], bytes]:
        async with self._session.request(method, url, headers=headers,
                                         data=body) as response:
            return response.status, response.headers, await response.read()

    async def sleep(self, seconds: float) -> None:
        await asyncio.sleep(seconds)