Django

Adapter for the Django cache framework: allows using a pre-configured Django cache for Cachetory's Cache.

Tip

Django backend allows using Cachetory with an existing configured Django cache. That may be useful for:

• migrating from Django cache to Cachetory
• using Cachetory's @cached with Django

Supported URLs

• django://<cache-name>

---

cachetory.backends.sync.DjangoBackend

Synchronous Django cache adapter.

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

class DjangoBackend(SyncBackend[WireT], Generic[WireT]):
    """Synchronous Django cache adapter."""

    __slots__ = ("_cache",)

    @classmethod
    def from_url(cls, url: str) -> DjangoBackend[WireT]:
        return DjangoBackend(caches[urlparse(url).hostname])

    def __init__(self, cache: BaseCache = cache) -> None:
        """Initialize backend with the Django cache instance."""
        self._cache = cache

    def get(self, key: str) -> WireT:
        if (value := self._cache.get(key, _SENTINEL)) is not _SENTINEL:
            return value  # type: ignore[no-any-return]
        raise KeyError(key)

    def get_many(self, *keys: str) -> Iterable[tuple[str, WireT]]:
        return self._cache.get_many(keys).items()  # type: ignore[no-any-return]

    def set(  # noqa: A003
        self,
        key: str,
        value: WireT,
        *,
        time_to_live: timedelta | None = None,
        if_not_exists: bool = False,
    ) -> bool:
        timeout = self._to_timeout(time_to_live)
        if if_not_exists:
            self._cache.get_or_set(key, value, timeout)
        else:
            self._cache.set(key, value, timeout)
        return True

    def set_many(self, items: Iterable[tuple[str, WireT]]) -> None:
        self._cache.set_many(dict(items))

    def delete(self, key: str) -> bool:
        return self._cache.delete(key)  # type: ignore[no-any-return]

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

    def expire_in(self, key: str, time_to_live: timedelta | None = None) -> None:
        self._cache.touch(key, self._to_timeout(time_to_live))

    def expire_at(self, key: str, deadline: datetime | None) -> None:
        self.expire_in(key, make_time_to_live(deadline))

    @staticmethod
    def _to_timeout(time_to_live: timedelta | None) -> object:
        return time_to_live.total_seconds() if time_to_live is not None else DEFAULT_TIMEOUT

__init__

__init__(cache: BaseCache = cache) -> None

Initialize backend with the Django cache instance.

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

def __init__(self, cache: BaseCache = cache) -> None:
    """Initialize backend with the Django cache instance."""
    self._cache = cache

---

cachetory.backends.async_.DjangoBackend

Asynchronous Django cache adapter.

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

class DjangoBackend(AsyncBackend[WireT], Generic[WireT]):
    """Asynchronous Django cache adapter."""

    __slots__ = ("_cache",)

    @classmethod
    def from_url(cls, url: str) -> DjangoBackend[WireT]:
        return DjangoBackend(caches[urlparse(url).hostname])

    def __init__(self, cache: BaseCache = cache) -> None:
        """Initialize backend with the Django cache instance."""
        self._cache = cache

    async def get(self, key: str) -> WireT:
        if (value := await self._cache.aget(key, _SENTINEL)) is not _SENTINEL:
            return value  # type: ignore[no-any-return]
        raise KeyError(key)

    async def get_many(self, *keys: str) -> AsyncIterable[tuple[str, WireT]]:
        for item in (await self._cache.aget_many(keys)).items():
            yield item

    async def set(  # noqa: A003
        self,
        key: str,
        value: WireT,
        *,
        time_to_live: timedelta | None = None,
        if_not_exists: bool = False,
    ) -> bool:
        timeout = self._to_timeout(time_to_live)
        if if_not_exists:
            await self._cache.aget_or_set(key, value, timeout)
        else:
            await self._cache.aset(key, value, timeout)
        return True

    async def set_many(self, items: Iterable[tuple[str, WireT]]) -> None:
        await self._cache.aset_many(dict(items))

    async def delete(self, key: str) -> bool:
        return await self._cache.adelete(key)  # type: ignore[no-any-return]

    async def clear(self) -> None:
        await self._cache.aclear()

    async def expire_in(self, key: str, time_to_live: timedelta | None = None) -> None:
        await self._cache.atouch(key, self._to_timeout(time_to_live))

    async def expire_at(self, key: str, deadline: datetime | None) -> None:
        await self.expire_in(key, make_time_to_live(deadline))

    @staticmethod
    def _to_timeout(time_to_live: timedelta | None) -> object:
        return time_to_live.total_seconds() if time_to_live is not None else DEFAULT_TIMEOUT

__init__

__init__(cache: BaseCache = cache) -> None

Initialize backend with the Django cache instance.

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

def __init__(self, cache: BaseCache = cache) -> None:
    """Initialize backend with the Django cache instance."""
    self._cache = cache