Spaces:
Sleeping
Sleeping
File size: 2,655 Bytes
24fe247 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | import hashlib
import json
from typing import Any, Optional
from src.core.config import REDIS_URL, REDIS_TTL_INFERENCE, REDIS_TTL_SEARCH
_redis_client = None
async def get_redis():
global _redis_client
if _redis_client is None and REDIS_URL:
try:
import redis.asyncio as aioredis
_redis_client = aioredis.from_url(
REDIS_URL,
encoding="utf-8",
decode_responses=False,
socket_connect_timeout=2,
socket_timeout=2,
retry_on_timeout=True,
)
await _redis_client.ping()
except Exception:
_redis_client = None
return _redis_client
async def cache_get(key: str) -> Optional[Any]:
r = await get_redis()
if not r:
return None
try:
data = await r.get(key)
return json.loads(data) if data else None
except Exception:
return None
async def cache_set(key: str, value: Any, ttl: int = 3600) -> bool:
r = await get_redis()
if not r:
return False
try:
await r.setex(key, ttl, json.dumps(value, default=_numpy_serializer))
return True
except Exception:
return False
async def cache_delete_pattern(pattern: str) -> int:
r = await get_redis()
if not r:
return 0
try:
keys = await r.keys(pattern)
if keys:
return await r.delete(*keys)
return 0
except Exception:
return 0
async def cache_flush_all() -> bool:
r = await get_redis()
if not r:
return False
try:
await r.flushdb()
return True
except Exception:
return False
def make_inference_key(image_bytes: bytes, detect_faces: bool, mode: str) -> str:
h = hashlib.md5(image_bytes[:65536]).hexdigest()
return f"inf:{h}:{int(detect_faces)}:{mode}"
def make_face_search_key(vec: list, threshold: float) -> str:
h = hashlib.md5(str(vec[:8]).encode()).hexdigest()[:16]
return f"face:{h}:{threshold:.2f}"
def make_obj_search_key(vec: list) -> str:
h = hashlib.md5(str(vec[:8]).encode()).hexdigest()[:16]
return f"obj:{h}"
def make_categories_key(cloud_name: str) -> str:
return f"cats:{cloud_name}"
def _numpy_serializer(obj):
try:
import numpy as np
if isinstance(obj, np.ndarray):
return obj.tolist()
if isinstance(obj, np.floating):
return float(obj)
if isinstance(obj, np.integer):
return int(obj)
except ImportError:
pass
raise TypeError(f"Object of type {type(obj)} is not JSON serializable")
|