Remove deprecated ExpiringCacheMem

src/eduid/webapp/idp/sso_cache.py

@@ -56,158 +56,14 @@
 # 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
 
 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