diff --git a/Github/cache.py b/Github/cache.py index de99050..3597251 100644 --- a/Github/cache.py +++ b/Github/cache.py @@ -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 + + self.__setitem__(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)) + +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 - -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 + def update(self, **kwargs: Any) -> None: + for key, value in dict(**kwargs).items(): + key: K + value: V + + self.__setitem__(key, value) \ No newline at end of file diff --git a/Github/main.py b/Github/main.py index 0f307e4..699bc1b 100644 --- a/Github/main.py +++ b/Github/main.py @@ -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