Use own command handler

.env.example

@@ -4,6 +4,9 @@ SPOTIPY_CLIENT_SECRET=SPOTIFY_CLIENT_SECRET
 POSTGRESQL_DSN=postgresql://user:pass@ip:port/database
 LOG_LEVEL=INFO  # Defaults to INFO if not present.
 GUILD_LOGS_WEBHOOK_URL=
+TOPGG_TOKEN=
+TOPGG_WEBHOOK_AUTH=
+GUILD_IDS=1234,5678
 
 # This one is optional, use it at your own risk.
 DISCORD_BROWSER=

.github/workflows/python-app.yml

@@ -15,14 +15,11 @@ jobs:
       uses: actions/setup-python@v2
       with:
         python-version: 3.9
-    - name: Install dependencies
+    - name: Format, lint and run tests
       run: |
         python -m pip install --upgrade pip
         pip install tox
-        tox -epy39
-    - name: Format, lint and run tests
-      run: |
-        tox -eisort,black,lint,typecheck,test
+        tox -eisort,black,lint,typecheck
     - name: Check for modified files
       id: git-check
       run: echo ::set-output name=modified::$(if git diff-index --quiet HEAD --; then echo "false"; else echo "true"; fi)
@@ -33,4 +30,4 @@ jobs:
         git config --global user.email '41898282+github-actions[bot]@users.noreply.github.com'
         git remote set-url origin https://x-access-token:${{ secrets.GITHUB_TOKEN }}@github.com/${{ github.repository }}
         git commit -am "Apply auto-formatting" -m "Based on isort and black."
-        git push
+        git push

.gitignore

@@ -18,4 +18,5 @@ nokari.log
 nokari.pyz
 
 # tmp
-tmp/
+tmp/
+kita_test/

.pylintrc

@@ -11,7 +11,8 @@ disable=
     protected-access,
     too-few-public-methods,
     format,
-    cyclic-import
+    cyclic-import,
+    redefined-builtin
 
 [BASIC]
 argument-naming-style=snake_case
@@ -31,4 +32,5 @@ ignore-imports=no
 min-similarity-lines=4
 
 [DESIGN]
-max-parents=8
+max-parents=8
+max-args=10

kita/__init__.py

@@ -0,0 +1,18 @@
+__author__ = "Norizon"
+__version__ = "SNAPSHOT"
+__license__ = "MPL-2.0"
+
+from kita.buckets import *
+from kita.checks import *
+from kita.command_handlers import *
+from kita.commands import *
+from kita.contexts import *
+from kita.cooldowns import *
+from kita.data import *
+from kita.errors import *
+from kita.events import *
+from kita.extensions import *
+from kita.options import *
+from kita.responses import *
+from kita.typedefs import *
+from kita.utils import *

kita/buckets.py

