Slipstream
ed83a9da85
Introduces a flexible caching infrastructure with time-to-live support and configurable member caching based on status, voice state, and join events. Adds AudioSink abstract base class to support audio output handling in voice connections. Replaces direct dictionary access with cache objects throughout the client, enabling automatic expiration and intelligent member filtering based on user-defined flags. Updates guild parsing to incorporate presence and voice state data for more accurate member caching decisions.
86 lines
2.4 KiB
Python
86 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import TYPE_CHECKING, Dict, Generic, Optional, TypeVar
|
|
|
|
if TYPE_CHECKING:
|
|
from .models import Channel, Guild, Member
|
|
from .caching import MemberCacheFlags
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class Cache(Generic[T]):
|
|
"""Simple in-memory cache with optional TTL support."""
|
|
|
|
def __init__(self, ttl: Optional[float] = None) -> None:
|
|
self.ttl = ttl
|
|
self._data: Dict[str, tuple[T, Optional[float]]] = {}
|
|
|
|
def set(self, key: str, value: T) -> None:
|
|
expiry = time.monotonic() + self.ttl if self.ttl is not None else None
|
|
self._data[key] = (value, expiry)
|
|
|
|
def get(self, key: str) -> Optional[T]:
|
|
item = self._data.get(key)
|
|
if not item:
|
|
return None
|
|
value, expiry = item
|
|
if expiry is not None and expiry < time.monotonic():
|
|
self.invalidate(key)
|
|
return None
|
|
return value
|
|
|
|
def invalidate(self, key: str) -> None:
|
|
self._data.pop(key, None)
|
|
|
|
def clear(self) -> None:
|
|
self._data.clear()
|
|
|
|
def values(self) -> list[T]:
|
|
now = time.monotonic()
|
|
items = []
|
|
for key, (value, expiry) in list(self._data.items()):
|
|
if expiry is not None and expiry < now:
|
|
self.invalidate(key)
|
|
else:
|
|
items.append(value)
|
|
return items
|
|
|
|
|
|
class GuildCache(Cache["Guild"]):
|
|
"""Cache specifically for :class:`Guild` objects."""
|
|
|
|
|
|
class ChannelCache(Cache["Channel"]):
|
|
"""Cache specifically for :class:`Channel` objects."""
|
|
|
|
|
|
class MemberCache(Cache["Member"]):
|
|
"""
|
|
A cache for :class:`Member` objects that respects :class:`MemberCacheFlags`.
|
|
"""
|
|
|
|
def __init__(self, flags: MemberCacheFlags, ttl: Optional[float] = None) -> None:
|
|
super().__init__(ttl)
|
|
self.flags = flags
|
|
|
|
def _should_cache(self, member: Member) -> bool:
|
|
"""Determines if a member should be cached based on the flags."""
|
|
if self.flags.all:
|
|
return True
|
|
if self.flags.none:
|
|
return False
|
|
|
|
if self.flags.online and member.status != "offline":
|
|
return True
|
|
if self.flags.voice and member.voice_state is not None:
|
|
return True
|
|
if self.flags.joined and getattr(member, "_just_joined", False):
|
|
return True
|
|
return False
|
|
|
|
def set(self, key: str, value: Member) -> None:
|
|
if self._should_cache(value):
|
|
super().set(key, value)
|