diff --git a/Github/main.py b/Github/main.py index 699bc1b..260dcce 100644 --- a/Github/main.py +++ b/Github/main.py @@ -1,26 +1,45 @@ -#== main.py ==# +# == main.py ==# from __future__ import annotations -from datetime import datetime -__all__ = ( - 'GHClient', -) +__all__ = ("GHClient",) import asyncio import functools -from typing import Any, Union, List, Dict - import aiohttp +from typing import ( + TYPE_CHECKING, + Awaitable, + Callable, + Literal, + Any, + Coroutine, + Dict, + Generator, + Optional, + Union, + List, + overload, + TypeVar, +) +from typing_extensions import Self, ParamSpec, Concatenate + from . import exceptions from .cache import ObjectCache from .http import http from .objects import Gist, Issue, Organization, Repository, User, File + +T = TypeVar("T") +P = ParamSpec("P") + + class GHClient: - _auth = None - has_started = False - http: http + if TYPE_CHECKING: + http: http + + has_started: bool = False + def __init__( self, *, @@ -28,47 +47,69 @@ class GHClient: token: Union[str, None] = None, user_cache_size: int = 30, repo_cache_size: int = 15, - custom_headers: dict[str, Union[str, int]] = {} + custom_headers: dict[str, Union[str, int]] = {}, ): """The main client, used to start most use-cases.""" self._headers = custom_headers - + self._user_cache = ObjectCache[Any, User](user_cache_size) self._repo_cache = ObjectCache[Any, Repository](repo_cache_size) if username and token: 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 - def __await__(self) -> 'GHClient': + def __call__(self, *args: Any, **kwargs: Any) -> Coroutine[Any, Any, Self]: + return self.start(*args, **kwargs) + + def __await__(self) -> Generator[Any, Any, Self]: return self.start().__await__() def __repr__(self) -> str: - return f'<{self.__class__.__name__} has_auth={bool(self._auth)}>' + return f"<{self.__class__.__name__} has_auth={bool(self._auth)}>" def __del__(self): - asyncio.create_task(self.http.session.close()) + asyncio.create_task( + self.http.session.close(), name="cleanup-session-github-api-wrapper" + ) - def check_limits(self, as_dict: bool = False) -> dict[str, str | int] | list[str]: + @overload + def check_limits(self, as_dict: Literal[True] = True) -> Dict[str, Union[str, int]]: + ... + + @overload + 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]]: if not self.has_started: raise exceptions.NotStarted if not as_dict: - output = [] - for key, value in self.http.session._rates._asdict().items(): - output.append(f'{key} : {value}') + output: List[str] = [] + for key, value in self.http.session._rates._asdict().items(): # type: ignore + output.append(f"{key} : {value}") + return output - return self.http.session._rates + + return self.http.session._rates # type: ignore async def update_auth(self, username: str, token: str) -> None: """Allows you to input auth information after instantiating the client.""" - #check if username and token is valid + # check if username and token is valid await self.http.update_auth(username=username, token=token) try: await self.http.get_self() except exceptions.InvalidToken as exc: raise exceptions.InvalidToken from exc - async def start(self) -> 'GHClient': + async def start(self) -> Self: """Main entry point to the wrapper, this creates the ClientSession.""" if self.has_started: raise exceptions.AlreadyStarted @@ -83,78 +124,103 @@ class GHClient: self.has_started = True return self - def _cache(*args, **kwargs): - target_type = kwargs.get('type') - def wrapper(func): + def _cache( + self: Self, *, type: str + ) -> Callable[ + [Callable[Concatenate[Self, P], Awaitable[T]]], + Callable[Concatenate[Self, P], Awaitable[Optional[Union[T, User, Repository]]]], + ]: + def wrapper( + func: Callable[Concatenate[Self, P], Awaitable[T]] + ) -> Callable[ + Concatenate[Self, P], Awaitable[Optional[Union[T, User, Repository]]] + ]: @functools.wraps(func) - async def wrapped(self, *args, **kwargs): - if target_type == 'User': - if (obj := self._user_cache.get(kwargs.get('user'))): + 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 - else: - res = await func(self, *args, **kwargs) - self._user_cache[kwargs.get('user')] = res - return res - if target_type == 'Repo': - if (obj := self._repo_cache.get(kwargs.get('repo'))): + + user: User = await func(self, *args, **kwargs) # type: ignore + self._user_cache[kwargs.get("user")] = user + return user + if type == "repo": + if obj := self._repo_cache.get(kwargs.get("repo")): return obj - else: - res = await func(self, *args, **kwargs) - self._repo_cache[kwargs.get('repo')] = res - return res + + repo: Repository = await func(self, *args, **kwargs) # type: ignore + self._repo_cache[kwargs.get("repo")] = repo + return repo + return wrapped + return wrapper - #@_cache(type='User') + # @_cache(type='User') async def get_self(self) -> User: """Returns the authenticated User object.""" if self._auth: - return User(await self.http.get_self(), self.http.session) + return User(await self.http.get_self(), self.http) else: raise exceptions.NoAuthProvided - @_cache(type='User') async def get_user(self, *, user: str) -> User: """Fetch a Github user from their username.""" - return User(await self.http.get_user(user), self.http.session) + return User(await self.http.get_user(user), self.http) - @_cache(type='Repo') async def get_repo(self, *, owner: str, repo: str) -> Repository: """Fetch a Github repository from it's name.""" - return Repository(await self.http.get_repo(owner, repo), self.http.session) + return Repository(await self.http.get_repo(owner, repo), self.http) 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.session) + return Issue( + await self.http.get_repo_issue(owner, repo, issue), self.http + ) - async def create_repo(self, name: str, description: str = 'Repository created using Github-Api-Wrapper.', public: bool = False,gitignore: str = None, license: str = None) -> Repository: - return Repository(await self.http.create_repo(name,description,public,gitignore,license), self.http.session) + async def create_repo( + self, + name: str, + description: str = "Repository created using Github-Api-Wrapper.", + public: bool = False, + gitignore: Optional[str] = None, + license: Optional[str] = None, + ) -> Repository: + return Repository( + await self.http.create_repo(name, description, public, gitignore, license), + self.http, + ) - async def delete_repo(self, repo: str= None, owner: str = None) -> None: + async def delete_repo(self, repo: str, owner: str) -> Optional[str]: """Delete a Github repository, requires authorisation.""" owner = owner or self.username return await self.http.delete_repo(owner, repo) async def get_gist(self, gist: int) -> Gist: """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) - 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), self.http.session) + return Gist( + await self.http.create_gist( + files=files, description=description, public=public + ), + self.http, + ) - async def delete_gist(self, gist: int) -> None: + async def delete_gist(self, gist: int) -> Optional[str]: """Delete a Github gist, requires authorisation.""" return await self.http.delete_gist(gist) async def get_org(self, org: str) -> Organization: """Fetch a Github organization from it's name.""" - return Organization(await self.http.get_org(org), self.http.session) + return Organization(await self.http.get_org(org), self.http) async def latency(self) -> float: """Returns the latency of the client.""" return await self.http.latency() - - - -