Spaces:
Running on Zero
Running on Zero
| """In-memory user snapshots (so most updates need ZERO database round-trips) | |
| plus per-user locks (same user's updates run one at a time).""" | |
| import asyncio | |
| import time | |
| import weakref | |
| from typing import Optional | |
| _TTL = 600 # seconds | |
| _MAX = 20_000 | |
| _snapshots: dict[int, tuple[float, dict]] = {} | |
| _locks: "weakref.WeakValueDictionary[int, asyncio.Lock]" = weakref.WeakValueDictionary() | |
| def get(user_id: int) -> Optional[dict]: | |
| item = _snapshots.get(user_id) | |
| if item is None: | |
| return None | |
| expires, data = item | |
| if expires < time.monotonic(): | |
| _snapshots.pop(user_id, None) | |
| return None | |
| return dict(data) | |
| def put(user_id: int, data: dict) -> None: | |
| if len(_snapshots) >= _MAX: | |
| _snapshots.clear() | |
| _snapshots[user_id] = (time.monotonic() + _TTL, dict(data)) | |
| def invalidate(user_id: int) -> None: | |
| _snapshots.pop(user_id, None) | |
| def lock_for(user_id: int) -> asyncio.Lock: | |
| lock = _locks.get(user_id) | |
| if lock is None: | |
| lock = asyncio.Lock() | |
| _locks[user_id] = lock | |
| return lock | |