mirror of
https://github.com/RGBCube/GitHubWrapper
synced 2025-05-19 23:45:09 +00:00
Try 3.8 fixing?
This commit is contained in:
parent
a3aa1a51a8
commit
720b9f5e8d
3 changed files with 22 additions and 21 deletions
|
@ -8,7 +8,7 @@ import re
|
||||||
from collections import namedtuple
|
from collections import namedtuple
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from types import SimpleNamespace
|
from types import SimpleNamespace
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING, Dict, Union, List
|
||||||
import platform
|
import platform
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
|
@ -61,7 +61,7 @@ trace_config = aiohttp.TraceConfig()
|
||||||
trace_config.on_request_start.append(on_req_start)
|
trace_config.on_request_start.append(on_req_start)
|
||||||
trace_config.on_request_end.append(on_req_end)
|
trace_config.on_request_end.append(on_req_end)
|
||||||
|
|
||||||
async def make_session(*, headers: dict[str, str], authorization: aiohttp.BasicAuth | None) -> aiohttp.ClientSession:
|
async def make_session(*, headers: Dict[str, str], authorization: Union[aiohttp.BasicAuth, None]) -> aiohttp.ClientSession:
|
||||||
"""This makes the ClientSession, attaching the trace config and ensuring a UA header is present."""
|
"""This makes the ClientSession, attaching the trace config and ensuring a UA header is present."""
|
||||||
if not headers.get('User-Agent'):
|
if not headers.get('User-Agent'):
|
||||||
headers['User-Agent'] = f'Github-API-Wrapper (https://github.com/VarMonke/Github-Api-Wrapper) @ {__version__} Python {platform.python_version()} aiohttp {aiohttp.__version__}'
|
headers['User-Agent'] = f'Github-API-Wrapper (https://github.com/VarMonke/Github-Api-Wrapper) @ {__version__} Python {platform.python_version()} aiohttp {aiohttp.__version__}'
|
||||||
|
@ -81,7 +81,7 @@ class Paginator:
|
||||||
self.session = session
|
self.session = session
|
||||||
self.response = response
|
self.response = response
|
||||||
self.should_paginate = bool(self.response.headers.get('Link', False))
|
self.should_paginate = bool(self.response.headers.get('Link', False))
|
||||||
types: dict[str, APIObject] = {
|
types: Dict[str, APIObject] = {
|
||||||
'user': User,
|
'user': User,
|
||||||
'gist' : Gist,
|
'gist' : Gist,
|
||||||
'repo' : Repository
|
'repo' : Repository
|
||||||
|
@ -93,15 +93,15 @@ class Paginator:
|
||||||
self.next_page = self.current_page + 1
|
self.next_page = self.current_page + 1
|
||||||
self.parse_header(response)
|
self.parse_header(response)
|
||||||
|
|
||||||
async def fetch_page(self, link) -> dict[str, str | int]:
|
async def fetch_page(self, link) -> Dict[str, Union[str, int]]:
|
||||||
"""Fetches a specific page and returns the JSON."""
|
"""Fetches a specific page and returns the JSON."""
|
||||||
return await (await self.session.get(link)).json()
|
return await (await self.session.get(link)).json()
|
||||||
|
|
||||||
async def early_return(self) -> list[APIObject]:
|
async def early_return(self) -> List[APIObject]:
|
||||||
# I don't rightly remember what this does differently, may have a good ol redesign later
|
# I don't rightly remember what this does differently, may have a good ol redesign later
|
||||||
return [self.target_type(data, self.session) for data in await self.response.json()]
|
return [self.target_type(data, self.session) for data in await self.response.json()]
|
||||||
|
|
||||||
async def exhaust(self) -> list[APIObject]:
|
async def exhaust(self) -> List[APIObject]:
|
||||||
"""Iterates through all of the pages for the relevant object and creates them."""
|
"""Iterates through all of the pages for the relevant object and creates them."""
|
||||||
if self.should_paginate:
|
if self.should_paginate:
|
||||||
return await self.early_return()
|
return await self.early_return()
|
||||||
|
@ -121,12 +121,12 @@ class Paginator:
|
||||||
raise WillExceedRatelimit(response, self.max_page)
|
raise WillExceedRatelimit(response, self.max_page)
|
||||||
self.bare_link = groups[0][0][:-1]
|
self.bare_link = groups[0][0][:-1]
|
||||||
|
|
||||||
GithubUserData = GithubRepoData = GithubIssueData = GithubOrgData = GithubGistData = dict[str, str | int]
|
GithubUserData = GithubRepoData = GithubIssueData = GithubOrgData = GithubGistData = Dict[str, Union [str, int]]
|
||||||
|
|
||||||
class http:
|
class http:
|
||||||
def __init__(self, headers: dict[str, str | int], auth: aiohttp.BasicAuth | None):
|
def __init__(self, headers: Dict[str, Union[str, int]], auth: Union[aiohttp.BasicAuth, None]):
|
||||||
if not headers.get('User-Agent'):
|
if not headers.get('User-Agent'):
|
||||||
headers['User-Agent'] = f'Github-API-Wrapper (https://github.com/VarMonke/Github-Api-Wrapper) @ {__version__} Python {platform.python_version()} aiohttp {aiohttp.__version__}'
|
headers['User-Agent'] = f'Github-API-Wrapper (https://github.com/VarMonke/Github-Api-Wrapper) @ {__version__} Python/{platform.python_version()} aiohttp/{aiohttp.__version__}'
|
||||||
self._rates = Rates('', '', '', '', '')
|
self._rates = Rates('', '', '', '', '')
|
||||||
self.headers = headers
|
self.headers = headers
|
||||||
self.auth = auth
|
self.auth = auth
|
||||||
|
@ -142,7 +142,7 @@ class http:
|
||||||
)
|
)
|
||||||
return self
|
return self
|
||||||
|
|
||||||
def update_headers(self, *, flush: bool = False, new_headers: dict[str, str | int]):
|
def update_headers(self, *, flush: bool = False, new_headers: Dict[str, Union[str, int]]):
|
||||||
if flush:
|
if flush:
|
||||||
from multidict import CIMultiDict
|
from multidict import CIMultiDict
|
||||||
self.session.headers = CIMultiDict(**new_headers)
|
self.session.headers = CIMultiDict(**new_headers)
|
||||||
|
@ -181,21 +181,21 @@ class http:
|
||||||
return await result.json()
|
return await result.json()
|
||||||
raise UserNotFound
|
raise UserNotFound
|
||||||
|
|
||||||
async def get_user_repos(self, _user: User) -> list[GithubRepoData]:
|
async def get_user_repos(self, _user: User) -> List[GithubRepoData]:
|
||||||
result = await self.session.get(USER_REPOS_URL.format(_user.login))
|
result = await self.session.get(USER_REPOS_URL.format(_user.login))
|
||||||
if 200 <= result.status <= 299:
|
if 200 <= result.status <= 299:
|
||||||
return await result.json()
|
return await result.json()
|
||||||
else:
|
else:
|
||||||
print('This shouldn\'t be reachable')
|
print('This shouldn\'t be reachable')
|
||||||
|
|
||||||
async def get_user_gists(self, _user: User) -> list[GithubGistData]:
|
async def get_user_gists(self, _user: User) -> List[GithubGistData]:
|
||||||
result = await self.session.get(USER_GISTS_URL.format(_user.login))
|
result = await self.session.get(USER_GISTS_URL.format(_user.login))
|
||||||
if 200 <= result.status <= 299:
|
if 200 <= result.status <= 299:
|
||||||
return await result.json()
|
return await result.json()
|
||||||
else:
|
else:
|
||||||
print('This shouldn\'t be reachable')
|
print('This shouldn\'t be reachable')
|
||||||
|
|
||||||
async def get_user_orgs(self, _user: User) -> list[GithubOrgData]:
|
async def get_user_orgs(self, _user: User) -> List[GithubOrgData]:
|
||||||
result = await self.session.get(USER_ORGS_URL.format(_user.login))
|
result = await self.session.get(USER_ORGS_URL.format(_user.login))
|
||||||
if 200 <= result.status <= 299:
|
if 200 <= result.status <= 299:
|
||||||
return await result.json()
|
return await result.json()
|
||||||
|
@ -216,7 +216,7 @@ class http:
|
||||||
return await result.json()
|
return await result.json()
|
||||||
raise IssueNotFound
|
raise IssueNotFound
|
||||||
|
|
||||||
async def delete_repo(self, owner: str,repo_name: str) -> None:
|
async def delete_repo(self, owner: str, repo_name: str) -> None:
|
||||||
"""Deletes a Repo from the given owner and repo name."""
|
"""Deletes a Repo from the given owner and repo name."""
|
||||||
result = await self.session.delete(REPO_URL.format(owner, repo_name))
|
result = await self.session.delete(REPO_URL.format(owner, repo_name))
|
||||||
if 204 <= result.status <= 299:
|
if 204 <= result.status <= 299:
|
||||||
|
@ -251,7 +251,7 @@ class http:
|
||||||
async def create_gist(
|
async def create_gist(
|
||||||
self,
|
self,
|
||||||
*,
|
*,
|
||||||
files: list['File'] = [],
|
files: List['File'] = [],
|
||||||
description: str = 'Default description',
|
description: str = 'Default description',
|
||||||
public: bool = False
|
public: bool = False
|
||||||
) -> GithubGistData:
|
) -> GithubGistData:
|
||||||
|
|
|
@ -7,6 +7,7 @@ __all__ = (
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import functools
|
import functools
|
||||||
|
from typing import Union, List
|
||||||
|
|
||||||
import aiohttp
|
import aiohttp
|
||||||
|
|
||||||
|
@ -23,11 +24,11 @@ class GHClient:
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
*,
|
*,
|
||||||
username: str | None = None,
|
username: Union[str, None] = None,
|
||||||
token: str | None = None,
|
token: str | None = None,
|
||||||
user_cache_size: int = 30,
|
user_cache_size: int = 30,
|
||||||
repo_cache_size: int = 15,
|
repo_cache_size: int = 15,
|
||||||
custom_headers: dict[str, str | int] = {}
|
custom_headers: dict[str, Union[str, int]] = {}
|
||||||
):
|
):
|
||||||
"""The main client, used to start most use-cases."""
|
"""The main client, used to start most use-cases."""
|
||||||
self._headers = custom_headers
|
self._headers = custom_headers
|
||||||
|
@ -138,7 +139,7 @@ class GHClient:
|
||||||
"""Fetch a Github gist from it's id."""
|
"""Fetch a Github gist from it's id."""
|
||||||
return Gist(await self.http.get_gist(gist), self.http.session)
|
return Gist(await self.http.get_gist(gist), self.http.session)
|
||||||
|
|
||||||
async def create_gist(self, *, files: list[File], description: str, public: bool) -> Gist:
|
async def create_gist(self, *, files: List[File], description: str, public: bool) -> Gist:
|
||||||
"""Creates a Gist with the given files, requires authorisation."""
|
"""Creates a Gist with the given files, requires authorisation."""
|
||||||
return Gist(await self.http.create_gist(files=files, description=description, public=public), self.http.session)
|
return Gist(await self.http.create_gist(files=files, description=description, public=public), self.http.session)
|
||||||
|
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
#== objects.py ==#
|
#== objects.py ==#
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING, Any, Union, List, Dict
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from .http import http
|
from .http import http
|
||||||
|
@ -38,7 +38,7 @@ class APIObject:
|
||||||
'_http'
|
'_http'
|
||||||
)
|
)
|
||||||
|
|
||||||
def __init__(self, response: dict[str, str | int | dict[str, str | int]], _http: http) -> None:
|
def __init__(self, response: Dict[str, Any] , _http: http) -> None:
|
||||||
self._http = _http
|
self._http = _http
|
||||||
self._response = response
|
self._response = response
|
||||||
|
|
||||||
|
@ -238,7 +238,7 @@ class Issue(APIObject):
|
||||||
#=== Gist stuff ===#
|
#=== Gist stuff ===#
|
||||||
|
|
||||||
class File:
|
class File:
|
||||||
def __init__(self, fp: str | io.StringIO, filename: str = 'DefaultFilename.txt'):
|
def __init__(self, fp: Union[str, io.StringIO], filename: str = 'DefaultFilename.txt'):
|
||||||
self.fp = fp
|
self.fp = fp
|
||||||
self.filename = filename
|
self.filename = filename
|
||||||
|
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue