"""
Provide the Cache API.
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import Self, Generic, TypeAlias, AsyncContextManager, overload, Literal
from typing_extensions import TypeVar
_CacheItemValueT = TypeVar("_CacheItemValueT")
_CacheItemValueCoT = TypeVar("_CacheItemValueCoT", covariant=True)
_CacheItemValueContraT = TypeVar("_CacheItemValueContraT", contravariant=True)
[docs]
class CacheItem(Generic[_CacheItemValueCoT]):
"""
A cache item.
"""
@property
def modified(self) -> int | float:
"""
Get the time this cache item was last modified, in seconds.
"""
raise NotImplementedError
[docs]
async def value(self) -> _CacheItemValueCoT:
"""
Get this cache item's value.
"""
raise NotImplementedError
[docs]
class Cache(Generic[_CacheItemValueContraT]):
"""
Provide a cache.
Implementations MUST be thread-safe.
"""
[docs]
def with_scope(self, scope: str) -> Self:
"""
Return a new nested cache with the given scope.
"""
raise NotImplementedError
[docs]
def get(
self, cache_item_id: str
) -> AsyncContextManager[CacheItem[_CacheItemValueContraT] | None]:
"""
Get the cache item with the given ID.
"""
raise NotImplementedError
[docs]
def getset(
self, cache_item_id: str, *, wait: bool = True
) -> AsyncContextManager[
tuple[
CacheItem[_CacheItemValueContraT] | None,
CacheItemValueSetter[_CacheItemValueContraT] | None,
]
]:
"""
Get the cache item with the given ID, and provide a setter to add or update it within the same atomic operation.
If ``wait`` is ``False`` and no lock can be acquired, return ``None, None``.
Otherwise return:
0. A cache item if one could be found, or else ``None``.
1. An asynchronous setter that takes the cache item's value as its only argument.
"""
raise NotImplementedError
[docs]
async def delete(self, cache_item_id: str) -> None:
"""
Delete the cache item with the given ID.
"""
raise NotImplementedError
[docs]
async def clear(self) -> None:
"""
Clear all items from the cache.
"""
raise NotImplementedError