From 8b91010b122fa910d7fd9f3bff7719a5efb74e88 Mon Sep 17 00:00:00 2001 From: sudosnok Date: Thu, 7 Apr 2022 22:56:01 +0100 Subject: [PATCH] Massive shift to avoid circular imports, functionally the same, though objects are forced to be in one file to avoid the circular nature --- Github/cache.py | 21 +-- Github/http.py | 39 ++++-- Github/main.py | 12 +- Github/objects.py | 276 +++++++++++++++++++++++++++++++++++++ Github/objects/__init__.py | 7 - Github/objects/gists.py | 41 ------ Github/objects/objects.py | 29 ---- Github/objects/org.py | 52 ------- Github/objects/repo.py | 106 -------------- Github/objects/user.py | 78 ----------- 10 files changed, 326 insertions(+), 335 deletions(-) create mode 100644 Github/objects.py delete mode 100644 Github/objects/__init__.py delete mode 100644 Github/objects/gists.py delete mode 100644 Github/objects/objects.py delete mode 100644 Github/objects/org.py delete mode 100644 Github/objects/repo.py delete mode 100644 Github/objects/user.py diff --git a/Github/cache.py b/Github/cache.py index 3f976a3..de99050 100644 --- a/Github/cache.py +++ b/Github/cache.py @@ -1,5 +1,7 @@ #== cache.py ==# +from __future__ import annotations + __all__ = ( 'UserCache', 'RepoCache', @@ -7,7 +9,8 @@ __all__ = ( ) from collections import deque -from .objects import APIOBJECT, User, Repository, Organization + +from .objects import APIObject, User, Repository, Organization class _BaseCache(dict): @@ -19,12 +22,12 @@ class _BaseCache(dict): self._lru_keys = deque(maxlen=self._max_size) super().__init__(args) - def __getitem__(self, __k: str) -> APIOBJECT: + def __getitem__(self, __k: str) -> APIObject: target = self._lru_keys.pop(self._lru_keys.index(__k)) self._lru_keys.appendleft(target) return super().__getitem__(__k) - def __setitem__(self, __k: str, __v: APIOBJECT) -> None: + def __setitem__(self, __k: str, __v: APIObject) -> None: if len(self) == self._max_size: to_pop = self._lru_keys.pop(-1) del self[to_pop] @@ -37,12 +40,12 @@ class _BaseCache(dict): class UserCache(_BaseCache): """This adjusts the typehints to reflect User objects""" - def __getitem__(self, __k: str) -> User: + def __getitem__(self, __k: str) -> 'User': target = self._lru_keys.pop(self._lru_keys.index(__k)) self._lru_keys.appendleft(target) return super().__getitem__(__k) - def __setitem__(self, __k: str, __v: User) -> None: + def __setitem__(self, __k: str, __v: 'User') -> None: if len(self) == self._max_size: to_pop = self._lru_keys.pop(-1) del self[to_pop] @@ -55,12 +58,12 @@ class UserCache(_BaseCache): class RepoCache(_BaseCache): """This adjusts the typehints to reflect Repo objects""" - def __getitem__(self, __k: str) -> Repository: + def __getitem__(self, __k: str) -> 'Repository': target = self._lru_keys.pop(self._lru_keys.index(__k)) self._lru_keys.appendleft(target) return super().__getitem__(__k) - def __setitem__(self, __k: str, __v: Repository) -> None: + def __setitem__(self, __k: str, __v: 'Repository') -> None: if len(self) == self._max_size: to_pop = self._lru_keys.pop(-1) del self[to_pop] @@ -72,12 +75,12 @@ class RepoCache(_BaseCache): self[key] = value class OrgCache(_BaseCache): - def __getitem__(self, __k: str) -> Organization: + def __getitem__(self, __k: str) -> 'Organization': target = self._lru_keys.pop(self._lru_keys.index(__k)) self._lru_keys.appendleft(target) return super().__getitem__(__k) - def __setitem__(self, __k: str, __v: Organization) -> None: + def __setitem__(self, __k: str, __v: 'Organization') -> None: if len(self) == self._max_size: to_pop = self._lru_keys.pop(-1) del self[to_pop] diff --git a/Github/http.py b/Github/http.py index 779fe03..a883a6e 100644 --- a/Github/http.py +++ b/Github/http.py @@ -1,15 +1,17 @@ #== http.py ==# -import aiohttp + +import json +import re from collections import namedtuple from datetime import datetime from types import SimpleNamespace -import re -import json + +import aiohttp from .exceptions import * -from .exceptions import RepositoryAlreadyExists, GistNotFound -from .objects import * +from .exceptions import GistNotFound, RepositoryAlreadyExists +from .objects import APIObject, User from .urls import * __all__ = ( @@ -71,7 +73,7 @@ class Paginator: self.session = session self.response = response self.should_paginate = bool(self.response.headers.get('Link', False)) - types: dict[str, APIOBJECT] = { + types: dict[str, APIObject] = { 'user': User, } self.target_type = types[target_type] @@ -85,11 +87,11 @@ class Paginator: """Fetches a specific page and returns the 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 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.""" if self.should_paginate: return await self.early_return() @@ -162,6 +164,27 @@ class http: return await result.json() raise UserNotFound + async def get_user_repos(self, _user: User) -> list[GithubRepoData]: + result = await self.session.get(USER_REPOS_URL.format(_user.login)) + if 200 <= result.status <= 299: + return await result.json() + else: + print('This shouldn\'t be reachable') + + async def get_user_gists(self, _user: User) -> list[GithubGistData]: + result = await self.session.get(USER_GISTS_URL.format(_user.login)) + if 200 <= result.status <= 299: + return await result.json() + else: + print('This shouldn\'t be reachable') + + async def get_user_orgs(self, _user: User) -> list[GithubOrgData]: + result = await self.session.get(USER_ORGS_URL.format(_user.login)) + if 200 <= result.status <= 299: + return await result.json() + else: + print('This shouldn\'t be reachable') + async def get_repo(self, owner: str, repo_name: str) -> GithubRepoData: """Returns a Repo's raw JSON from the given owner and repo name.""" result = await self.session.get(REPO_URL.format(owner, repo_name)) diff --git a/Github/main.py b/Github/main.py index aa36df9..736d2dc 100644 --- a/Github/main.py +++ b/Github/main.py @@ -1,18 +1,20 @@ #== main.py ==# +from __future__ import annotations __all__ = ( 'GHClient', ) -import aiohttp import asyncio import functools -from getpass import getpass -from .http import http +import aiohttp + from . import exceptions -from .objects import User, PartialUser, Repository, Organization, Issue, Gist -from .cache import UserCache, RepoCache +from .cache import RepoCache, UserCache +from .http import http +from .objects import Gist, Issue, Organization, Repository, User + class GHClient: _auth = None diff --git a/Github/objects.py b/Github/objects.py new file mode 100644 index 0000000..d1a969b --- /dev/null +++ b/Github/objects.py @@ -0,0 +1,276 @@ +#== objects.py ==# +from __future__ import annotations + +from datetime import datetime + +__all__ = ( + 'APIObject', + 'dt_formatter', + 'repr_dt', + 'PartialUser', + 'User', + 'Repository', + 'Issue', + 'Gist', + 'Organization', +) + +def dt_formatter(time_str: str) -> 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__ = ( + '_response', + '_http' + ) + + def __init__(self, response: dict[str, str | int | dict[str, str | int]], _http: 'http') -> None: + self._http = _http + self._response = response + + def __repr__(self) -> str: + return f'<{self.__class__.__name__}>' + + +#=== User stuff ===# + +class _BaseUser(APIObject): + __slots__ = ( + 'login', + 'id', + ) + def __init__(self, response: dict, _http: 'http') -> None: + super().__init__(response, _http) + self._http = _http + self.login = response.get('login') + self.id = response.get('id') + + def __repr__(self) -> str: + 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) + return [Repository(data, self._http) for data in results] + + async def gists(self) -> list[Gist]: + results = await self._http.get_user_gists(self) + return [Gist(data, self._http) for data in results] + + async def orgs(self) -> list[Organization]: + results = await self._http.get_user_orgs(self) + return [Organization(data, self._http) for data in results] + + +class User(_BaseUser): + __slots__ = ( + 'login', + 'id', + 'avatar_url', + 'html_url', + 'public_repos', + 'public_gists', + 'followers', + 'following', + 'created_at', + ) + def __init__(self, response: dict, _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} + for key, value in keys.items(): + if '_at' in key and value is not None: + setattr(self, key, dt_formatter(value)) + continue + else: + setattr(self, key, value) + continue + + def __repr__(self) -> str: + return f'' + + +class PartialUser(_BaseUser): + __slots__ = ( + 'site_admin', + 'html_url', + 'avatar_url', + ) + _BaseUser.__slots__ + + def __init__(self, response: dict, _http: 'http') -> None: + super().__init__(response, _http) + self.site_admin = response.get('site_admin') + self.html_url = response.get('html_url') + self.avatar_url = response.get('avatar_url') + + def __repr__(self) -> str: + return f'' + + async def _get_user(self) -> User: + """Upgrades the PartialUser to a User object.""" + response = await self._http.get_user(self.login) + return User(response, self._http) + + +#=== Repository stuff ===# + +class Repository(APIObject): + __slots__ = ( + 'id', + 'name', + 'owner', + 'size' + 'created_at', + 'url', + 'html_url', + 'archived', + 'disabled', + 'updated_at', + 'open_issues_count', + 'default_branch', + 'clone_url', + 'stargazers_count', + 'watchers_count', + 'forks', + 'license', + ) + def __init__(self, response: dict, _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} + for key, value in keys.items(): + if key == 'owner': + setattr(self, key, PartialUser(value, self._http)) + continue + + if key == 'name': + setattr(self, key, value) + continue + + if '_at' in key and value is not None: + setattr(self, key, dt_formatter(value)) + continue + + if 'license' in key and value is None: + setattr(self, key, None) + continue + + if 'license' in key and value is not None: + setattr(self, key, value['name']) + continue + + else: + setattr(self, key, value) + continue + + def __repr__(self) -> str: + return f'' + + +class Issue(APIObject): + __slots__ = ( + 'id', + 'title', + 'user', + 'labels', + 'state', + 'created_at', + 'closed_by', + ) + + def __init__(self, response: dict, _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} + for key, value in keys.items(): + if key == 'user': + setattr(self, key, PartialUser(value, self._http)) + continue + + if key == 'labels': + setattr(self, key, [label['name'] for label in value]) + continue + + if key == 'closed_by': + setattr(self, key, User(value, self._http)) + continue + + else: + setattr(self, key, value) + continue + + def __repr__(self) -> str: + return f'' + + +#=== Gist stuff ===# + +class Gist(APIObject): + __slots__ = ( + 'id', + 'description', + 'html_url', + 'node_id', + 'files', + 'public', + 'owner', + 'created_at', + 'comments', + 'truncated', + ) + def __init__(self, response: dict, _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} + for key, value in keys.items(): + if key == 'owner': + setattr(self, key, PartialUser(value, self._http)) + continue + if key == 'created_at': + setattr(self, key, dt_formatter(value)) + continue + else: + setattr(self, key, value) + + def __repr__(self) -> str: + return f'' + + +#=== Organization stuff ===# + +class Organization(APIObject): + __slots__ = ( + 'login', + 'id', + 'html_url', + 'is_verified', + 'public_repos', + 'public_gists', + 'followers', + 'following', + 'created_at', + 'avatar_url', + ) + + def __init__(self, response: dict, _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} + for key, value in keys.items(): + if key == 'login': + setattr(self, key, value) + continue + if '_at' in key and value is not None: + setattr(self, key, dt_formatter(value)) + continue + + else: + setattr(self, key, value) + continue + + def __repr__(self): + return f'' diff --git a/Github/objects/__init__.py b/Github/objects/__init__.py deleted file mode 100644 index dfa1f19..0000000 --- a/Github/objects/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -#== __init__.py ==# - -from .objects import * -from .user import * -from .repo import * -from .org import * -from .gists import * \ No newline at end of file diff --git a/Github/objects/gists.py b/Github/objects/gists.py deleted file mode 100644 index 8b7cccb..0000000 --- a/Github/objects/gists.py +++ /dev/null @@ -1,41 +0,0 @@ -#== gists.py ==# - -import aiohttp - -from .objects import APIOBJECT, dt_formatter -from . import PartialUser, User -from .. import http - -__all__ = ( - 'Gist', - ) - -class Gist(APIOBJECT): - __slots__ = ( - 'id', - 'description', - 'html_url', - 'node_id', - 'files', - 'public', - 'owner', - 'created_at', - 'comments', - 'truncated', - ) - def __init__(self, response: dict, session: aiohttp.ClientSession) -> None: - super().__init__(response, session) - tmp = self.__slots__ + APIOBJECT.__slots__ - 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, session)) - continue - if key == 'created_at': - setattr(self, key, dt_formatter(value)) - continue - else: - setattr(self, key, value) - - def __repr__(self) -> str: - return f'' \ No newline at end of file diff --git a/Github/objects/objects.py b/Github/objects/objects.py deleted file mode 100644 index 0ede485..0000000 --- a/Github/objects/objects.py +++ /dev/null @@ -1,29 +0,0 @@ -#== objects.py ==# - -from datetime import datetime -import aiohttp - -__all__ = ( - 'APIOBJECT', -) - -def dt_formatter(time_str): - if time_str is not None: - return datetime.strptime(time_str, r"%Y-%m-%dT%H:%M:%SZ") - return None - -def repr_dt(time_str): - return time_str.strftime(r'%d-%m-%Y, %H:%M:%S') - -class APIOBJECT: - __slots__ = ( - '_response', - 'session' - ) - - def __init__(self, response: dict[str, str | int | dict[str, str | int]], session: aiohttp.ClientSession) -> None: - self._response = response - self.session = session - - def __repr__(self) -> str: - return f'<{self.__class__.__name__}>' \ No newline at end of file diff --git a/Github/objects/org.py b/Github/objects/org.py deleted file mode 100644 index 6fdea7a..0000000 --- a/Github/objects/org.py +++ /dev/null @@ -1,52 +0,0 @@ -#== org.py ==# - -import aiohttp - -from .objects import APIOBJECT, dt_formatter -from . import PartialUser -from .. import http - -__all__ = ( - 'Organization', -) - -class Organization(APIOBJECT): - __slots__ = ( - 'login', - 'id', - 'html_url', - 'is_verified', - 'public_repos', - 'public_gists', - 'followers', - 'following', - 'created_at', - 'avatar_url', - ) - - def __init__(self, response: dict, session: aiohttp.ClientSession) -> None: - super().__init__(response, session) - tmp = self.__slots__ + APIOBJECT.__slots__ - 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) - continue - if '_at' in key and value is not None: - setattr(self, key, dt_formatter(value)) - continue - - else: - setattr(self, key, value) - continue - - def __repr__(self): - return f'' - - @classmethod - async def from_name(cls, session: aiohttp.ClientSession, org: str) -> 'Organization': - """Fetch a repository from its name.""" - response = await http.get_repo_from_name(session, org) - return Organization(response, session) - - diff --git a/Github/objects/repo.py b/Github/objects/repo.py deleted file mode 100644 index 62ac6ea..0000000 --- a/Github/objects/repo.py +++ /dev/null @@ -1,106 +0,0 @@ -#== repo.py ==# - -import aiohttp - -from .objects import APIOBJECT, dt_formatter -from . import PartialUser, User -from .. import http - -__all__ = ( - 'Repository', - 'Issue' -) - -class Repository(APIOBJECT): - __slots__ = ( - 'id', - 'name', - 'owner', - 'size' - 'created_at', - 'url', - 'html_url', - 'archived', - 'disabled', - 'updated_at', - 'open_issues_count', - 'default_branch', - 'clone_url', - 'stargazers_count', - 'watchers_count', - 'forks', - 'license', - ) - def __init__(self, response: dict, session: aiohttp.ClientSession) -> None: - super().__init__(response, session) - tmp = self.__slots__ + APIOBJECT.__slots__ - 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, session)) - continue - - if key == 'name': - setattr(self, key, value) - continue - - if '_at' in key and value is not None: - setattr(self, key, dt_formatter(value)) - continue - - if 'license' in key and value is None: - setattr(self, key, None) - continue - - if 'license' in key and value is not None: - setattr(self, key, value['name']) - continue - - else: - setattr(self, key, value) - continue - - def __repr__(self) -> str: - return f'' - - @classmethod - async def from_name(cls, session: aiohttp.ClientSession,owner: str, repo_name: str) -> 'Repository': - """Fetch a repository from its name.""" - response = await http.get_repo_from_name(session, owner, repo_name) - return Repository(response, session) - - -class Issue(APIOBJECT): - __slots__ = ( - 'id', - 'title', - 'user', - 'labels', - 'state', - 'created_at', - 'closed_by', - ) - - def __init__(self, response: dict, session: aiohttp.ClientSession) -> None: - super().__init__(response, session) - tmp = self.__slots__ + APIOBJECT.__slots__ - 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, session)) - continue - - if key == 'labels': - setattr(self, key, [label['name'] for label in value]) - continue - - if key == 'closed_by': - setattr(self, key, User(value, session)) - continue - - else: - setattr(self, key, value) - continue - - def __repr__(self) -> str: - return f'' \ No newline at end of file diff --git a/Github/objects/user.py b/Github/objects/user.py deleted file mode 100644 index 2e50f74..0000000 --- a/Github/objects/user.py +++ /dev/null @@ -1,78 +0,0 @@ -#== user.py ==# - -import aiohttp - -from .objects import APIOBJECT, dt_formatter -from .. import http - -__all__ = ( - 'User', - 'PartialUser' -) - -class _BaseUser(APIOBJECT): - __slots__ = ( - 'login', - 'id', - ) - def __init__(self, response: dict, session: aiohttp.ClientSession) -> None: - super().__init__(response, session) - self.login = response.get('login') - self.id = response.get('id') - - def __repr__(self) -> str: - return f'<{self.__class__.__name__}; id = {self.id}, login = {self.login!r}>' - -class User(_BaseUser): - __slots__ = ( - 'login', - 'id', - 'avatar_url', - 'html_url', - 'public_repos', - 'public_gists', - 'followers', - 'following', - 'created_at', - ) - def __init__(self, response: dict, session: aiohttp.ClientSession) -> None: - super().__init__(response, session) - tmp = self.__slots__ + _BaseUser.__slots__ - 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)) - continue - else: - setattr(self, key, value) - continue - - def __repr__(self) -> str: - return f'' - - @classmethod - async def get_user(cls, session: aiohttp.ClientSession, username: str) -> 'User': - """Returns a User object from the username, with the mentions slots.""" - response = await http.get_user(session, username) - return User(response, session) - -class PartialUser(_BaseUser): - __slots__ = ( - 'site_admin', - 'html_url', - 'avatar_url', - ) + _BaseUser.__slots__ - - def __init__(self, response: dict, session: aiohttp.ClientSession) -> None: - super().__init__(response, session) - self.site_admin = response.get('site_admin') - self.html_url = response.get('html_url') - self.avatar_url = response.get('avatar_url') - - def __repr__(self) -> str: - return f'' - - async def _get_user(self) -> User: - """Upgrades the PartialUser to a User object.""" - response = await http.get_user(self.session, self.login) - return User(response, self.session) \ No newline at end of file