@@ -0,0 +1,180 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from time import monotonic
+from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union, cast
+
+from hikari.events.interaction_events import InteractionCreateEvent
+
+from kita.typedefs import BucketHash, HashGetter, LimitGetter, PeriodGetter
+
+if TYPE_CHECKING:
+    from kita.command_handlers import GatewayCommandHandler
+
+EXPIRE_AFTER = 30.0
+
+__all__ = ("Bucket", "BucketManager")
+_LOGGER = logging.getLogger("kita.buckets")
+
+
+class EmptyBucketError(Exception):
+    def __init__(self, *args: Any, retry_after: float) -> None:
+        self.retry_after = retry_after
+        super().__init__(*args)
+
+
+class Bucket:
+    __slots__ = ("manager", "tokens", "reset_at", "hash", "limit", "period")
+
+    def __init__(self, manager: BucketManager, hash_: BucketHash) -> None:
+        self.manager: BucketManager = manager
+        self.tokens: int = 0
+        self.reset_at: float = 0.0
+        self.hash = hash_
+        self.limit: int
+        self.period: float
+
+    def set_constraints(
+        self, limit: Optional[int] = None, period: Optional[float] = None
+    ) -> None:
+        if limit is not None:
+            self.limit = limit
+
+        if period is not None:
+            self.period = period
+
+    def consume_token(self) -> None:
+        self.tokens -= 1
+
+    def acquire(self) -> None:
+        if self.is_exhausted:
+            raise EmptyBucketError(
+                "you run out of tokens.", retry_after=self.next_window
+            )
+
+        self.consume_token()
+        return None
+
+    @property
+    def is_exhausted(self) -> bool:
+        if self.reset_at > (now := monotonic()):
+            return self.tokens <= 0
+
+        self.reset_at = now + self.period
+        self.tokens = self.limit
+        return False
+
+    @property
+    def next_window(self) -> float:
+        return self.reset_at - monotonic()
+
+    @property
+    def is_expired(self) -> bool:
+        return self.reset_at + EXPIRE_AFTER < monotonic()
+
+    def invalidate(self) -> None:
+        del self.manager.buckets[self.hash]
+
+    def __repr__(self) -> str:
+        return (
+            f"<Bucket manager={self.manager.name!r} "
+            f"limit={self.limit} period={self.period} "
+            f"tokens={self.tokens} next_window={self.next_window}>"
+        )
+
+
+class BucketManager:
+    __slots__ = ("name", "buckets", "hash_getter", "period", "limit", "gc_task")
+
+    def __init__(
+        self,
+        name: str,
+        hash_getter: HashGetter,
+        limit: Union[LimitGetter, int],
+        period: Union[PeriodGetter, float],
+    ):
+        self.name = name
+        self.buckets: Dict[BucketHash, Bucket] = {}
+        self.hash_getter = hash_getter
+        self.limit = limit
+        self.period = period
+        self.gc_task: Optional[asyncio.Task[None]] = None
+
+    async def get_limit(self, handler: GatewayCommandHandler, hash_: BucketHash) -> int:
+        return (
+            await handler._invoke_callback(self.limit, hash_)
+            if callable(self.limit)
+            else self.limit
+        )
+
+    async def get_period(
+        self, handler: GatewayCommandHandler, hash_: BucketHash
+    ) -> float:
+        return (
+            await handler._invoke_callback(self.period, hash_)
+            if callable(self.period)
+            else self.period
+        )
+
+    async def get_or_create_bucket(
+        self, handler: GatewayCommandHandler, event: InteractionCreateEvent
+    ) -> Bucket:
+        bucket_hash = await handler._invoke_callback(self.hash_getter, event)
+        if not (bucket := self.buckets.get(bucket_hash)):
+            self.buckets[bucket_hash] = bucket = Bucket(self, bucket_hash)
+
+            bucket.set_constraints(
+                await self.get_limit(handler, bucket_hash),
+                await self.get_period(handler, bucket_hash),
+            )
+
+        self.ensure_gc_task()
+        return bucket
+
+    def ensure_gc_task(self) -> None:
+        if self.is_running:
+            return
+
+        self.gc_task = asyncio.create_task(self._do_gc())
+
+    def close(self) -> None:
+        if not self.is_running:
+            return
+
+        assert self.gc_task is not None
+        self.gc_task.cancel()
+        self.gc_task = None
+
+    async def _do_gc(self) -> None:
+        _LOGGER.debug("started running gc (%s bucket manager)", self.name)
+        while 1:
+            # runs periodically every `EXPIRE_AFTER` seconds
+            await asyncio.sleep(EXPIRE_AFTER)
+
+            if not self.buckets:
+                # if there's no bucket yet or all the buckets have been dead
+                # the task shall stop.
+                _LOGGER.debug(
+                    "no buckets were found, stopping the task... (%s bucket manager)",
+                    self.name,
+                )
+                self.close()
+                return
+
+            dead_buckets = [
+                bucket for bucket in self.buckets.values() if bucket.is_expired
+            ]
+
+            for bucket in dead_buckets:
+                bucket.invalidate()
+
+            _LOGGER.debug(
+                "%d buckets were invalidated (%s bucket manager)",
+                len(dead_buckets),
+                self.name,
+            )
+
+    @property
+    def is_running(self) -> bool:
+        return self.gc_task is not None

