68 lines
1.6 KiB
Python
68 lines
1.6 KiB
Python
import os
|
|
import json
|
|
import asyncio
|
|
from typing import Optional, Any, Callable, Awaitable
|
|
from datetime import timedelta
|
|
|
|
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
|
|
CACHE_TTL = int(os.getenv("CACHE_TTL", "300"))
|
|
|
|
_redis_pool = None
|
|
_lock = asyncio.Lock()
|
|
|
|
|
|
async def get_redis():
|
|
global _redis_pool
|
|
if _redis_pool is None:
|
|
import aioredis
|
|
async with _lock:
|
|
if _redis_pool is None:
|
|
_redis_pool = await aioredis.from_url(REDIS_URL, decode_responses=True)
|
|
return _redis_pool
|
|
|
|
|
|
async def cache_get(key: str) -> Optional[Any]:
|
|
try:
|
|
r = await get_redis()
|
|
val = await r.get(key)
|
|
if val:
|
|
return json.loads(val)
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
async def cache_set(key: str, value: Any, ttl: int = CACHE_TTL):
|
|
try:
|
|
r = await get_redis()
|
|
await r.setex(key, ttl, json.dumps(value, default=str))
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
async def cache_delete(key: str):
|
|
try:
|
|
r = await get_redis()
|
|
await r.delete(key)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
async def cache_get_or_set(key: str, factory: Callable[[], Awaitable[Any]], ttl: int = CACHE_TTL) -> Any:
|
|
cached = await cache_get(key)
|
|
if cached is not None:
|
|
return cached
|
|
value = await factory()
|
|
await cache_set(key, value, ttl)
|
|
return value
|
|
|
|
|
|
async def cache_invalidate_pattern(pattern: str):
|
|
try:
|
|
r = await get_redis()
|
|
keys = await r.keys(pattern)
|
|
if keys:
|
|
await r.delete(*keys)
|
|
except Exception:
|
|
pass
|