forked from BasedHardware/omi
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdev_cache.py
More file actions
38 lines (28 loc) · 1.39 KB
/
Copy pathdev_cache.py
File metadata and controls
38 lines (28 loc) · 1.39 KB
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
"""Tiny, dependency-light cache of developer status (does a user hold at least
one developer API key).
This lives in its own module, separate from the heavyweight ``app_integrations``,
so the developer/dev-key router can invalidate it on a key change without pulling
the notification/LLM import chain into that lightweight surface. The proactive
notification cap exempts developers and is checked on every attempt, so caching
keeps the check off the Firestore hot path; the short TTL bounds staleness and the
explicit invalidate makes a create/delete take effect immediately.
"""
import threading
from typing import Optional
from cachetools import TTLCache
_DEV_STATUS_CACHE: TTLCache[str, bool] = TTLCache(maxsize=4096, ttl=300)
_lock = threading.Lock()
def get_cached_developer(uid: str) -> Optional[bool]:
"""Cached developer status, or None when not cached."""
with _lock:
return _DEV_STATUS_CACHE.get(uid)
def set_cached_developer(uid: str, is_developer: bool) -> None:
with _lock:
_DEV_STATUS_CACHE[uid] = is_developer
def invalidate_developer_cache(uid: str) -> None:
"""Drop a user's cached developer status so a dev-key create/delete takes
effect immediately instead of waiting out the TTL — otherwise a brand-new
developer could stay capped, or a just-revoked one stay exempt, until the
entry expires."""
with _lock:
_DEV_STATUS_CACHE.pop(uid, None)