kita/checks.py

@@ -0,0 +1,112 @@
+from typing import Any, Callable, Literal, MutableMapping, Type, cast
+
+from hikari.events.interaction_events import InteractionCreateEvent
+from hikari.interactions.command_interactions import CommandInteraction
+from hikari.permissions import Permissions
+
+from kita.command_handlers import GatewayCommandHandler
+from kita.contexts import Context
+from kita.data import data
+from kita.errors import (
+    CheckAnyError,
+    DMOnlyError,
+    GuildOnlyError,
+    MissingAnyPermissionsError,
+    MissingPermissionsError,
+    OwnerOnlyError,
+)
+from kita.typedefs import CallableProto, ICommandCallback
+from kita.utils import ensure_checks
+
+__all__ = (
+    "with_check",
+    "with_check_any",
+    "guild_only",
+    "dm_only",
+    "owner_only",
+    "has_all_permissions",
+    "has_any_permissions",
+)
+
+
+def with_check(
+    predicate: CallableProto,
+) -> Callable[[CallableProto], ICommandCallback]:
+    def decorator(_command: CallableProto) -> ICommandCallback:
+        command = cast(ICommandCallback, _command)
+        ensure_checks(command)
+        command.__checks__.append(predicate)
+        return command
+
+    return decorator
+
+
+def with_check_any(
+    *predicates: CallableProto,
+) -> Callable[[CallableProto], ICommandCallback]:
+    async def _inner(
+        ctx: Context = data(Context),
+    ) -> Literal[True]:
+        exceptions = []
+        extra_env: MutableMapping[Type[Any], Any] = {
+            InteractionCreateEvent: ctx.event,
+            CommandInteraction: ctx.interaction,
+            type(ctx): ctx,
+        }
+        for predicate in predicates:
+            try:
+                if await ctx.handler._invoke_callback(predicate, extra_env=extra_env):
+                    return True
+            except Exception as exc:
+                exceptions.append(exc)
+
+        raise CheckAnyError(predicates, exceptions)
+
+    return with_check(_inner)
+
+
+def guild_only(
+    interaction: CommandInteraction = data(CommandInteraction),
+) -> Literal[True]:
+    if interaction.guild_id is None:
+        raise GuildOnlyError(f"command {interaction.command_name!r} is guild only.")
+    return True
+
+
+def dm_only(
+    interaction: CommandInteraction = data(CommandInteraction),
+) -> Literal[True]:
+    if interaction.guild_id is not None:
+        raise DMOnlyError(f"command {interaction.command_name!r} is dm only.")
+    return True
+
+
+def owner_only(
+    interaction: CommandInteraction = data(CommandInteraction),
+    handler: GatewayCommandHandler = data(GatewayCommandHandler),
+) -> Literal[True]:
+    if interaction.user.id not in handler.owner_ids:
+        raise OwnerOnlyError(f"command {interaction.command_name!r} is owner only.")
+    return True
+
+
+def has_all_permissions(perms: Permissions) -> CallableProto:
+    def inner(
+        interaction: CommandInteraction = data(CommandInteraction),
+    ) -> Literal[True]:
+        if not (member := interaction.member) or (member.permissions & perms) != perms:
+            raise MissingPermissionsError(perms)
+        return True
+
+    return inner
+
+
+def has_any_permissions(perms: Permissions) -> CallableProto:
+    def inner(
+        interaction: CommandInteraction = data(CommandInteraction),
+    ) -> Literal[True]:
+        if not ((member := interaction.member) and member.permissions & perms):
+            raise MissingAnyPermissionsError(perms)
+        return True
+
+    return inner