Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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'
}
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)
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
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)