Memory

Simple memory backend that stores values in a plain dictionary.

Supported URLs

• memory://

Caveats

• This backend does not copy values. Meaning that mutating a stored value mutates it in the backend too. If this is not desirable, consider using another serializer or making up your own serializer which copies values in its serialize method.
• Expired items actually get deleted only when accessed. If you put a value into the backend and never try to retrieve it – it'll stay in memory forever.

---

cachetory.backends.sync.MemoryBackend

Memory backend that stores everything in a local dictionary.

Source code in cachetory/backends/sync/memory.py

class MemoryBackend(SyncBackend[WireT], Generic[WireT]):
    """Memory backend that stores everything in a local dictionary."""

    __slots__ = ("_entries",)

    @classmethod
    def from_url(cls, _url: str) -> MemoryBackend[WireT]:
        return MemoryBackend()

    def __init__(self) -> None:
        self._entries: dict[str, _Entry[WireT]] = {}

    def get(self, key: str) -> WireT:
        return self._get_entry(key).value

    def expire_at(self, key: str, deadline: datetime | None) -> None:
        try:
            entry = self._get_entry(key)
        except KeyError:
            pass
        else:
            entry.deadline = deadline

    def set(  # noqa: A003
        self,
        key: str,
        value: WireT,
        *,
        time_to_live: timedelta | None = None,
        if_not_exists: bool = False,
    ) -> bool:
        entry = _Entry[WireT](value, make_deadline(time_to_live))
        if if_not_exists:
            return self._entries.setdefault(key, entry) is entry
        else:
            self._entries[key] = _Entry[WireT](value, make_deadline(time_to_live))
            return True

    def delete(self, key: str) -> bool:
        return self._entries.pop(key, _SENTINEL) is not _SENTINEL

    def clear(self) -> None:
        self._entries.clear()

    def _get_entry(self, key: str) -> _Entry[WireT]:
        entry = self._entries[key]
        if entry.deadline is not None and entry.deadline <= datetime.now(timezone.utc):
            self._entries.pop(key, None)  # might get popped by another thread
            raise KeyError(f"`{key}` has expired")
        return entry

    @property
    def size(self) -> int:
        return len(self._entries)

---

cachetory.backends.async_.MemoryBackend

Memory backend that stores everything in a local dictionary.

Source code in cachetory/backends/async_/memory.py

class MemoryBackend(AsyncBackend[WireT], Generic[WireT]):
    """Memory backend that stores everything in a local dictionary."""

    __slots__ = ("_inner",)

    @classmethod
    def from_url(cls, _url: str) -> MemoryBackend[WireT]:
        return MemoryBackend()

    def __init__(self) -> None:
        # We'll simply delegate call to the wrapped backend.
        self._inner: SyncMemoryBackend[WireT] = SyncMemoryBackend()

    def get(self, key: str) -> Coroutine[Any, Any, WireT]:
        return postpone(self._inner.get, key)

    def expire_at(self, key: str, deadline: datetime | None) -> Coroutine[Any, Any, None]:
        return postpone(self._inner.expire_at, key, deadline)

    def set(  # noqa: A003
        self,
        key: str,
        value: WireT,
        *,
        time_to_live: timedelta | None = None,
        if_not_exists: bool = False,
    ) -> Coroutine[Any, Any, bool]:
        return postpone(
            self._inner.set,
            key,
            value,
            time_to_live=time_to_live,
            if_not_exists=if_not_exists,
        )

    def delete(self, key: str) -> Coroutine[Any, Any, bool]:
        return postpone(self._inner.delete, key)

    def clear(self) -> Coroutine[Any, Any, None]:
        return postpone(self._inner.clear)

    @property
    def size(self) -> int:
        return self._inner.size