Public/fastapi-cache
1
0
Fork 0
mirror of https://github.com/long2ice/fastapi-cache.git synced 2026-03-25 04:57:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add more type hints

Browse Source
This commit is contained in:
Ivan Moiseev
2022-10-22 20:59:37 +04:00
parent 1ef80ff457
commit 4c6abcf786
9 changed files with 71 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
fastapi_cache/backends/inmemory.py
View File

@@ -20,13 +20,14 @@ class InMemoryBackend(Backend):
def _now(self) -> int:
return int(time.time())
def _get(self, key: str):
def _get(self, key: str) -> Value | None:
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[str]]:
async with self._lock:
@@ -35,17 +36,18 @@ class InMemoryBackend(Backend):
return v.ttl_ts - self._now, v.data
return 0, None
async def get(self, key: str) -> str:
async def get(self, key: str) -> Optional[str]:
async with self._lock:
v = self._get(key)
if v:
return v.data
return None
async def set(self, key: str, value: str, expire: int = None):
async def set(self, key: str, value: str, expire: Optional[int] = None) -> None:
async with self._lock:
self._store[key] = Value(value, self._now + (expire or 0))
async def clear(self, namespace: str = None, key: str = None) -> int:
async def clear(self, namespace: Optional[str] = None, key: Optional[str] = None) -> int:
count = 0
if namespace:
keys = list(self._store.keys())