mirror of
https://github.com/long2ice/fastapi-cache.git
synced 2026-03-25 13:07:53 +00:00
d10f4af6d6
This ensures that any syntax issues are caught early (by type checkers and tests). Backends that are missing dependencies are skipped. By importing, this exposed an issue where the redis type annotations raised an exception, which has been fixed by using forward annotations. To help avoid import dependency hell, the Backend ABC has been moved to `fastapi_cache.types`. In the process, it has been made an actual ABC.
62 lines
1.6 KiB
Python
62 lines
1.6 KiB
Python
import time
|
|
from asyncio import Lock
|
|
from dataclasses import dataclass
|
|
from typing import Dict, Optional, Tuple
|
|
|
|
from fastapi_cache.types import Backend
|
|
|
|
|
|
@dataclass
|
|
class Value:
|
|
data: bytes
|
|
ttl_ts: int
|
|
|
|
|
|
class InMemoryBackend(Backend):
|
|
_store: Dict[str, Value] = {}
|
|
_lock = Lock()
|
|
|
|
@property
|
|
def _now(self) -> int:
|
|
return int(time.time())
|
|
|
|
def _get(self, key: str) -> Optional[Value]:
|
|
v = self._store.get(key)
|
|
if v:
|
|
if v.ttl_ts < self._now:
|
|
del self._store[key]
|
|
else:
|
|
return v
|
|
return None
|
|
|
|
async def get_with_ttl(self, key: str) -> Tuple[int, Optional[bytes]]:
|
|
async with self._lock:
|
|
v = self._get(key)
|
|
if v:
|
|
return v.ttl_ts - self._now, v.data
|
|
return 0, None
|
|
|
|
async def get(self, key: str) -> Optional[bytes]:
|
|
async with self._lock:
|
|
v = self._get(key)
|
|
if v:
|
|
return v.data
|
|
return None
|
|
|
|
async def set(self, key: str, value: bytes, expire: Optional[int] = None) -> None:
|
|
async with self._lock:
|
|
self._store[key] = Value(value, self._now + (expire or 0))
|
|
|
|
async def clear(self, namespace: Optional[str] = None, key: Optional[str] = None) -> int:
|
|
count = 0
|
|
if namespace:
|
|
keys = list(self._store.keys())
|
|
for key in keys:
|
|
if key.startswith(namespace):
|
|
del self._store[key]
|
|
count += 1
|
|
elif key:
|
|
del self._store[key]
|
|
count += 1
|
|
return count
|