Source code for blissoda.persistent.mock_redis

import logging
from collections import OrderedDict
from collections.abc import MutableMapping
from copy import deepcopy
from typing import Any
from typing import Dict
from typing import Mapping

logger = logging.getLogger(__name__)

# Global in-memory Redis-like store
_REDIS_STORE: Dict[str, dict] = {}




class Connection:
    """Minimal stand-in for redis connection."""



    def exists(self, key: str) -> bool:
        return key in _REDIS_STORE






class LocalHashObjSetting(MutableMapping):
    """In-memory drop-in replacement for HashObjSetting."""

    connection = Connection()

    _LOCAL_REASON = None
    _LOCAL_WARNING_EMITTED = False

    def __init__(self, key: str):
        self.key = key

        if (
            not LocalHashObjSetting._LOCAL_WARNING_EMITTED
            and LocalHashObjSetting._LOCAL_REASON
        ):
            logger.warning(
                "Persistency is disabled and local in-memory storage is being used (%s)",
                LocalHashObjSetting._LOCAL_REASON,
            )
            LocalHashObjSetting._LOCAL_WARNING_EMITTED = True

    @property
    def _store(self) -> dict:
        return _REDIS_STORE.setdefault(self.key, {})

    def __repr__(self):
        return repr(self._store)

    def __getitem__(self, k: str) -> Any:
        return self._store[k]

    def __setitem__(self, k: str, value: Any) -> None:
        if value is None:
            self._store.pop(k, None)
        else:
            self._store[k] = deepcopy(value) if isinstance(value, Mapping) else value

    def __delitem__(self, k: str) -> None:
        del self._store[k]

    def __iter__(self):
        return iter(self._store)

    def __len__(self):
        return len(self._store)



    def get_all(self) -> dict:
        return deepcopy(self._store)




    def clear(self) -> None:
        _REDIS_STORE.pop(self.key, None)






class LocalOrderedHashObjSetting(LocalHashObjSetting):
    """
    Same as LocalHashObjSetting, but stores values in an OrderedDict so that
    insertion order is preserved.

    Behaves like a Redis hash that returns fields in insertion order.
    """

    @property
    def _store(self) -> OrderedDict:
        # Upgrade to OrderedDict if necessary
        store = _REDIS_STORE.get(self.key)

        if store is None:
            store = OrderedDict()
            _REDIS_STORE[self.key] = store
        else:
            if not isinstance(store, OrderedDict):
                _REDIS_STORE[self.key] = OrderedDict(store)
                store = _REDIS_STORE[self.key]

        return store