1
Fork 0
mirror of https://github.com/RGBCube/GitHubWrapper synced 2025-05-25 18:25:09 +00:00

Started the cache hopefully

This commit is contained in:
sudosnok 2022-03-27 17:47:22 +01:00
parent fe977277be
commit 59001b7153
6 changed files with 253 additions and 0 deletions

75
Github/cache.py Normal file
View file

@ -0,0 +1,75 @@
#== cache.py ==#
__all__ = (
'UserCache',
'RepoCache',
'OrgCache',
)
from collections import deque
from .objects import APIOBJECT, User, Repository
class _BaseCache(dict):
"""This is a rough implementation of an LRU Cache using a deque and a dict."""
_max_size: int
_lru_keys: deque
def __init__(self, max_size: int, *args):
self._max_size = max(min(max_size, 15), 0) # bounding max_size to 15 for now
self._lru_keys = deque(maxlen=self._max_size)
super().__init__(args)
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:
if len(self) == self._max_size:
to_pop = self._lru_keys.pop(-1)
del self[to_pop]
self._lru_keys.appendleft(__k)
return super().__setitem__(__k, __v)
def update(self, *args, **kwargs) -> None:
for key, value in dict(*args, **kwargs).iteritems():
self[key] = value
class UserCache(_BaseCache):
"""This adjusts the typehints to reflect User objects"""
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:
if len(self) == self._max_size:
to_pop = self._lru_keys.pop(-1)
del self[to_pop]
self._lru_keys.appendleft(__k)
return super().__setitem__(__k, __v)
def update(self, *args, **kwargs) -> None:
for key, value in dict(*args, **kwargs).iteritems():
self[key] = value
class RepoCache(_BaseCache):
"""This adjusts the typehints to reflect Repo objects"""
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:
if len(self) == self._max_size:
to_pop = self._lru_keys.pop(-1)
del self[to_pop]
self._lru_keys.appendleft(__k)
return super().__setitem__(__k, __v)
def update(self, *args, **kwargs) -> None:
for key, value in dict(*args, **kwargs).iteritems():
self[key] = value
class OrgCache(_BaseCache):
pass

View file

@ -61,6 +61,20 @@ class GHClient:
self.has_started = True
return self
def _cache(func, cache_type: str):
async def wrapper(self: 'GHClient', name: str):
if cache_type == 'user':
if (user := self._user_cache.get(name)):
return user
else:
return await func(self, name)
if cache_type == 'repo':
if (repo := self._repo_cache.get(name)):
return repo
else:
return await func(self, name)
return wrapper
async def get_user(self, username: str) -> User:
"""Fetch a Github user from their username."""
return User(await http.get_user(self.session, username), self.session)

View file

@ -0,0 +1,5 @@
#== __init__.py ==#
from .objects import *
from .user import *
from .repo import *

View file

@ -0,0 +1,29 @@
#== 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], session: aiohttp.ClientSession) -> None:
self._response = response
self.session = session
def __repr__(self) -> str:
return f'<{self.__class__.__name__}>'

49
Github/objetcts/repo.py Normal file
View file

@ -0,0 +1,49 @@
#== repo.py ==#
import aiohttp
from .objects import APIOBJECT, dt_formatter
from . import PartialUser
from .. import http
__all__ = (
'Repository',
)
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 key.items():
if key == 'owner':
self.owner = PartialUser(value, self.session)
return
if '_at' in key and value is not None:
setattr(self, key, dt_formatter(value))
return
setattr(self, key, value)
def __repr__(self) -> str:
return f'<Repository; id: {self.id}, name: {self.name}, owner: {self.owner}, created_at: {self.created_at}, default_branch: {self.default_branch}, license: {self.license}, >'

81
Github/objetcts/user.py Normal file
View file

@ -0,0 +1,81 @@
#== 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}>'
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
setattr(self, key, value)
def __repr__(self):
return f'<User; login: {self.login}, id: {self.id}, created_at: {self.created_at}>'
@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',
'created_at',
) + _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):
return f'<PartialUser; login: {self.login}, id: {self.id}, site_admin: {self.site_admin}, html_url: {self.html_url}, created_at: {self.created_at}>'
async def _get_user(self):
"""Upgrades the PartialUser to a User object."""
response = await http.get_user(self.session, self.login)
return User(response, self.session)