diff --git a/src/eduid/webapp/idp/sso_cache.py b/src/eduid/webapp/idp/sso_cache.py index 225eca3fa..eedbafb6f 100644 --- a/src/eduid/webapp/idp/sso_cache.py +++ b/src/eduid/webapp/idp/sso_cache.py @@ -56,14 +56,7 @@ # either expressed or implied, of SUNET. import logging -import time -import warnings -from collections import deque -from collections.abc import Mapping -from threading import Lock -from typing import Any, cast -from eduid.common.decorators import deprecated from eduid.userdb.db import BaseDB from eduid.userdb.exceptions import EduIDDBError from eduid.webapp.idp.sso_session import SSOSession, SSOSessionId @@ -71,143 +64,6 @@ logger = logging.getLogger(__name__) -class NoOpLock: - """ - A No-op lock class, to avoid a lot of "if self.lock:" in code using locks. - """ - - def __init__(self) -> None: - pass - - # noinspection PyUnusedLocal - def acquire(self, _block: bool = True) -> bool: - """ - Fake acquiring a lock. - - :param _block: boolean, whether to block or not (NO-OP in this implementation) - """ - return True - - def release(self) -> None: - """ - Fake releasing a lock. - """ - - -@deprecated("This class seems unused") -class ExpiringCacheMem: - """ - Simplistic implementation of a cache that removes entrys as they become too old. - - This implementation invokes garbage collecting on every addition of data. This - is believed to be a pragmatic approach for small to medium sites. For a large - site with e.g. load balancers causing uneven traffic patterns, this might not - work that well and the use of an external cache such as memcache is recommended. - - :param name: name of cache as string, only used for debugging - :param logger: logging logger instance - :param ttl: data time to live in this cache, as seconds (integer) - :param lock: threading.Lock compatible locking instance - """ - - def __init__(self, name: str, logger: logging.Logger | None, ttl: int, lock: Lock | None = None) -> None: - self.logger = logger - self.ttl = ttl - self.name = name - self._data: dict[SSOSessionId, Any] = {} - self._ages: deque[tuple[float, SSOSessionId]] = deque() - self.lock = lock - if self.lock is None: - self.lock = cast(Lock, NoOpLock()) # intentionally lie to mypy - - if self.logger is not None: - warnings.warn("Object logger deprecated, using module_logger", DeprecationWarning, stacklevel=2) - - def add(self, key: SSOSessionId, info: object, now: int | None = None) -> None: - """ - Add entry to the cache. - - Ability to supply current time is only meant for test cases! - - :param key: Lookup key for entry - :param info: Value to be stored for 'key' - :param now: Current time - do not use unless testing! - """ - self._data[key] = info - # record when this entry shall be purged - _now = now - if _now is None: - _now = int(time.time()) - self._ages.append((_now, key)) - self._purge_expired(_now - self.ttl) - - def _purge_expired(self, timestamp: int) -> None: - """ - Purge expired records. - - :param timestamp: Purge any entrys older than this (integer) - """ - if not self.lock or not self.lock.acquire(False): - # if we don't get the lock, don't worry about it and just skip purging - return None - try: - # purge any expired records. self._ages have the _data entries listed with oldest first. - while True: - try: - (_exp_ts, _exp_key) = self._ages.popleft() - except IndexError: - break - if _exp_ts > timestamp: - # entry not expired - reinsert in queue and end purging - self._ages.appendleft((_exp_ts, _exp_key)) - break - logger.debug( - f"Purged {self.name!s} cache entry {timestamp - _exp_ts!s} seconds over limit : {_exp_key!s}" - ) - self.delete(_exp_key) - finally: - self.lock.release() - - def get(self, key: SSOSessionId) -> Mapping[str, Any] | None: - """ - Fetch data from cache based on `key'. - - :param key: hash key to use for lookup - :returns: Any data found matching `key', or None. - """ - return self._data.get(key) - - def update(self, key: SSOSessionId, info: object) -> None: - """ - Update an entry in the cache. - - :param key: Lookup key for entry - :param info: Value to be stored for 'key' - :return: None - """ - self._data[key] = info - - def delete(self, key: SSOSessionId) -> bool: - """ - Delete an item from the cache. - - :param key: hash key to delete - :return: True on success - """ - try: - del self._data[key] - return True - except KeyError: - logger.debug(f"Failed deleting key {key!r} from {self.name!s} cache (entry did not exist)") - return False - - def items(self) -> dict[SSOSessionId, Any]: - """ - Return all items from cache. - """ - return self._data - - class SSOSessionCacheError(EduIDDBError): pass