diff --git a/Github/__init__.py b/Github/__init__.py index a489e31..7a8a74e 100644 --- a/Github/__init__.py +++ b/Github/__init__.py @@ -1,4 +1,4 @@ -#== __init__.py ==# +# == __init__.py ==# __title__ = 'Github-Api-Wrapper' __authors__ = 'VarMonke', 'sudosnok' @@ -10,4 +10,4 @@ from .main import * from .objects import * from .http import * from .urls import * -from .exceptions import * \ No newline at end of file +from .exceptions import * diff --git a/Github/cache.py b/Github/cache.py index 3597251..f3e88f1 100644 --- a/Github/cache.py +++ b/Github/cache.py @@ -1,4 +1,4 @@ -#== cache.py ==# +# == cache.py ==# from __future__ import annotations @@ -6,9 +6,7 @@ from collections import deque from collections.abc import MutableMapping from typing import Any, Deque, Tuple, TypeVar -__all__: Tuple[str, ...] = ( - 'ObjectCache', -) +__all__: Tuple[str, ...] = ('ObjectCache',) K = TypeVar('K') @@ -17,26 +15,26 @@ V = TypeVar('V') class _BaseCache(MutableMapping[K, V]): """This is a rough implementation of an LRU Cache using a deque and a dict.""" - + __slots__: Tuple[str, ...] = ('_max_size', '_lru_keys') def __init__(self, max_size: int, *args: Any) -> None: - self._max_size: int = max(min(max_size, 15), 0) # bounding max_size to 15 for now + self._max_size: int = max(min(max_size, 15), 0) # bounding max_size to 15 for now self._lru_keys: Deque[K] = deque[K](maxlen=self._max_size) super().__init__(*args) - + def __getitem__(self, __k: K) -> V: index = self._lru_keys.index(__k) target = self._lru_keys[index] del self._lru_keys[index] - + self._lru_keys.appendleft(target) return super().__getitem__(__k) def __setitem__(self, __k: K, __v: V) -> None: if len(self) == self._max_size: self.__delitem__(self._lru_keys.pop()) - + self._lru_keys.appendleft(__k) return super().__setitem__(__k, __v) @@ -44,12 +42,13 @@ class _BaseCache(MutableMapping[K, V]): for key, value in dict(**kwargs).items(): key: K value: V - + self.__setitem__(key, value) class ObjectCache(_BaseCache[K, V]): """This adjusts the typehints to reflect Github objects.""" + def __getitem__(self, __k: K) -> V: index = self._lru_keys.index(__k) target = self._lru_keys[index] @@ -59,7 +58,7 @@ class ObjectCache(_BaseCache[K, V]): def __setitem__(self, __k: K, __v: V) -> None: if self.__len__() == self._max_size: self.__delitem__(self._lru_keys.pop()) - + self._lru_keys.appendleft(__k) return super().__setitem__(__k, __v) @@ -67,5 +66,5 @@ class ObjectCache(_BaseCache[K, V]): for key, value in dict(**kwargs).items(): key: K value: V - - self.__setitem__(key, value) \ No newline at end of file + + self.__setitem__(key, value) diff --git a/Github/exceptions.py b/Github/exceptions.py index 2e07202..948eca7 100644 --- a/Github/exceptions.py +++ b/Github/exceptions.py @@ -1,4 +1,4 @@ -#== exceptions.py ==# +# == exceptions.py ==# import datetime from typing import TYPE_CHECKING @@ -24,108 +24,141 @@ __all__ = ( 'IssueNotFound', 'OrganizationNotFound', 'RepositoryAlreadyExists', - ) +) + class APIError(Exception): """Base level exceptions raised by errors related to any API request or call.""" + pass + class HTTPException(Exception): """Base level exceptions raised by errors related to HTTP requests.""" + pass + class ClientException(Exception): """Base level exceptions raised by errors related to the client.""" + pass + class ResourceNotFound(Exception): """Base level exceptions raised when a resource is not found.""" + pass + class ResourceAlreadyExists(Exception): """Base level exceptions raised when a resource already exists.""" + pass + class Ratelimited(APIError): """Raised when the ratelimit from Github is reached or exceeded.""" + def __init__(self, reset_time: datetime.datetime): formatted = reset_time.strftime(r"%H:%M:%S %A, %d %b") msg = "We're being ratelimited, wait until {}.\nAuthentication raises the ratelimit.".format(formatted) super().__init__(msg) + class WillExceedRatelimit(APIError): """Raised when the library predicts the call will exceed the ratelimit, will abort the call by default.""" + def __init__(self, response: ClientRequest, count: int): msg = 'Performing this action will exceed the ratelimit, aborting.\n{} remaining available calls, calls to make: {}.' msg = msg.format(response.headers['X-RateLimit-Remaining'], count) super().__init__(msg) + class NoAuthProvided(ClientException): """Raised when no authentication is provided.""" + def __init__(self): msg = 'This action required autentication. Pass username and token kwargs to your client instance.' super().__init__(msg) + class InvalidToken(ClientException): """Raised when the token provided is invalid.""" + def __init__(self): msg = 'The token provided is invalid.' super().__init__(msg) + class InvalidAuthCombination(ClientException): """Raised when the username and token are both provided.""" + def __init__(self): msg = 'The username and token cannot be used together.' super().__init__(msg) + class LoginFailure(ClientException): """Raised when the login attempt fails.""" + def __init__(self): msg = 'The login attempt failed. Provide valid credentials.' super().__init__(msg) + class NotStarted(ClientException): """Raised when the client is not started.""" + def __init__(self): msg = 'The client is not started. Run Github.GHClient() to start.' super().__init__(msg) + class AlreadyStarted(ClientException): """Raised when the client is already started.""" + def __init__(self): msg = 'The client is already started.' super().__init__(msg) + class MissingPermissions(APIError): def __init__(self): msg = 'You do not have permissions to perform this action.' super().__init__(msg) + class UserNotFound(ResourceNotFound): def __init__(self): msg = 'The requested user was not found.' super().__init__(msg) + class RepositoryNotFound(ResourceNotFound): def __init__(self): msg = 'The requested repository is either private or does not exist.' super().__init__(msg) + class IssueNotFound(ResourceNotFound): def __init__(self): msg = 'The requested issue was not found.' super().__init__(msg) + class OrganizationNotFound(ResourceNotFound): def __init__(self): msg = 'The requested organization was not found.' super().__init__(msg) + class GistNotFound(ResourceNotFound): def __init__(self): msg = 'The requested gist was not found.' super().__init__(msg) + class RepositoryAlreadyExists(ResourceAlreadyExists): def __init__(self): msg = 'The requested repository already exists.' diff --git a/Github/http.py b/Github/http.py index 6ff70ed..af41620 100644 --- a/Github/http.py +++ b/Github/http.py @@ -1,4 +1,4 @@ -#== http.py ==# +# == http.py ==# from __future__ import annotations @@ -29,60 +29,57 @@ Rates = NamedTuple('Rates', 'remaining', 'used', 'total', 'reset_when', 'last_re # aiohttp request tracking / checking bits async def on_req_start( - session: aiohttp.ClientSession, - ctx: SimpleNamespace, - params: aiohttp.TraceRequestStartParams + session: aiohttp.ClientSession, ctx: SimpleNamespace, params: aiohttp.TraceRequestStartParams ) -> None: """Before-request hook to make sure we don't overrun the ratelimit.""" - #print(repr(session), repr(ctx), repr(params)) + # print(repr(session), repr(ctx), repr(params)) pass -async def on_req_end( - session: aiohttp.ClientSession, - ctx: SimpleNamespace, - params: aiohttp.TraceRequestEndParams -) -> None: + +async def on_req_end(session: aiohttp.ClientSession, ctx: SimpleNamespace, params: aiohttp.TraceRequestEndParams) -> None: """After-request hook to adjust remaining requests on this time frame.""" headers = params.response.headers - remaining = headers['X-RateLimit-Remaining'] - used = headers['X-RateLimit-Used'] - total = headers['X-RateLimit-Limit'] - reset_when = datetime.fromtimestamp(int(headers['X-RateLimit-Reset'])) - last_req = datetime.utcnow() + remaining = headers['X-RateLimit-Remaining'] + used = headers['X-RateLimit-Used'] + total = headers['X-RateLimit-Limit'] + reset_when = datetime.fromtimestamp(int(headers['X-RateLimit-Reset'])) + last_req = datetime.utcnow() session._rates = Rates(remaining, used, total, reset_when, last_req) + trace_config = aiohttp.TraceConfig() trace_config.on_request_start.append(on_req_start) trace_config.on_request_end.append(on_req_end) APIType: TypeAlias = Union[User, Gist, Repository] + 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.""" 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__}' - session = aiohttp.ClientSession( - auth=authorization, - headers=headers, - trace_configs=[trace_config] - ) - session._rates = Rates('', '' , '', '', '') + session = aiohttp.ClientSession(auth=authorization, headers=headers, trace_configs=[trace_config]) + session._rates = Rates('', '', '', '', '') return session + # pagination class Paginator: """This class handles pagination for objects like Repos and Orgs.""" + def __init__(self, session: aiohttp.ClientSession, response: aiohttp.ClientResponse, target_type: str): self.session = session self.response = response self.should_paginate = bool(self.response.headers.get('Link', False)) - types: Dict[str, Type[APIType]] = { # note: the type checker doesnt see subclasses like that + types: Dict[str, Type[APIType]] = { # note: the type checker doesnt see subclasses like that 'user': User, - 'gist' : Gist, - 'repo' : Repository + 'gist': Gist, + 'repo': Repository, } self.target_type: Type[APIType] = types[target_type] self.pages = {} @@ -97,18 +94,18 @@ class Paginator: async def early_return(self) -> List[APIType]: # I don't rightly remember what this does differently, may have a good ol redesign later - return [self.target_type(data, self) for data in await self.response.json()] # type: ignore + return [self.target_type(data, self) for data in await self.response.json()] # type: ignore async def exhaust(self) -> List[APIType]: """Iterates through all of the pages for the relevant object and creates them.""" if self.should_paginate: return await self.early_return() - + out: List[APIType] = [] - for page in range(1, self.max_page+1): + for page in range(1, self.max_page + 1): result = await self.session.get(self.bare_link + str(page)) - out.extend([self.target_type(item, self) for item in await result.json()]) # type: ignore - + out.extend([self.target_type(item, self) for item in await result.json()]) # type: ignore + self.is_exhausted = True return out @@ -121,14 +118,18 @@ class Paginator: raise WillExceedRatelimit(response, self.max_page) self.bare_link = groups[0][0][:-1] + # GithubUserData = GithubRepoData = GithubIssueData = GithubOrgData = GithubGistData = Dict[str, Union [str, int]] # Commentnig this out for now, consider using TypeDict's instead in the future <3 + class http: def __init__(self, headers: Dict[str, Union[str, int]], auth: Union[aiohttp.BasicAuth, None]) -> None: 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.headers = headers self.auth = auth @@ -138,7 +139,7 @@ class http: async def start(self): self.session = aiohttp.ClientSession( - headers=self.headers, # type: ignore + headers=self.headers, # type: ignore auth=self.auth, trace_configs=[trace_config], ) @@ -149,23 +150,20 @@ class http: def update_headers(self, *, flush: bool = False, new_headers: Dict[str, Union[str, int]]): if flush: from multidict import CIMultiDict - self.session._default_headers = CIMultiDict(**new_headers) # type: ignore + + self.session._default_headers = CIMultiDict(**new_headers) # type: ignore else: - self.session._default_headers = {**self.session.headers, **new_headers} # type: ignore - + self.session._default_headers = {**self.session.headers, **new_headers} # type: ignore + async def update_auth(self, *, username: str, token: str): auth = aiohttp.BasicAuth(username, token) headers = self.session.headers config = self.session.trace_configs await self.session.close() - self.session = aiohttp.ClientSession( - headers=headers, - auth=auth, - trace_configs=config - ) + self.session = aiohttp.ClientSession(headers=headers, auth=auth, trace_configs=config) def data(self): - #return session headers and auth + # return session headers and auth headers = {**self.session.headers} return {'headers': headers, 'auth': self.auth} @@ -175,52 +173,52 @@ class http: await self.session.get(BASE_URL) return (datetime.utcnow() - start).total_seconds() - async def get_self(self) -> Dict[str, Union [str, int]]: + async def get_self(self) -> Dict[str, Union[str, int]]: """Returns the authenticated User's data""" result = await self.session.get(SELF_URL) if 200 <= result.status <= 299: return await result.json() raise InvalidToken - async def get_user(self, username: str) -> Dict[str, Union [str, int]]: + async def get_user(self, username: str) -> Dict[str, Union[str, int]]: """Returns a user's public data in JSON format.""" result = await self.session.get(USERS_URL.format(username)) if 200 <= result.status <= 299: return await result.json() raise UserNotFound - async def get_user_repos(self, _user: User) -> List[Dict[str, Union [str, int]]]: + async def get_user_repos(self, _user: User) -> List[Dict[str, Union[str, int]]]: result = await self.session.get(USER_REPOS_URL.format(_user.login)) if 200 <= result.status <= 299: return await result.json() - + print('This shouldn\'t be reachable') return [] - async def get_user_gists(self, _user: User) -> List[Dict[str, Union [str, int]]]: + async def get_user_gists(self, _user: User) -> List[Dict[str, Union[str, int]]]: result = await self.session.get(USER_GISTS_URL.format(_user.login)) if 200 <= result.status <= 299: return await result.json() - + print('This shouldn\'t be reachable') return [] - async def get_user_orgs(self, _user: User) -> List[Dict[str, Union [str, int]]]: + async def get_user_orgs(self, _user: User) -> List[Dict[str, Union[str, int]]]: result = await self.session.get(USER_ORGS_URL.format(_user.login)) if 200 <= result.status <= 299: return await result.json() - + print('This shouldn\'t be reachable') return [] - async def get_repo(self, owner: str, repo_name: str) -> Dict[str, Union [str, int]]: + async def get_repo(self, owner: str, repo_name: str) -> Dict[str, Union[str, int]]: """Returns a Repo's raw JSON from the given owner and repo name.""" result = await self.session.get(REPO_URL.format(owner, repo_name)) if 200 <= result.status <= 299: return await result.json() raise RepositoryNotFound - - async def get_repo_issue(self, owner: str, repo_name: str, issue_number: int) -> Dict[str, Union [str, int]]: + + async def get_repo_issue(self, owner: str, repo_name: str, issue_number: int) -> Dict[str, Union[str, int]]: """Returns a single issue's JSON from the given owner and repo name.""" result = await self.session.get(REPO_ISSUE_URL.format(owner, repo_name, issue_number)) if 200 <= result.status <= 299: @@ -245,14 +243,14 @@ class http: raise MissingPermissions raise GistNotFound - async def get_org(self, org_name: str) -> Dict[str, Union [str, int]]: + async def get_org(self, org_name: str) -> Dict[str, Union[str, int]]: """Returns an org's public data in JSON format.""" result = await self.session.get(ORG_URL.format(org_name)) if 200 <= result.status <= 299: return await result.json() raise OrganizationNotFound - async def get_gist(self, gist_id: int) -> Dict[str, Union [str, int]]: + async def get_gist(self, gist_id: int) -> Dict[str, Union[str, int]]: """Returns a gist's raw JSON from the given gist id.""" result = await self.session.get(GIST_URL.format(gist_id)) if 200 <= result.status <= 299: @@ -260,29 +258,26 @@ class http: raise GistNotFound async def create_gist( - self, - *, - files: List['File'] = [], - description: str = 'Default description', - public: bool = False - ) -> Dict[str, Union [str, int]]: + self, *, files: List['File'] = [], description: str = 'Default description', public: bool = False + ) -> Dict[str, Union[str, int]]: data = {} data['description'] = description data['public'] = public data['files'] = {} for file in files: - data['files'][file.filename] = { - 'filename' : file.filename, # helps editing the file - 'content': file.read() - } + data['files'][file.filename] = {'filename': file.filename, 'content': file.read()} # helps editing the file data = json.dumps(data) _headers = dict(self.session.headers) - result = await self.session.post(CREATE_GIST_URL, data=data, headers=_headers|{'Accept': 'application/vnd.github.v3+json'}) + result = await self.session.post( + CREATE_GIST_URL, data=data, headers=_headers | {'Accept': 'application/vnd.github.v3+json'} + ) if 201 == result.status: return await result.json() raise InvalidToken - async def create_repo(self, name: str, description: str, public: bool, gitignore: Optional[str], license: Optional[str]) -> Dict[str, Union [str, int]]: + async def create_repo( + self, name: str, description: str, public: bool, gitignore: Optional[str], license: Optional[str] + ) -> Dict[str, Union[str, int]]: """Creates a repo for you with given data""" data = { 'name': name, diff --git a/Github/main.py b/Github/main.py index 594f14b..8c5c706 100644 --- a/Github/main.py +++ b/Github/main.py @@ -58,11 +58,11 @@ class GHClient: self.username = username self.token = token self._auth = aiohttp.BasicAuth(username, token) - + # Cache manegent - self._cache(type='user')(self.get_self) # type: ignore - self._cache(type='user')(self.get_user) # type: ignore - self._cache(type='repo')(self.get_repo) # type: ignore + self._cache(type='user')(self.get_self) # type: ignore + self._cache(type='user')(self.get_user) # type: ignore + self._cache(type='repo')(self.get_repo) # type: ignore def __call__(self, *args: Any, **kwargs: Any) -> Coroutine[Any, Any, Self]: return self.start(*args, **kwargs) @@ -74,9 +74,7 @@ class GHClient: return f'<{self.__class__.__name__} has_auth={bool(self._auth)}>' def __del__(self): - asyncio.create_task( - self.http.session.close(), name='cleanup-session-github-api-wrapper' - ) + asyncio.create_task(self.http.session.close(), name='cleanup-session-github-api-wrapper') @overload def check_limits(self, as_dict: Literal[True] = True) -> Dict[str, Union[str, int]]: @@ -86,9 +84,7 @@ class GHClient: def check_limits(self, as_dict: Literal[False] = False) -> List[str]: ... - def check_limits( - self, as_dict: bool = False - ) -> Union[Dict[str, Union[str, int]], List[str]]: + def check_limits(self, as_dict: bool = False) -> Union[Dict[str, Union[str, int]], List[str]]: if not self.has_started: raise exceptions.NotStarted if not as_dict: @@ -132,13 +128,9 @@ class GHClient: ]: def wrapper( func: Callable[Concatenate[Self, P], Awaitable[T]] - ) -> Callable[ - Concatenate[Self, P], Awaitable[Optional[Union[T, User, Repository]]] - ]: + ) -> Callable[Concatenate[Self, P], Awaitable[Optional[Union[T, User, Repository]]]]: @functools.wraps(func) - async def wrapped( - self: Self, *args: P.args, **kwargs: P.kwargs - ) -> Optional[Union[T, User, Repository]]: + async def wrapped(self: Self, *args: P.args, **kwargs: P.kwargs) -> Optional[Union[T, User, Repository]]: if type == 'user': if obj := self._user_cache.get(kwargs.get('user')): return obj @@ -176,9 +168,7 @@ class GHClient: async def get_issue(self, *, owner: str, repo: str, issue: int) -> Issue: """Fetch a Github Issue from it's name.""" - return Issue( - await self.http.get_repo_issue(owner, repo, issue), self.http - ) + return Issue(await self.http.get_repo_issue(owner, repo, issue), self.http) async def create_repo( self, @@ -202,14 +192,10 @@ class GHClient: """Fetch a Github gist from it's id.""" return Gist(await self.http.get_gist(gist), self.http) - 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.""" return Gist( - await self.http.create_gist( - files=files, description=description, public=public - ), + await self.http.create_gist(files=files, description=description, public=public), self.http, ) diff --git a/Github/objects.py b/Github/objects.py index b1ce5fd..8eed407 100644 --- a/Github/objects.py +++ b/Github/objects.py @@ -1,4 +1,4 @@ -#== objects.py ==# +# == objects.py ==# from __future__ import annotations from typing import TYPE_CHECKING, Any, Optional, Tuple, Union, Dict @@ -23,23 +23,22 @@ __all__ = ( 'Organization', ) + def dt_formatter(time_str: str) -> Optional[datetime]: if time_str is not None: return datetime.strptime(time_str, r"%Y-%m-%dT%H:%M:%SZ") - + return None + def repr_dt(_datetime: datetime) -> str: return _datetime.strftime(r'%d-%m-%Y, %H:%M:%S') class APIObject: - __slots__: Tuple[str, ...] = ( - '_response', - '_http' - ) + __slots__: Tuple[str, ...] = ('_response', '_http') - def __init__(self, response: Dict[str, Any] , _http: http) -> None: + def __init__(self, response: Dict[str, Any], _http: http) -> None: self._http = _http self._response = response @@ -47,13 +46,15 @@ class APIObject: return f'<{self.__class__.__name__}>' -#=== User stuff ===# +# === User stuff ===# + class _BaseUser(APIObject): __slots__ = ( 'login', 'id', - ) + ) + def __init__(self, response: Dict[str, Any], _http: http) -> None: super().__init__(response, _http) self._http = _http @@ -64,19 +65,19 @@ class _BaseUser(APIObject): return f'<{self.__class__.__name__} id = {self.id}, login = {self.login!r}>' async def repos(self) -> list[Repository]: - results = await self._http.get_user_repos(self) # type: ignore + results = await self._http.get_user_repos(self) # type: ignore return [Repository(data, self._http) for data in results] async def gists(self) -> list[Gist]: - results = await self._http.get_user_gists(self) # type: ignore + results = await self._http.get_user_gists(self) # type: ignore return [Gist(data, self._http) for data in results] async def orgs(self) -> list[Organization]: - results = await self._http.get_user_orgs(self) # type: ignore + results = await self._http.get_user_orgs(self) # type: ignore return [Organization(data, self._http) for data in results] -class User(_BaseUser): +class User(_BaseUser): __slots__ = ( 'login', 'id', @@ -87,11 +88,12 @@ class User(_BaseUser): 'followers', 'following', 'created_at', - ) + ) + def __init__(self, response: Dict[str, Any], _http: http) -> None: super().__init__(response, _http) tmp = self.__slots__ + _BaseUser.__slots__ - keys = {key: value for key,value in self._response.items() if key in tmp} + keys = {key: value for key, value in self._response.items() if key in tmp} for key, value in keys.items(): if '_at' in key and value is not None: setattr(self, key, dt_formatter(value)) @@ -109,7 +111,7 @@ class PartialUser(_BaseUser): 'site_admin', 'html_url', 'avatar_url', - ) + _BaseUser.__slots__ + ) + _BaseUser.__slots__ def __init__(self, response: Dict[str, Any], _http: http) -> None: super().__init__(response, _http) @@ -121,25 +123,25 @@ class PartialUser(_BaseUser): return f'<{self.__class__.__name__} login: {self.login!r}, id: {self.id}, site_admin: {self.site_admin}, html_url: {self.html_url}>' async def _get_user(self) -> User: - """Upgrades the PartialUser to a User object.""" + """Upgrades the PartialUser to a User object.""" response = await self._http.get_user(self.login) return User(response, self._http) -#=== Repository stuff ===# +# === Repository stuff ===# + class Repository(APIObject): if TYPE_CHECKING: id: int name: str owner: str - + __slots__ = ( 'id', 'name', 'owner', - 'size' - 'created_at', + 'size' 'created_at', 'url', 'html_url', 'archived', @@ -151,11 +153,12 @@ class Repository(APIObject): 'stargazers_count', 'watchers_count', 'license', - ) + ) + def __init__(self, response: Dict[str, Any], _http: http) -> None: super().__init__(response, _http) tmp = self.__slots__ + APIObject.__slots__ - keys = {key: value for key,value in self._response.items() if key in tmp} + keys = {key: value for key, value in self._response.items() if key in tmp} for key, value in keys.items(): if key == 'owner': setattr(self, key, PartialUser(value, self._http)) @@ -198,6 +201,7 @@ class Repository(APIObject): def forks(self) -> int: return self._response.get('forks') + class Issue(APIObject): __slots__ = ( 'id', @@ -212,7 +216,7 @@ class Issue(APIObject): def __init__(self, response: Dict[str, Any], _http: http) -> None: super().__init__(response, _http) tmp = self.__slots__ + APIObject.__slots__ - keys = {key: value for key,value in self._response.items() if key in tmp} + keys = {key: value for key, value in self._response.items() if key in tmp} for key, value in keys.items(): if key == 'user': setattr(self, key, PartialUser(value, self._http)) @@ -241,7 +245,9 @@ class Issue(APIObject): def html_url(self) -> str: return self._response.get('html_url') -#=== Gist stuff ===# + +# === Gist stuff ===# + class File: def __init__(self, fp: Union[str, io.StringIO], filename: str = 'DefaultFilename.txt') -> None: @@ -253,17 +259,18 @@ class File: if os.path.exists(self.fp): with open(self.fp) as fp: data = fp.read() - + return data - + return self.fp elif isinstance(self.fp, io.BytesIO): return self.fp.read() - elif isinstance(self.fp, io.StringIO): # type: ignore + elif isinstance(self.fp, io.StringIO): # type: ignore return self.fp.getvalue() - + raise TypeError(f'Expected str, io.StringIO, or io.BytesIO, got {type(self.fp)}') + class Gist(APIObject): __slots__ = ( 'id', @@ -274,11 +281,12 @@ class Gist(APIObject): 'owner', 'created_at', 'truncated', - ) + ) + def __init__(self, response: Dict[str, Any], _http: http) -> None: super().__init__(response, _http) tmp = self.__slots__ + APIObject.__slots__ - keys = {key: value for key,value in self._response.items() if key in tmp} + keys = {key: value for key, value in self._response.items() if key in tmp} for key, value in keys.items(): if key == 'owner': setattr(self, key, PartialUser(value, self._http)) @@ -309,25 +317,26 @@ class Gist(APIObject): return self._response -#=== Organization stuff ===# +# === Organization stuff ===# + class Organization(APIObject): __slots__ = ( - 'login', - 'id', - 'is_verified', - 'public_repos', - 'public_gists', - 'followers', - 'following', - 'created_at', - 'avatar_url', + 'login', + 'id', + 'is_verified', + 'public_repos', + 'public_gists', + 'followers', + 'following', + 'created_at', + 'avatar_url', ) def __init__(self, response: Dict[str, Any], _http: http) -> None: super().__init__(response, _http) tmp = self.__slots__ + APIObject.__slots__ - keys = {key: value for key,value in self._response.items() if key in tmp} + keys = {key: value for key, value in self._response.items() if key in tmp} for key, value in keys.items(): if key == 'login': setattr(self, key, value) @@ -349,4 +358,4 @@ class Organization(APIObject): @property def html_url(self): - return self._response.get('html_url') \ No newline at end of file + return self._response.get('html_url') diff --git a/Github/urls.py b/Github/urls.py index ae6c4b4..6c18259 100644 --- a/Github/urls.py +++ b/Github/urls.py @@ -1,9 +1,9 @@ -#== urls.py ==# +# == urls.py ==# BASE_URL = 'https://api.github.com' -#== user urls ==# +# == user urls ==# USERS_URL = BASE_URL + '/users/{0}' USER_HTML_URL = 'https://github.com/users/{0}' @@ -21,19 +21,19 @@ USER_FOLLOWERS_URL = USERS_URL + '/followers' USER_FOLLOWING_URL = USERS_URL + '/following' -#== repo urls ==# -CREATE_REPO_URL = BASE_URL + '/user/repos' #_auth repo create +# == repo urls ==# +CREATE_REPO_URL = BASE_URL + '/user/repos' # _auth repo create -REPOS_URL = BASE_URL + '/repos/{0}' # repos of a user +REPOS_URL = BASE_URL + '/repos/{0}' # repos of a user -REPO_URL = BASE_URL + '/repos/{0}/{1}' # a specific repo +REPO_URL = BASE_URL + '/repos/{0}/{1}' # a specific repo -REPO_ISSUE_URL = REPO_URL + '/issues/{2}' # a specific issue +REPO_ISSUE_URL = REPO_URL + '/issues/{2}' # a specific issue -#== gist urls ==# -GIST_URL = BASE_URL + '/gists/{0}' # specific gist +# == gist urls ==# +GIST_URL = BASE_URL + '/gists/{0}' # specific gist -CREATE_GIST_URL = BASE_URL + '/gists' # create a gist +CREATE_GIST_URL = BASE_URL + '/gists' # create a gist -#== org urls ==# -ORG_URL = BASE_URL + '/orgs/{0}' \ No newline at end of file +# == org urls ==# +ORG_URL = BASE_URL + '/orgs/{0}' diff --git a/setup.py b/setup.py index 0cf5679..8cc0fff 100644 --- a/setup.py +++ b/setup.py @@ -26,4 +26,4 @@ setup( long_description=readme, install_requires=requirements, python_requires='>=3.8.0', -) \ No newline at end of file +)