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

update cache to reflect objects easier

This commit is contained in:
NextChai 2022-04-30 01:54:17 -04:00
parent 8091515ee9
commit b551458c03
2 changed files with 51 additions and 72 deletions

View file

@ -2,91 +2,70 @@
from __future__ import annotations
__all__ = (
'UserCache',
'RepoCache',
'OrgCache',
from collections import deque
from collections.abc import MutableMapping
from typing import Any, Deque, Tuple, TypeVar
__all__: Tuple[str, ...] = (
'ObjectCache',
)
from collections import deque
from .objects import APIObject, User, Repository, Organization
K = TypeVar('K')
V = TypeVar('V')
class _BaseCache(dict):
class _BaseCache(MutableMapping[K, V]):
"""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))
__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._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: str, __v: APIObject) -> None:
def __setitem__(self, __k: K, __v: V) -> None:
if len(self) == self._max_size:
to_pop = self._lru_keys.pop(-1)
del self[to_pop]
self.__delitem__(self._lru_keys.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
def update(self, **kwargs: Any) -> None:
for key, value in dict(**kwargs).items():
key: K
value: V
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.__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]
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]
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)
def update(self, *args, **kwargs) -> None:
for key, value in dict(*args, **kwargs).iteritems():
self[key] = value
def update(self, **kwargs: Any) -> None:
for key, value in dict(**kwargs).items():
key: K
value: V
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):
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:
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
self.__setitem__(key, value)

View file

@ -8,12 +8,12 @@ __all__ = (
import asyncio
import functools
from typing import Union, List, Dict
from typing import Any, Union, List, Dict
import aiohttp
from . import exceptions
from .cache import RepoCache, UserCache
from .cache import ObjectCache
from .http import http
from .objects import Gist, Issue, Organization, Repository, User, File
@ -32,9 +32,9 @@ class GHClient:
):
"""The main client, used to start most use-cases."""
self._headers = custom_headers
bound = lambda hi, lo, value: max(min(value, hi), lo)
self._user_cache = UserCache(bound(50, 0, user_cache_size))
self._repo_cache = RepoCache(bound(50, 0, repo_cache_size))
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