Implementation Engineer
67 lines
1.8 KiB
Python
67 lines
1.8 KiB
Python
"""Redis cache client and helpers."""
|
|
import json
|
|
from typing import Any
|
|
import redis.asyncio as redis
|
|
from app.config import get_settings
|
|
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
async def get_redis() -> redis.Redis:
|
|
"""Return a Redis client instance."""
|
|
return redis.from_url(settings.redis_url, decode_responses=True)
|
|
|
|
|
|
class CacheClient:
|
|
"""Wrapper around Redis for equipment caching."""
|
|
|
|
def __init__(self) -> None:
|
|
self._redis: redis.Redis | None = None
|
|
|
|
async def connect(self) -> None:
|
|
self._redis = await get_redis()
|
|
|
|
async def disconnect(self) -> None:
|
|
if self._redis:
|
|
await self._redis.close()
|
|
self._redis = None
|
|
|
|
async def get(self, key: str) -> Any | None:
|
|
if not self._redis:
|
|
await self.connect()
|
|
assert self._redis is not None
|
|
data = await self._redis.get(key)
|
|
if data:
|
|
return json.loads(data)
|
|
return None
|
|
|
|
async def set(self, key: str, value: Any, ttl: int = 3600) -> None:
|
|
if not self._redis:
|
|
await self.connect()
|
|
assert self._redis is not None
|
|
await self._redis.setex(key, ttl, json.dumps(value, default=str))
|
|
|
|
async def delete_pattern(self, pattern: str) -> int:
|
|
if not self._redis:
|
|
await self.connect()
|
|
assert self._redis is not None
|
|
keys = []
|
|
async for key in self._redis.scan_iter(match=pattern):
|
|
keys.append(key)
|
|
if keys:
|
|
return await self._redis.delete(*keys)
|
|
return 0
|
|
|
|
async def incr_rate(self, key: str, window: int = 60) -> int:
|
|
if not self._redis:
|
|
await self.connect()
|
|
assert self._redis is not None
|
|
count = await self._redis.incr(key)
|
|
if count == 1:
|
|
await self._redis.expire(key, window)
|
|
return count
|
|
|
|
|
|
cache = CacheClient()
|