36 lines
1.2 KiB
Python
36 lines
1.2 KiB
Python
"""Cog wrapper for command caching."""
|
|
from datetime import datetime, timedelta
|
|
|
|
from dis_snek import InteractionContext, Scale, Snake
|
|
from dis_snek.ext.tasks.task import Task
|
|
from dis_snek.ext.tasks.triggers import IntervalTrigger
|
|
|
|
from jarvis.utils import find
|
|
|
|
|
|
class CacheCog(Scale):
|
|
"""Cog wrapper for command caching."""
|
|
|
|
def __init__(self, bot: Snake):
|
|
self.bot = bot
|
|
self.cache = {}
|
|
self._expire_interaction.start()
|
|
|
|
def check_cache(self, ctx: InteractionContext, **kwargs: dict) -> dict:
|
|
"""Check the cache."""
|
|
if not kwargs:
|
|
kwargs = {}
|
|
return find(
|
|
lambda x: x["command"] == ctx.subcommand_name # noqa: W503
|
|
and x["user"] == ctx.author.id # noqa: W503
|
|
and x["guild"] == ctx.guild.id # noqa: W503
|
|
and all(x[k] == v for k, v in kwargs.items()), # noqa: W503
|
|
self.cache.values(),
|
|
)
|
|
|
|
@Task.create(IntervalTrigger(minutes=1))
|
|
async def _expire_interaction(self) -> None:
|
|
keys = list(self.cache.keys())
|
|
for key in keys:
|
|
if self.cache[key]["timeout"] <= datetime.utcnow() + timedelta(minutes=1):
|
|
del self.cache[key]
|