diff --git a/.gitignore b/.gitignore index 429ac009..250f982c 100644 --- a/.gitignore +++ b/.gitignore @@ -1,12 +1,7 @@ -dblpy.egg-info/ -topggpy.egg-info/ -topgg/__pycache__/ +**/__pycache__/ +.ruff_cache/ +.vscode/ build/ +docs/_build/ dist/ -/docs/_build -/docs/_templates -.vscode -/.idea/ -__pycache__ -.coverage -.pytest_cache/ +topggpy.egg-info/ \ No newline at end of file diff --git a/LICENSE b/LICENSE index 96aaaf80..3a68f837 100644 --- a/LICENSE +++ b/LICENSE @@ -1,7 +1,22 @@ -Copyright 2021 Assanali Mukhanov & Top.gg +The MIT License (MIT) -Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: +Copyright (c) 2021 Assanali Mukhanov & Top.gg +Copyright (c) 2024-2025 null8626 & Top.gg -The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. \ No newline at end of file diff --git a/mypy.ini b/mypy.ini deleted file mode 100644 index 0f2c80ed..00000000 --- a/mypy.ini +++ /dev/null @@ -1,21 +0,0 @@ -# Global options: - -[mypy] -python_version = 3.7 -check_untyped_defs = True -no_implicit_optional = True -ignore_missing_imports = True - -# Allows -allow_untyped_globals = False -allow_redefinition = True - -# Disallows -disallow_incomplete_defs = True -disallow_untyped_defs = True - -# Warns -warn_redundant_casts = True -warn_unreachable = True -warn_unused_configs = True -warn_unused_ignores = True \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 7f800f43..e814a877 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,36 +3,33 @@ requires = ["setuptools"] [project] name = "topggpy" -version = "3.0.0" -description = "A community-maintained Python API Client for the Top.gg API." +version = "1.5.0" +description = "A simple API wrapper for Top.gg written in Python." readme = "README.md" license = { text = "MIT" } authors = [{ name = "null8626" }, { name = "Top.gg" }] keywords = ["discord", "discord-bot", "topgg"] -dependencies = ["aiohttp>=3.12.15"] +dependencies = ["aiohttp>=3.13.1"] classifiers = [ "Development Status :: 5 - Production/Stable", "Intended Audience :: Developers", "Natural Language :: English", "Operating System :: OS Independent", "Programming Language :: Python :: 3 :: Only", - "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", "Topic :: Internet", "Topic :: Software Development :: Libraries", "Topic :: Software Development :: Libraries :: Python Modules", "Topic :: Utilities" ] -requires-python = ">=3.9" - -[project.optional-dependencies] -dev = ["mock>=5.2.0", "pytest>=8.4.2", "pytest-asyncio>=1.2.0", "pytest-mock>=3.15.0", "pytest-cov>=7.0.0", "ruff>=0.13.0"] +requires-python = ">=3.10" [project.urls] Documentation = "https://topggpy.readthedocs.io/en/latest/" "Raw API Documentation" = "https://docs.top.gg/docs/" Repository = "https://github.com/top-gg-community/python-sdk" -"Support server" = "https://discord.gg/dbl" \ No newline at end of file +"Support server" = "https://discord.gg/EYHTgJX" \ No newline at end of file diff --git a/pytest.ini b/pytest.ini deleted file mode 100644 index 919cbc45..00000000 --- a/pytest.ini +++ /dev/null @@ -1,8 +0,0 @@ -[pytest] -xfail_strict = true -norecursedirs = docs *.egg-info .git - -filterwarnings = - ignore::DeprecationWarning - -addopts = --cov=topgg \ No newline at end of file diff --git a/scripts/format.sh b/scripts/format.sh deleted file mode 100644 index 4fa2e31d..00000000 --- a/scripts/format.sh +++ /dev/null @@ -1,2 +0,0 @@ -black . -isort . \ No newline at end of file diff --git a/tests/test_autopost.py b/tests/test_autopost.py index b83a0cc0..726355bd 100644 --- a/tests/test_autopost.py +++ b/tests/test_autopost.py @@ -1,95 +1,95 @@ -import datetime - -import mock -import pytest -from aiohttp import ClientSession -from pytest_mock import MockerFixture - -from topgg import DBLClient -from topgg.autopost import AutoPoster -from topgg.errors import HTTPException, TopGGException - - -MOCK_TOKEN = ".eyJfdCI6IiIsImlkIjoiMzY0ODA2MDI5ODc2NTU1Nzc2In0=." - - -@pytest.fixture -def session() -> ClientSession: - return mock.Mock(ClientSession) - - -@pytest.fixture -def autopost(session: ClientSession) -> AutoPoster: - return AutoPoster(DBLClient(MOCK_TOKEN, session=session)) - - -@pytest.mark.asyncio -async def test_AutoPoster_breaks_autopost_loop_on_401( - mocker: MockerFixture, session: ClientSession -) -> None: - response = mock.Mock("reason, status") - response.reason = "Unauthorized" - response.status = 401 - - mocker.patch( - "topgg.DBLClient.post_guild_count", side_effect=HTTPException(response, {}) - ) - - callback = mock.Mock() - autopost = DBLClient(MOCK_TOKEN, session=session).autopost().stats(callback) - assert isinstance(autopost, AutoPoster) - assert not isinstance(autopost.stats()(callback), AutoPoster) - - with pytest.raises(HTTPException): - await autopost.start() - - callback.assert_called_once() - assert not autopost.is_running - - -@pytest.mark.asyncio -async def test_AutoPoster_raises_missing_stats(autopost: AutoPoster) -> None: - with pytest.raises( - TopGGException, match="you must provide a callback that returns the stats." - ): - await autopost.start() - - -@pytest.mark.asyncio -async def test_AutoPoster_raises_already_running(autopost: AutoPoster) -> None: - autopost.stats(mock.Mock()).start() - with pytest.raises(TopGGException, match="the autopost is already running."): - await autopost.start() - - -@pytest.mark.asyncio -async def test_AutoPoster_interval_too_short(autopost: AutoPoster) -> None: - with pytest.raises(ValueError, match="interval must be greated than 900 seconds."): - autopost.set_interval(50) - - -@pytest.mark.asyncio -async def test_AutoPoster_error_callback( - mocker: MockerFixture, autopost: AutoPoster -) -> None: - error_callback = mock.Mock() - response = mock.Mock("reason, status") - response.reason = "Internal Server Error" - response.status = 500 - side_effect = HTTPException(response, {}) - - mocker.patch("topgg.DBLClient.post_guild_count", side_effect=side_effect) - task = autopost.on_error(error_callback).stats(mock.Mock()).start() - autopost.stop() - await task - error_callback.assert_called_once_with(side_effect) - - -def test_AutoPoster_interval(autopost: AutoPoster): - assert autopost.interval == 900 - autopost.set_interval(datetime.timedelta(hours=1)) - assert autopost.interval == 3600 - autopost.interval = datetime.timedelta(hours=2) - assert autopost.interval == 7200 - autopost.interval = 3600 - assert autopost.interval == 3600 +import datetime + +import mock +import pytest +from aiohttp import ClientSession +from pytest_mock import MockerFixture + +from topgg import DBLClient +from topgg.autopost import AutoPoster +from topgg.errors import HTTPException, TopGGException + + +MOCK_TOKEN = ".eyJfdCI6IiIsImlkIjoiMzY0ODA2MDI5ODc2NTU1Nzc2In0=." + + +@pytest.fixture +def session() -> ClientSession: + return mock.Mock(ClientSession) + + +@pytest.fixture +def autopost(session: ClientSession) -> AutoPoster: + return AutoPoster(DBLClient(MOCK_TOKEN, session=session)) + + +@pytest.mark.asyncio +async def test_AutoPoster_breaks_autopost_loop_on_401( + mocker: MockerFixture, session: ClientSession +) -> None: + response = mock.Mock("reason, status") + response.reason = "Unauthorized" + response.status = 401 + + mocker.patch( + "topgg.DBLClient.post_guild_count", side_effect=HTTPException(response, {}) + ) + + callback = mock.Mock() + autopost = DBLClient(MOCK_TOKEN, session=session).autopost().stats(callback) + assert isinstance(autopost, AutoPoster) + assert not isinstance(autopost.stats()(callback), AutoPoster) + + with pytest.raises(HTTPException): + await autopost.start() + + callback.assert_called_once() + assert not autopost.is_running + + +@pytest.mark.asyncio +async def test_AutoPoster_raises_missing_stats(autopost: AutoPoster) -> None: + with pytest.raises( + TopGGException, match="you must provide a callback that returns the stats." + ): + await autopost.start() + + +@pytest.mark.asyncio +async def test_AutoPoster_raises_already_running(autopost: AutoPoster) -> None: + autopost.stats(mock.Mock()).start() + with pytest.raises(TopGGException, match="the autopost is already running."): + await autopost.start() + + +@pytest.mark.asyncio +async def test_AutoPoster_interval_too_short(autopost: AutoPoster) -> None: + with pytest.raises(ValueError, match="interval must be greated than 900 seconds."): + autopost.set_interval(50) + + +@pytest.mark.asyncio +async def test_AutoPoster_error_callback( + mocker: MockerFixture, autopost: AutoPoster +) -> None: + error_callback = mock.Mock() + response = mock.Mock("reason, status") + response.reason = "Internal Server Error" + response.status = 500 + side_effect = HTTPException(response, {}) + + mocker.patch("topgg.DBLClient.post_guild_count", side_effect=side_effect) + task = autopost.on_error(error_callback).stats(mock.Mock()).start() + autopost.stop() + await task + error_callback.assert_called_once_with(side_effect) + + +def test_AutoPoster_interval(autopost: AutoPoster): + assert autopost.interval == 900 + autopost.set_interval(datetime.timedelta(hours=1)) + assert autopost.interval == 3600 + autopost.interval = datetime.timedelta(hours=2) + assert autopost.interval == 7200 + autopost.interval = 3600 + assert autopost.interval == 3600 diff --git a/tests/test_ratelimiter.py b/tests/test_ratelimiter.py index 998a7357..53692fe4 100644 --- a/tests/test_ratelimiter.py +++ b/tests/test_ratelimiter.py @@ -1,28 +1,28 @@ -import pytest - -from topgg.ratelimiter import Ratelimiter - -n = period = 10 - - -@pytest.fixture -def limiter() -> Ratelimiter: - return Ratelimiter(max_calls=n, period=period) - - -@pytest.mark.asyncio -async def test_AsyncRateLimiter_calls(limiter: Ratelimiter) -> None: - for _ in range(n): - async with limiter: - pass - - assert len(limiter._Ratelimiter__calls) == limiter._Ratelimiter__max_calls == n - - -@pytest.mark.asyncio -async def test_AsyncRateLimiter_timespan_property(limiter: Ratelimiter) -> None: - for _ in range(n): - async with limiter: - pass - - assert limiter._timespan < period +import pytest + +from topgg.ratelimiter import Ratelimiter + +n = period = 10 + + +@pytest.fixture +def limiter() -> Ratelimiter: + return Ratelimiter(max_calls=n, period=period) + + +@pytest.mark.asyncio +async def test_AsyncRateLimiter_calls(limiter: Ratelimiter) -> None: + for _ in range(n): + async with limiter: + pass + + assert len(limiter._Ratelimiter__calls) == limiter._Ratelimiter__max_calls == n + + +@pytest.mark.asyncio +async def test_AsyncRateLimiter_timespan_property(limiter: Ratelimiter) -> None: + for _ in range(n): + async with limiter: + pass + + assert limiter._timespan < period diff --git a/tests/test_webhook.py b/tests/test_webhook.py index 8ef3c71d..93fbc1d4 100644 --- a/tests/test_webhook.py +++ b/tests/test_webhook.py @@ -1,80 +1,80 @@ -import typing as t - -import aiohttp -import mock -import pytest - -from topgg import WebhookManager, WebhookType -from topgg.errors import TopGGException - -auth = "youshallnotpass" - - -@pytest.fixture -def webhook_manager() -> WebhookManager: - return ( - WebhookManager() - .endpoint() - .type(WebhookType.BOT) - .auth(auth) - .route("/dbl") - .callback(print) - .add_to_manager() - .endpoint() - .type(WebhookType.GUILD) - .auth(auth) - .route("/dsl") - .callback(print) - .add_to_manager() - ) - - -def test_WebhookManager_routes(webhook_manager: WebhookManager) -> None: - assert len(webhook_manager.app.router.routes()) == 2 - - -@pytest.mark.asyncio -@pytest.mark.parametrize( - "headers, result, state", - [({"authorization": auth}, 200, True), ({}, 401, False)], -) -async def test_WebhookManager_validates_auth( - webhook_manager: WebhookManager, headers: t.Dict[str, str], result: int, state: bool -) -> None: - await webhook_manager.start(5000) - - try: - for path in ("dbl", "dsl"): - async with aiohttp.request( - "POST", f"http://localhost:5000/{path}", headers=headers, json={} - ) as r: - assert r.status == result - finally: - await webhook_manager.close() - assert not webhook_manager.is_running - - -def test_WebhookEndpoint_callback_unset(webhook_manager: WebhookManager): - with pytest.raises( - TopGGException, - match="endpoint missing callback.", - ): - webhook_manager.endpoint().add_to_manager() - - -def test_WebhookEndpoint_route_unset(webhook_manager: WebhookManager): - with pytest.raises( - TopGGException, - match="endpoint missing type.", - ): - webhook_manager.endpoint().callback(mock.Mock()).add_to_manager() - - -def test_WebhookEndpoint_type_unset(webhook_manager: WebhookManager): - with pytest.raises( - TopGGException, - match="endpoint missing route.", - ): - webhook_manager.endpoint().callback(mock.Mock()).type( - WebhookType.BOT - ).add_to_manager() +import typing as t + +import aiohttp +import mock +import pytest + +from topgg import WebhookManager, WebhookType +from topgg.errors import TopGGException + +auth = "youshallnotpass" + + +@pytest.fixture +def webhook_manager() -> WebhookManager: + return ( + WebhookManager() + .endpoint() + .type(WebhookType.BOT) + .auth(auth) + .route("/dbl") + .callback(print) + .add_to_manager() + .endpoint() + .type(WebhookType.GUILD) + .auth(auth) + .route("/dsl") + .callback(print) + .add_to_manager() + ) + + +def test_WebhookManager_routes(webhook_manager: WebhookManager) -> None: + assert len(webhook_manager.app.router.routes()) == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "headers, result, state", + [({"authorization": auth}, 200, True), ({}, 401, False)], +) +async def test_WebhookManager_validates_auth( + webhook_manager: WebhookManager, headers: t.Dict[str, str], result: int, state: bool +) -> None: + await webhook_manager.start(5000) + + try: + for path in ("dbl", "dsl"): + async with aiohttp.request( + "POST", f"http://localhost:5000/{path}", headers=headers, json={} + ) as r: + assert r.status == result + finally: + await webhook_manager.close() + assert not webhook_manager.is_running + + +def test_WebhookEndpoint_callback_unset(webhook_manager: WebhookManager): + with pytest.raises( + TopGGException, + match="endpoint missing callback.", + ): + webhook_manager.endpoint().add_to_manager() + + +def test_WebhookEndpoint_route_unset(webhook_manager: WebhookManager): + with pytest.raises( + TopGGException, + match="endpoint missing type.", + ): + webhook_manager.endpoint().callback(mock.Mock()).add_to_manager() + + +def test_WebhookEndpoint_type_unset(webhook_manager: WebhookManager): + with pytest.raises( + TopGGException, + match="endpoint missing route.", + ): + webhook_manager.endpoint().callback(mock.Mock()).type( + WebhookType.BOT + ).add_to_manager() diff --git a/topgg/__init__.py b/topgg/__init__.py index 58c1bce7..b5290786 100644 --- a/topgg/__init__.py +++ b/topgg/__init__.py @@ -1,27 +1,105 @@ -# -*- coding: utf-8 -*- - """ -Top.gg Python API Wrapper -~~~~~~~~~~~~~~~~~~~~~~~~~ -A basic wrapper for the Top.gg API. -:copyright: (c) 2021 Assanali Mukhanov & Top.gg -:license: MIT, see LICENSE for more details. +The MIT License (MIT) + +Copyright (c) 2021 Assanali Mukhanov & Top.gg +Copyright (c) 2024-2025 null8626 & Top.gg + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. """ +from .autopost import AutoPoster +from .client import DBLClient +from .data import data, DataContainerMixin +from .errors import ( + ClientException, + ClientStateException, + HTTPException, + TopGGException, + Ratelimited, +) +from .types import ( + BotData, + BotsData, + BotStatsData, + BotVoteData, + BriefUserData, + GuildVoteData, + ServerVoteData, + SocialData, + SortBy, + StatsWrapper, + UserData, + VoteDataDict, + WidgetOptions, + WidgetProjectType, + WidgetType, +) from .version import VERSION +from .webhook import ( + BoundWebhookEndpoint, + endpoint, + WebhookEndpoint, + WebhookManager, + WebhookType, +) + __title__ = "topggpy" -__author__ = "Assanali Mukhanov" -__maintainer__ = "Norizon" +__author__ = "null8626 & Top.gg" +__credits__ = ("null8626", "Top.gg") +__maintainer__ = "null8626" +__status__ = "Production" __license__ = "MIT" +__copyright__ = "Copyright (c) 2021 Assanali Mukhanov & Top.gg; Copyright (c) 2024-2025 null8626 & Top.gg" __version__ = VERSION - -from .autopost import * -from .client import * -from .data import * -from .errors import * - -# can't be added to __all__ since they'd clash with automodule -from .types import * -from .types import BotVoteData, GuildVoteData -from .webhook import * +__all__ = ( + "AutoPoster", + "BotData", + "BotsData", + "BotStatsData", + "BotVoteData", + "BoundWebhookEndpoint", + "BriefUserData", + "ClientException", + "ClientStateException", + "data", + "DataContainerMixin", + "DBLClient", + "endpoint", + "GuildVoteData", + "HTTPException", + "Ratelimited", + "RequestError", + "ServerVoteData", + "SocialData", + "SortBy", + "StatsWrapper", + "TopGGException", + "UserData", + "VERSION", + "VoteDataDict", + "VoteEvent", + "Voter", + "WebhookEndpoint", + "WebhookManager", + "WebhookType", + "WidgetOptions", + "WidgetProjectType", + "WidgetType", +) diff --git a/topgg/autopost.py b/topgg/autopost.py index c1953f89..4ee10630 100644 --- a/topgg/autopost.py +++ b/topgg/autopost.py @@ -1,26 +1,27 @@ -# The MIT License (MIT) - -# Copyright (c) 2021 Norizon - -# Permission is hereby granted, free of charge, to any person obtaining a -# copy of this software and associated documentation files (the "Software"), -# to deal in the Software without restriction, including without limitation -# the rights to use, copy, modify, merge, publish, distribute, sublicense, -# and/or sell copies of the Software, and to permit persons to whom the -# Software is furnished to do so, subject to the following conditions: - -# The above copyright notice and this permission notice shall be included in -# all copies or substantial portions of the Software. - -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -# DEALINGS IN THE SOFTWARE. - -__all__ = ["AutoPoster"] +""" +The MIT License (MIT) + +Copyright (c) 2021 Norizon & Top.gg +Copyright (c) 2024-2025 null8626 & Top.gg + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. +""" import asyncio import datetime @@ -28,33 +29,28 @@ import traceback import typing as t -from topgg import errors - -from .types import StatsWrapper +from . import errors if t.TYPE_CHECKING: - import asyncio - from .client import DBLClient + from .types import StatsWrapper + CallbackT = t.Callable[..., t.Any] -StatsCallbackT = t.Callable[[], StatsWrapper] +StatsCallbackT = t.Callable[[], "StatsWrapper"] class AutoPoster: """ - A helper class for autoposting. Takes in a :obj:`~.client.DBLClient` to instantiate. + Automatically update the statistics in your Discord bot's Top.gg page every few minutes. - Note: - You should not instantiate this unless you know what you're doing. - Generally, you'd better use the :meth:`~.client.DBLClient.autopost` method. + Note that you should NOT instantiate this directly unless you know what you're doing. Generally, it's recommended to use the :meth:`.DBLClient.autopost` method instead. - Args: - client (:obj:`~.client.DBLClient`) - An instance of DBLClient. + :param client: The client instance to use. + :type client: :class:`.DBLClient` """ - __slots__ = ( + __slots__: tuple[str, ...] = ( "_error", "_success", "_interval", @@ -64,153 +60,167 @@ class AutoPoster: "_stopping", ) - _success: CallbackT - _stats: CallbackT - _interval: float - _task: t.Optional["asyncio.Task[None]"] - def __init__(self, client: "DBLClient") -> None: super().__init__() + self.client = client - self._interval: float = 900 + self._interval = 900 self._error = self._default_error_handler self._refresh_state() + def __repr__(self) -> str: + return f"<{__class__.__name__} is_running={self.is_running}>" + + def __bool__(self) -> bool: + return self.is_running + def _default_error_handler(self, exception: Exception) -> None: print("Ignoring exception in auto post loop:", file=sys.stderr) + traceback.print_exception( type(exception), exception, exception.__traceback__, file=sys.stderr ) @t.overload - def on_success(self, callback: None) -> t.Callable[[CallbackT], CallbackT]: - ... + def on_success(self, callback: None) -> t.Callable[[CallbackT], CallbackT]: ... @t.overload - def on_success(self, callback: CallbackT) -> "AutoPoster": - ... + def on_success(self, callback: CallbackT) -> "AutoPoster": ... - def on_success(self, callback: t.Any = None) -> t.Any: + def on_success(self, callback: t.Any = None) -> t.Union[t.Any, "AutoPoster"]: """ Registers an autopost success callback. The callback can be either sync or async. The callback is not required to take in arguments, but you can have injected :obj:`~.data.data`. This method can be used as a decorator or a decorator factory. - :Example: - .. code-block:: python + .. code-block:: python - # The following are valid. - autopost = dblclient.autopost().on_success(lambda: print("Success!")) + # The following are valid. + autoposter = client.autopost().on_success(lambda: print("Success!")) - # Used as decorator, the decorated function will become the AutoPoster object. - @autopost.on_success - def autopost(): - ... - # Used as decorator factory, the decorated function will still be the function itself. - @autopost.on_success() - def on_success(): - ... + # Used as decorator, the decorated function will become the AutoPoster object. + @autoposter.on_success + def on_success() -> None: ... + + + # Used as decorator factory, the decorated function will still be the function itself. + @autoposter.on_success() + def on_success() -> None: ... + + :param callback: The autoposter's new success callback. + :type callback: Any + + :returns: The object itself if ``callback`` is not :py:obj:`None`, otherwise the decorator function. + :rtype: Union[Any, :class:`.AutoPoster`] """ + if callback is not None: self._success = callback + return self def decorator(callback: CallbackT) -> CallbackT: self._success = callback + return callback return decorator @t.overload - def on_error(self, callback: None) -> t.Callable[[CallbackT], CallbackT]: - ... + def on_error(self, callback: None) -> t.Callable[[CallbackT], CallbackT]: ... @t.overload - def on_error(self, callback: CallbackT) -> "AutoPoster": - ... + def on_error(self, callback: CallbackT) -> "AutoPoster": ... - def on_error(self, callback: t.Any = None) -> t.Any: + def on_error(self, callback: t.Any = None) -> t.Union[t.Any, "AutoPoster"]: """ Registers an autopost error callback. The callback can be either sync or async. - The callback is expected to take in the exception being raised, you can also - have injected :obj:`~.data.data`. + The callback is expected to take in the exception being raised, you can also have injected :obj:`~.data.data`. This method can be used as a decorator or a decorator factory. - Note: - If you don't provide an error callback, the default error handler will be called. + Note that if you don't provide an error callback, the default error handler will be called. + + .. code-block:: python + + # The following are valid. + autoposter = client.autopost().on_error(lambda err: print(f"Error! {err!r}")) - :Example: - .. code-block:: python - # The following are valid. - autopost = dblclient.autopost().on_error(lambda exc: print("Failed posting stats!", exc)) + # Used as decorator, the decorated function will become the AutoPoster object. + @autoposter.on_error + def on_error(err: Exception) -> None: ... - # Used as decorator, the decorated function will become the AutoPoster object. - @autopost.on_error - def autopost(exc: Exception): - ... - # Used as decorator factory, the decorated function will still be the function itself. - @autopost.on_error() - def on_error(exc: Exception): - ... + # Used as decorator factory, the decorated function will still be the function itself. + @autoposter.on_error() + def on_error(err: Exception) -> None: ... + + :param callback: The autoposter's new error callback. + :type callback: Any + + :returns: The object itself if ``callback`` is not :py:obj:`None`, otherwise the decorator function. + :rtype: Union[Any, :class:`.AutoPoster`] """ if callback is not None: self._error = callback + return self def decorator(callback: CallbackT) -> CallbackT: self._error = callback + return callback return decorator @t.overload - def stats(self, callback: None) -> t.Callable[[StatsCallbackT], StatsCallbackT]: - ... + def stats(self, callback: None) -> t.Callable[[StatsCallbackT], StatsCallbackT]: ... @t.overload - def stats(self, callback: StatsCallbackT) -> "AutoPoster": - ... + def stats(self, callback: StatsCallbackT) -> "AutoPoster": ... - def stats(self, callback: t.Any = None) -> t.Any: + def stats(self, callback: t.Any = None) -> t.Union[t.Any, "AutoPoster"]: """ - Registers a function that returns an instance of :obj:`~.types.StatsWrapper`. + Registers a function that returns an instance of :class:`.StatsWrapper`. The callback can be either sync or async. - The callback can be either sync or async. The callback is not required to take in arguments, but you can have injected :obj:`~.data.data`. This method can be used as a decorator or a decorator factory. - :Example: - .. code-block:: python + .. code-block:: python + + # The following are valid. + autoposter = client.autopost().stats(lambda: topgg.StatsWrapper(bot.server_count)) + - import topgg + # Used as decorator, the decorated function will become the AutoPoster object. + @autoposter.stats + def get_stats() -> topgg.StatsWrapper: + return topgg.StatsWrapper(bot.server_count) - # In this example, we fetch the stats from a Discord client instance. - client = Client(...) - dblclient = topgg.DBLClient(TOKEN).set_data(client) - autopost = ( - dblclient - .autopost() - .on_success(lambda: print("Successfully posted the stats!") - ) - @autopost.stats() - def get_stats(client: Client = topgg.data(Client)): - return topgg.StatsWrapper(guild_count=len(client.guilds), shard_count=len(client.shards)) + # Used as decorator factory, the decorated function will still be the function itself. + @autoposter.stats() + def get_stats() -> topgg.StatsWrapper: + return topgg.StatsWrapper(bot.server_count) - # somewhere after the event loop has started - autopost.start() + :param callback: The autoposter's new statistics callback. + :type callback: Any + + :returns: The object itself if ``callback`` is not :py:obj:`None`, otherwise the decorator function. + :rtype: Union[Any, :class:`.AutoPoster`] """ + if callback is not None: self._stats = callback + return self def decorator(callback: StatsCallbackT) -> StatsCallbackT: self._stats = callback + return callback return decorator @@ -218,37 +228,42 @@ def decorator(callback: StatsCallbackT) -> StatsCallbackT: @property def interval(self) -> float: """The interval between posting stats.""" + return self._interval @interval.setter def interval(self, seconds: t.Union[float, datetime.timedelta]) -> None: - """Alias to :meth:`~.autopost.AutoPoster.set_interval`.""" + """Alias of :meth:`.AutoPoster.set_interval`.""" + self.set_interval(seconds) def set_interval(self, seconds: t.Union[float, datetime.timedelta]) -> "AutoPoster": """ Sets the interval between posting stats. - Args: - seconds (:obj:`typing.Union` [ :obj:`float`, :obj:`datetime.timedelta` ]) - The interval. + :param seconds: The interval in seconds. + :type seconds: Union[:py:class:`float`, :py:class:`~datetime.timedelta`] - Raises: - :obj:`ValueError` - If the provided interval is less than 900 seconds. + :exception ValueError: The provided interval is less than 900 seconds. + + :returns: The object itself. + :rtype: :class:`.AutoPoster` """ + if isinstance(seconds, datetime.timedelta): seconds = seconds.total_seconds() if seconds < 900: - raise ValueError("interval must be greated than 900 seconds.") + raise ValueError("interval must be greater than 900 seconds.") self._interval = seconds + return self @property def is_running(self) -> bool: - """Whether or not the autopost is running.""" + """Whether the autoposter is running.""" + return self._task is not None and not self._task.done() def _refresh_state(self) -> None: @@ -257,27 +272,30 @@ def _refresh_state(self) -> None: def _fut_done_callback(self, future: "asyncio.Future") -> None: self._refresh_state() + if future.cancelled(): return + future.exception() async def _internal_loop(self) -> None: try: while 1: stats = await self.client._invoke_callback(self._stats) + try: await self.client.post_guild_count(stats) except Exception as err: await self.client._invoke_callback(self._error, err) + if isinstance(err, errors.HTTPException) and err.code == 401: raise err from None else: - on_success = getattr(self, "_success", None) - if on_success: + if on_success := getattr(self, "_success", None): await self.client._invoke_callback(on_success) if self._stopping: - return None + return await asyncio.sleep(self.interval) finally: @@ -285,51 +303,49 @@ async def _internal_loop(self) -> None: def start(self) -> "asyncio.Task[None]": """ - Starts the autoposting loop. + Starts the autoposter loop. + + Note that this method must be called when the event loop is already running! - Note: - This method must be called when the event loop has already running! + :exception TopGGException: There's no callback provided or the autoposter is already running. - Raises: - :obj:`~.errors.TopGGException` - If there's no callback provided or the autopost is already running. + :returns: The autoposter loop's :class:`~asyncio.Task`. + :rtype: :class:`~asyncio.Task`. """ + if not hasattr(self, "_stats"): raise errors.TopGGException( - "you must provide a callback that returns the stats." + "You must provide a callback that returns the stats." ) - - if self.is_running: - raise errors.TopGGException("the autopost is already running.") + elif self.is_running: + raise errors.TopGGException("The autoposter is already running.") self._task = task = asyncio.ensure_future(self._internal_loop()) task.add_done_callback(self._fut_done_callback) + return task def stop(self) -> None: """ - Stops the autoposting loop. + Stops the autoposter loop. - Note: - This differs from :meth:`~.autopost.AutoPoster.cancel` - because this will post once before stopping as opposed to cancel immediately. + Not to be confused with :meth:`.AutoPoster.cancel`, which stops the loop immediately instead of waiting for another post before stopping. """ + if not self.is_running: - return None + return self._stopping = True def cancel(self) -> None: """ - Cancels the autoposting loop. + Cancels the autoposter loop. - Note: - This differs from :meth:`~.autopost.AutoPoster.stop` - because this will stop the loop right away. + Not to be confused with :meth:`.AutoPoster.stop`, which waits for another post before stopping instead of stopping the loop immediately. """ + if self._task is None: return self._task.cancel() self._refresh_state() - return None diff --git a/topgg/client.py b/topgg/client.py index 42c714e7..e71be24f 100644 --- a/topgg/client.py +++ b/topgg/client.py @@ -23,69 +23,103 @@ SOFTWARE. """ -__all__ = ["DBLClient"] - +from aiohttp import ClientResponseError, ClientSession, ClientTimeout +from typing import Any, Optional, overload, Union from collections import namedtuple from base64 import b64decode from time import time -import typing as t import binascii -import aiohttp +import warnings import asyncio import json +from . import errors, types from .autopost import AutoPoster -from . import errors, types, VERSION -from .data import DataContainerMixin from .ratelimiter import Ratelimiter, Ratelimiters +from .data import DataContainerMixin +from .version import VERSION -BASE_URL = 'https://top.gg/api' +BASE_URL = "https://top.gg/api" MAXIMUM_DELAY_THRESHOLD = 5.0 class DBLClient(DataContainerMixin): - """Represents a client connection that connects to Top.gg. + """ + Interact with the API's endpoints. + + Examples: + + .. code-block:: python + + # Explicit cleanup + client = topgg.DBLClient(os.getenv('TOPGG_TOKEN')) - This class is used to interact with the Top.gg API. + # ... - .. _aiohttp session: https://aiohttp.readthedocs.io/en/stable/client_reference.html#client-session + await client.close() - Args: - token (:obj:`str`): Your Top.gg API Token. + # Implicit cleanup + async with topgg.DBLClient(os.getenv('TOPGG_TOKEN')) as client: + # ... - Keyword Args: - session (:class:`aiohttp.ClientSession`) - An `aiohttp session`_ to use for requests to the API. + :param token: Your Top.gg API token. + :type token: :py:class:`str` + :param session: Whether to use an existing :class:`~aiohttp.ClientSession` for requesting. Defaults to :py:obj:`None` (creates a new one instead) + :type session: Optional[:class:`~aiohttp.ClientSession`] + + :exception TypeError: ``token`` is not a :py:class:`str` or is empty. + :exception ValueError: ``token`` is not a valid API token. """ - __slots__ = ("id", "_token", "_autopost", "__session", "__own_session", "__ratelimiter", "__ratelimiters", "__current_ratelimit") + id: int + """This project's ID.""" + + __slots__: tuple[str, ...] = ( + "__own_session", + "__session", + "__token", + "__ratelimiter", + "__ratelimiters", + "__current_ratelimit", + "_autopost", + "id", + ) def __init__( - self, - token: str, - *, - session: t.Optional[aiohttp.ClientSession] = None, - ) -> None: + self, token: str, *, session: Optional[ClientSession] = None, **kwargs + ): super().__init__() + if not isinstance(token, str) or not token: + raise TypeError("An API token is required to use this API.") + + if kwargs.pop("default_bot_id", None): + warnings.warn( + "The default bot ID is now derived from the Top.gg API token itself", + DeprecationWarning, + ) + + for key in kwargs.keys(): + warnings.warn(f"Ignored keyword argument: {key}", DeprecationWarning) + + self._autopost = None self.__own_session = session is None - self.__session = session or aiohttp.ClientSession( - timeout=aiohttp.ClientTimeout(total=MAXIMUM_DELAY_THRESHOLD * 1000.0) + self.__session = session or ClientSession( + timeout=ClientTimeout(total=MAXIMUM_DELAY_THRESHOLD * 1000.0) ) - self._token = token - self._autopost: t.Optional[AutoPoster] = None + self.__token = token try: - encoded_json = token.split('.')[1] - encoded_json += '=' * (4 - (len(encoded_json) % 4)) + encoded_json = token.split(".")[1] + encoded_json += "=" * (4 - (len(encoded_json) % 4)) encoded_json = json.loads(b64decode(encoded_json)) - - self.id = int(encoded_json['id']) + + self.id = int(encoded_json["id"]) except (IndexError, ValueError, binascii.Error, json.decoder.JSONDecodeError): - raise ValueError('Got a malformed API token.') from None + raise ValueError("Got a malformed API token.") from None - endpoint_ratelimits = namedtuple('EndpointRatelimits', 'global_ bot') + endpoint_ratelimits = namedtuple("EndpointRatelimits", "global_ bot") self.__ratelimiter = endpoint_ratelimits( global_=Ratelimiter(99, 1), bot=Ratelimiter(59, 60) @@ -93,15 +127,27 @@ def __init__( self.__ratelimiters = Ratelimiters(self.__ratelimiter) self.__current_ratelimit = None + def __repr__(self) -> str: + return f"<{__class__.__name__} {self.__session!r}>" + + def __int__(self) -> int: + return self.id + + @property + def is_closed(self) -> bool: + """Whether the client is closed.""" + + return self.__session.closed + async def __request( self, method: str, path: str, - params: t.Optional[dict] = None, - body: t.Optional[dict] = None, + params: Optional[dict] = None, + body: Optional[dict] = None, ) -> dict: if self.is_closed: - raise errors.ClientStateException('Client session is already closed.') + raise errors.ClientStateException("Client session is already closed.") if self.__current_ratelimit is not None: current_time = time() @@ -112,47 +158,49 @@ async def __request( self.__current_ratelimit = None ratelimiter = ( - self.__ratelimiters if path.startswith('/bots') else self.__ratelimiter.global_ + self.__ratelimiters + if path.startswith("/bots") + else self.__ratelimiter.global_ ) kwargs = {} if body: - kwargs['json'] = body + kwargs["json"] = body if params: - kwargs['params'] = params + kwargs["params"] = params - response = None + status = None retry_after = None output = None async with ratelimiter: try: - response = await self.__session.request( + async with self.__session.request( method, BASE_URL + path, headers={ - 'Authorization': f'Bearer {self.__token}', - 'Content-Type': 'application/json', - 'User-Agent': f'topggpy (https://github.com/top-gg-community/python-sdk {VERSION}) Python/', + "Authorization": f"Bearer {self.__token}", + "Content-Type": "application/json", + "User-Agent": f"topggpy (https://github.com/top-gg-community/python-sdk {VERSION}) Python/", }, **kwargs, - ) - - retry_after = float(response.headers.get('Retry-After', 0)) + ) as resp: + status = resp.status + retry_after = float(resp.headers.get("Retry-After", 0)) - if 'json' in response.headers['Content-Type']: - try: - output = await response.json() - except json.decoder.JSONDecodeError: - pass + if "json" in resp.headers["Content-Type"]: + try: + output = await resp.json() + except json.decoder.JSONDecodeError: + pass - response.raise_for_status() + resp.raise_for_status() - return output - except aiohttp.ClientResponseError: - if response.status == 429: + return output + except ClientResponseError: + if status == 429: if retry_after > MAXIMUM_DELAY_THRESHOLD: self.__current_ratelimit = time() + retry_after @@ -162,183 +210,323 @@ async def __request( return await self.__request(method, path) - raise errors.HTTPException(response, output) from None + raise errors.HTTPException( + output and output.get("message", output.get("detail")), status + ) from None - @property - def is_closed(self) -> bool: - return self.__session.closed + async def get_bot_info(self, id: Optional[int]) -> types.BotData: + """ + Fetches a Discord bot from its ID. - async def get_weekend_status(self) -> bool: - """Gets weekend status from Top.gg. + Example: + + .. code-block:: python + + bot = await client.get_bot_info(432610292342587392) + + :param id: The bot's ID. Defaults to your bot's ID. + :type id: Optional[:py:class:`int`] + + :exception ClientStateException: The client is already closed. + :exception HTTPException: Such query does not exist or the client has received other unfavorable responses from the API. + :exception Ratelimited: Ratelimited from sending more requests. + + :returns: The requested bot. + :rtype: :class:`.BotData` + """ + + return types.BotData(await self.__request("GET", f"/bots/{id or self.id}")) + + async def get_bots( + self, + limit: Optional[int] = None, + offset: Optional[int] = None, + sort: Optional[types.SortBy] = None, + *args, + **kwargs, + ) -> types.BotsData: + """ + Fetches Discord bots that matches the specified query. + + Examples: + + .. code-block:: python + + # With defaults + bots = await client.get_bots() + + # With explicit arguments + bots = await client.get_bots(limit=250, offset=50, sort=topgg.SortBy.MONTHLY_VOTES) + + for bot in bots: + print(bot) + + :param limit: The maximum amount of bots to be returned. + :type limit: Optional[:py:class:`int`] + :param offset: The amount of bots to be skipped. + :type offset: Optional[:py:class:`int`] + :param sort: The criteria to sort results by. Results will always be descending. + :type sort: Optional[:class:`.SortBy`] + + :exception ClientStateException: The client is already closed. + :exception HTTPException: Received an unfavorable response from the API. + :exception Ratelimited: Ratelimited from sending more requests. + + :returns: The requested bots. + :rtype: :class:`.BotsData` + """ + + params = {} + + if limit is not None: + params["limit"] = max(min(limit, 500), 1) + + if offset is not None: + params["offset"] = max(min(offset, 499), 0) + + if sort is not None: + if not isinstance(sort, types.SortBy): + if isinstance(sort, str) and sort in types.SortBy: + warnings.warn( + "The sort argument now expects a SortBy enum, not a str", + DeprecationWarning, + ) + + params["sort"] = sort + else: + raise TypeError( + f"Expected sort to be a SortBy enum, got {sort.__class__.__name__}." + ) + else: + params["sort"] = sort.value + + for arg in args: + warnings.warn(f"Ignored extra argument: {arg!r}", DeprecationWarning) - Returns: - :obj:`bool`: The boolean value of weekend status. + for key in kwargs.keys(): + warnings.warn(f"Ignored keyword argument: {key}", DeprecationWarning) - Raises: - :obj:`~.errors.ClientStateException` - If the client has been closed. + return types.BotsData(await self.__request("GET", "/bots", params=params)) + + async def get_guild_count(self) -> Optional[types.BotStatsData]: """ - - response = await self.__request('GET', '/weekend') + Fetches your Discord bot's posted statistics. + + Example: - return response['is_weekend'] + .. code-block:: python - @t.overload - async def post_guild_count(self, stats: types.StatsWrapper) -> None: - ... + stats = await client.get_guild_count() + server_count = stats.server_count + + :exception ClientStateException: The client is already closed. + :exception HTTPException: Received an unfavorable response from the API. + :exception Ratelimited: Ratelimited from sending more requests. + + :returns: The posted statistics. + :rtype: Optional[:py:class:`.BotStatsData`] + """ - @t.overload + stats = await self.__request("GET", "/bots/stats") + + return stats and types.BotStatsData(stats) + + @overload + async def post_guild_count(self, stats: types.StatsWrapper) -> None: ... + + @overload async def post_guild_count( self, *, - guild_count: t.Union[int, t.List[int]], - shard_count: t.Optional[int] = None, - shard_id: t.Optional[int] = None, - ) -> None: - ... + guild_count: Union[int, list[int]], + shard_count: Optional[int] = None, + shard_id: Optional[int] = None, + ) -> None: ... async def post_guild_count( - self, - stats: t.Any = None, - *, - guild_count: t.Any = None, + self, stats: Any = None, *, guild_count: Any = None, **kwargs ) -> None: - """Posts your bot's guild count and shards info to Top.gg. + """ + Updates the statistics in your Discord bot's Top.gg page. - .. _0 based indexing : https://en.wikipedia.org/wiki/Zero-based_numbering + Example: - Warning: - You can't provide both args and kwargs at once. + .. code-block:: python - Args: - stats (:obj:`~.types.StatsWrapper`) - An instance of StatsWrapper containing guild_count, shard_count, and shard_id. + await client.post_guild_count(topgg.StatsWrapper(bot.server_count)) - Keyword Arguments: - guild_count (:obj:`typing.Optional` [:obj:`typing.Union` [ :obj:`int`, :obj:`list` [ :obj:`int` ]]]) - Number of guilds the bot is in. Applies the number to a shard instead if shards are specified. - If not specified, length of provided client's property `.guilds` will be posted. + :param stats: The updated statistics. + :type stats: :class:`.StatsWrapper` + :param guild_count: The updated server count. + :type guild_count: Union[:py:class:`int`, list[:py:class:`int`]] - Raises: - TypeError - If no argument is provided. - :obj:`~.errors.ClientStateException` - If the client has been closed. + :exception ValueError: Got an invalid server count. + :exception ClientStateException: The client is already closed. + :exception HTTPException: Received an unfavorable response from the API. + :exception Ratelimited: Ratelimited from sending more requests. """ - if stats: - guild_count = stats.guild_count - elif guild_count is None: - raise TypeError("stats or guild_count must be provided.") - await self.__request('POST', '/bots/stats', body={'server_count': guild_count}) + for key in kwargs.keys(): + warnings.warn(f"Ignored keyword argument: {key}", DeprecationWarning) - async def get_guild_count(self) -> types.BotStatsData: - """Gets a bot's guild count and shard info from Top.gg. + if isinstance(stats, types.StatsWrapper): + guild_count = stats.server_count - Args: - bot_id (int) - ID of the bot you want to look up. Defaults to the provided Client object. + if not guild_count or guild_count <= 0: + raise ValueError(f"Got an invalid server count. Got {guild_count!r}.") - Returns: - :obj:`~.types.BotStatsData`: - The guild count and shards of a bot on Top.gg. + await self.__request("POST", "/bots/stats", body={"server_count": guild_count}) - Raises: - :obj:`~.errors.ClientStateException` - If the client has been closed. + async def get_weekend_status(self) -> bool: """ + Checks if the weekend multiplier is active, where a single vote counts as two. + + Example: + + .. code-block:: python + + is_weekend = await client.get_weekend_status() + + :exception ClientStateException: The client is already closed. + :exception HTTPException: Received an unfavorable response from the API. + :exception Ratelimited: Ratelimited from sending more requests. + + :returns: Whether the weekend multiplier is active. + :rtype: :py:class:`bool` + """ + + response = await self.__request("GET", "/weekend") + + return response["is_weekend"] + + async def get_bot_votes(self, page: int = 1) -> list[types.BriefUserData]: + """ + Fetches your project's recent 100 unique voters. + + Examples: - response = await self.__request('GET', '/bots/stats') + .. code-block:: python - return types.BotStatsData(**response) + # First page + voters = await client.get_bot_votes() - async def get_bot_votes(self) -> t.List[types.BriefUserData]: - """Gets information about last 1000 votes for your bot on Top.gg. + # Subsequent pages + voters = await client.get_bot_votes(2) - Note: - This API endpoint is only available to the bot's owner. + for voter in voters: + print(voter) - Returns: - :obj:`list` [ :obj:`~.types.BriefUserData` ]: - Users who voted for your bot. + :param page: The page number. Each page can only have at most 100 voters. Defaults to 1. + :type page: :py:class:`int` - Raises: - :obj:`~.errors.ClientStateException` - If the client has been closed. + :exception ClientStateException: The client is already closed. + :exception HTTPException: Received an unfavorable response from the API. + :exception Ratelimited: Ratelimited from sending more requests. + + :returns: The requested voters. + :rtype: list[:class:`.BriefUserData`] """ - response = await self.__request('GET', f'/bots/{self.id}/votes') - - return [types.BriefUserData(**user) for user in response] + return [ + types.BriefUserData(data) + for data in await self.__request( + "GET", f"/bots/{self.id}/votes", params={"page": max(page, 1)} + ) + ] + + async def get_user_info(self, user_id: int) -> types.UserData: + """ + Fetches a Top.gg user from their ID. - async def get_bot_info(self, bot_id: t.Optional[int] = None) -> types.BotData: - """This function is a coroutine. + .. deprecated:: 1.5.0 + No longer supported by API v0. - Gets information about a bot from Top.gg. + """ - Args: - bot_id (int) - ID of the bot to look up. Defaults to the provided Client object. + warnings.warn("get_user_info() is no longer supported", DeprecationWarning) - Returns: - :obj:`~.types.BotData`: - Information on the bot you looked up. Returned data can be found - `here `_. + raise errors.HTTPException("User not found", 404) - Raises: - :obj:`~.errors.ClientStateException` - If the client has been closed. + async def get_user_vote(self, id: int) -> bool: """ - bot_id = bot_id or self.id - response = await self.__request('GET', f'/bots/{bot_id}') - return types.BotData(**response) + Checks if a Top.gg user has voted for your project in the past 12 hours. - async def get_user_vote(self, user_id: int) -> bool: - """Gets information about a user's vote for your bot on Top.gg. + Example: - Args: - user_id (int) - ID of the user. + .. code-block:: python - Returns: - :obj:`bool`: Info about the user's vote. + has_voted = await client.get_user_vote(661200758510977084) - Raises: - :obj:`~.errors.ClientStateException` - If the client has been closed. + :param id: The user's ID. + :type id: :py:class:`int` + + :exception ClientStateException: The client is already closed. + :exception HTTPException: The specified user has not logged in to Top.gg or the client has received other unfavorable responses from the API. + :exception Ratelimited: Ratelimited from sending more requests. + + :returns: Whether the user has voted in the past 12 hours. + :rtype: :py:class:`bool` """ - data = await self.__request('GET', '/bots/check', params={'userId': user_id}) - - return bool(data["voted"]) + response = await self.__request("GET", "/bots/check", params={"userId": id}) + + return bool(response["voted"]) def autopost(self) -> AutoPoster: - """Returns a helper instance for auto-posting. + """ + Creates an autoposter instance that automatically updates the statistics in your Discord bot's Top.gg page every few minutes. + + Note that the after you call this method, subsequent calls will always return the same instance. + + .. code-block:: python + + autoposter = client.autopost() + + + @autoposter.stats + def get_stats() -> int: + return topgg.StatsWrapper(bot.server_count) - Note: - The second time you call this method, it'll return the same instance - as the one returned from the first call. - Returns: - :obj:`~.autopost.AutoPoster`: An instance of AutoPoster. + @autoposter.on_success + def success() -> None: + print("Successfully posted statistics to the Top.gg API!") + + + @autoposter.on_error + def error(exc: Exception) -> None: + print(f"Error: {exc!r}") + + + autoposter.start() + + :returns: The autoposter instance. + :rtype: :class:`.AutoPoster` """ + if self._autopost is not None: return self._autopost self._autopost = AutoPoster(self) + return self._autopost - + async def close(self) -> None: - """Closes all connections.""" + """ + Closes the client. + + Example: + + .. code-block:: python + + await client.close() + """ - if self._autopost: - self._autopost.cancel() - - if self.__own_session and not self.__session.closed: + if self.__own_session and not self.is_closed: await self.__session.close() async def __aenter__(self) -> "DBLClient": return self async def __aexit__(self, *_, **__) -> None: - await self.close() \ No newline at end of file + await self.close() diff --git a/topgg/data.py b/topgg/data.py index 7126d3bf..364cef11 100644 --- a/topgg/data.py +++ b/topgg/data.py @@ -1,31 +1,33 @@ -# The MIT License (MIT) - -# Copyright (c) 2021 Norizon - -# Permission is hereby granted, free of charge, to any person obtaining a -# copy of this software and associated documentation files (the "Software"), -# to deal in the Software without restriction, including without limitation -# the rights to use, copy, modify, merge, publish, distribute, sublicense, -# and/or sell copies of the Software, and to permit persons to whom the -# Software is furnished to do so, subject to the following conditions: - -# The above copyright notice and this permission notice shall be included in -# all copies or substantial portions of the Software. - -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -# DEALINGS IN THE SOFTWARE. - -__all__ = ["data", "DataContainerMixin"] +""" +The MIT License (MIT) + +Copyright (c) 2021 Norizon & Top.gg +Copyright (c) 2024-2025 null8626 & Top.gg + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. +""" import inspect import typing as t -from topgg.errors import TopGGException +from .errors import TopGGException + T = t.TypeVar("T") DataContainerT = t.TypeVar("DataContainerT", bound="DataContainerMixin") @@ -33,99 +35,107 @@ def data(type_: t.Type[T]) -> T: """ - Represents the injected data. This should be set as the parameter's default value. + The injected data. This should be set as the parameter's default value. - Args: - `type_` (:obj:`type` [ :obj:`T` ]) - The type of the injected data. + .. code-block:: python - Returns: - :obj:`T`: The injected data of type T. + client = topgg.DBLClient(os.getenv("BOT_TOKEN")).set_data(bot) + autoposter = client.autopost() - :Example: - .. code-block:: python - import topgg + @autoposter.stats() + def get_stats(bot: MyBot = topgg.data(MyBot)) -> topgg.StatsWrapper: + return topgg.StatsWrapper(bot.server_count) - # In this example, we fetch the stats from a Discord client instance. - client = Client(...) - dblclient = topgg.DBLClient(TOKEN).set_data(client) - autopost: topgg.AutoPoster = dblclient.autopost() + :param type_: The type of the injected data. + :type type_: Any - @autopost.stats() - def get_stats(client: Client = topgg.data(Client)): - return topgg.StatsWrapper(guild_count=len(client.guilds), shard_count=len(client.shards)) + :returns: The injected data. + :rtype: T """ + return t.cast(T, Data(type_)) class Data(t.Generic[T]): - __slots__ = ("type",) + __slots__: tuple[str, ...] = ("type",) def __init__(self, type_: t.Type[T]) -> None: - self.type: t.Type[T] = type_ + self.type = type_ class DataContainerMixin: """ - A class that holds data. + A data container. - This is useful for injecting some data so that they're available - as arguments in your functions. + This is useful for injecting some data so that they're available as arguments in your functions. """ - __slots__ = ("_data",) + __slots__: tuple[str, ...] = ("_data",) def __init__(self) -> None: - self._data: t.Dict[t.Type, t.Any] = {type(self): self} + self._data = {type(self): self} def set_data( self: DataContainerT, data_: t.Any, *, override: bool = False ) -> DataContainerT: """ - Sets data to be available in your functions. + Sets the data to be available in your functions. - Args: - `data_` (:obj:`typing.Any`) - The data to be injected. - override (:obj:`bool`) - Whether or not to override another instance that already exists. + :param data_: The data to be injected. + :type data_: Any + :param override: Whether to override another instance that already exists. Defaults to :py:obj:`False`. + :type override: :py:class:`bool` - Raises: - :obj:`~.errors.TopGGException` - If override is False and another instance of the same type exists. + :exception TopGGException: Override is :py:obj:`False` and another instance of the same type already exists. + + :returns: The object itself. + :rtype: :class:`.DataContainerMixin` """ + type_ = type(data_) + if not override and type_ in self._data: raise TopGGException( f"{type_} already exists. If you wish to override it, pass True into the override parameter." ) self._data[type_] = data_ + return self @t.overload - def get_data(self, type_: t.Type[T]) -> t.Optional[T]: - ... + def get_data(self, type_: t.Type[T]) -> t.Optional[T]: ... @t.overload - def get_data(self, type_: t.Type[T], default: t.Any = None) -> t.Any: - ... + def get_data(self, type_: t.Type[T], default: t.Any = None) -> t.Any: ... def get_data(self, type_: t.Any, default: t.Any = None) -> t.Any: - """Gets the injected data.""" + """ + Gets the injected data. + + :param type_: The type of the injected data. + :type type_: Any + :param default: The default value in case the injected data does not exist. Defaults to :py:obj:`None`. + :type default: Any + + :returns: The injected data. + :rtype: Any + """ + return self._data.get(type_, default) async def _invoke_callback( self, callback: t.Callable[..., T], *args: t.Any, **kwargs: t.Any ) -> T: parameters: t.Mapping[str, inspect.Parameter] + try: parameters = inspect.signature(callback).parameters except (ValueError, TypeError): parameters = {} - signatures: t.Dict[str, Data] = { + signatures: dict[str, Data] = { k: v.default for k, v in parameters.items() if v.kind is inspect.Parameter.POSITIONAL_OR_KEYWORD @@ -136,6 +146,7 @@ async def _invoke_callback( signatures[k] = self._resolve_data(v.type) res = callback(*args, **{**signatures, **kwargs}) + if inspect.isawaitable(res): return await res diff --git a/topgg/errors.py b/topgg/errors.py index d110126e..88512e96 100644 --- a/topgg/errors.py +++ b/topgg/errors.py @@ -23,77 +23,63 @@ SOFTWARE. """ -__all__ = [ - "TopGGException", - "ClientException", - "ClientStateException", - "Ratelimited", - "HTTPException", -] - -from typing import TYPE_CHECKING, Union - -if TYPE_CHECKING: - from aiohttp import ClientResponse +from typing import Optional class TopGGException(Exception): - """Base exception class for topggpy. + """An error coming from this SDK. Extends :py:class:`Exception`.""" - Ideally speaking, this could be caught to handle any exceptions thrown from this library. - """ + __slots__: tuple[str, ...] = () class ClientException(TopGGException): - """Exception that's thrown when an operation in the :class:`~.DBLClient` fails. + """An operation failure in :class:`.DBLClient`. Extends :class:`.TopGGException`.""" - These are usually for exceptions that happened due to user input. - """ + __slots__: tuple[str, ...] = () class ClientStateException(ClientException): - """Exception that's thrown when an operation happens in a closed :obj:`~.DBLClient` instance.""" + """Attempted operation in a closed :class:`.DBLClient` instance. Extends :class:`.ClientException`.""" - -class Ratelimited(TopGGException): - """Exception that's thrown when the client is ratelimited.""" - - __slots__: tuple[str, ...] = ('retry_after',) - - retry_after: float - """How long the client should wait in seconds before it could send requests again without receiving a 429.""" - - def __init__(self, retry_after: float): - self.retry_after = retry_after - - super().__init__( - f'Blocked from sending more requests, try again in {retry_after} seconds.' - ) + __slots__: tuple[str, ...] = () class HTTPException(TopGGException): - """Exception that's thrown when an HTTP request operation fails. + """HTTP request failure. Extends :class:`.TopGGException`.""" + + __slots__: tuple[str, ...] = ("message", "code") + + message: Optional[str] + """The message returned from the API.""" - Attributes: - response (:class:`aiohttp.ClientResponse`) - The response of the failed HTTP request. - text (str) - The text of the error. Could be an empty string. - code (int) - The response status code. - """ + code: Optional[int] + """The status code returned from the API.""" - __slots__ = ("response", "text", "code") + def __init__(self, message: Optional[str], code: Optional[int]): + self.message = message + self.code = code - def __init__(self, response: "ClientResponse", message: Union[dict, str]) -> None: - self.response = response - self.code = response.status - self.text = message.get("message", message.get("detail", "")) if isinstance(message, dict) else message + super().__init__(f"Got {code}: {message!r}") - fmt = f"{self.response.reason} (status code: {self.response.status})" + def __repr__(self) -> str: + return f"<{__class__.__name__} message={self.message!r} code={self.code}>" - if self.text: - fmt = f"{fmt}: {self.text}" - super().__init__(fmt) +class Ratelimited(HTTPException): + """Ratelimited from sending more requests. Extends :class:`.HTTPException`.""" + + __slots__: tuple[str, ...] = ("retry_after",) + + retry_after: float + """How long the client should wait in seconds before it could send requests again without receiving a 429.""" + + def __init__(self, retry_after: float): + super().__init__( + f"Blocked from sending more requests, try again in {retry_after} seconds.", + 429, + ) + + self.retry_after = retry_after + def __repr__(self) -> str: + return f"<{__class__.__name__} retry_after={self.retry_after}>" diff --git a/topgg/ratelimiter.py b/topgg/ratelimiter.py index 87346418..ec213e92 100644 --- a/topgg/ratelimiter.py +++ b/topgg/ratelimiter.py @@ -24,20 +24,19 @@ """ from collections.abc import Iterable -from types import TracebackType from collections import deque from time import time import asyncio +import typing + +if typing.TYPE_CHECKING: + from types import TracebackType -__all__ = [ - "Ratelimiter", - "Ratelimiters", -] class Ratelimiter: """Handles ratelimits for a specific endpoint.""" - __slots__: tuple[str, ...] = ('__lock', '__max_calls', '__period', '__calls') + __slots__: tuple[str, ...] = ("__lock", "__max_calls", "__period", "__calls") def __init__( self, @@ -49,7 +48,7 @@ def __init__( self.__max_calls = max_calls self.__lock = asyncio.Lock() - async def __aenter__(self) -> 'Ratelimiter': + async def __aenter__(self) -> "Ratelimiter": """Delays the request to this endpoint if it could lead to a ratelimit.""" async with self.__lock: @@ -65,7 +64,7 @@ async def __aexit__( self, _exc_type: type[BaseException], _exc_val: BaseException, - _exc_tb: TracebackType, + _exc_tb: "TracebackType", ) -> None: """Stores the previous request's timestamp.""" @@ -85,12 +84,12 @@ def _timespan(self) -> float: class Ratelimiters: """Handles ratelimits for multiple endpoints.""" - __slots__: tuple[str, ...] = ('__ratelimiters',) + __slots__: tuple[str, ...] = ("__ratelimiters",) def __init__(self, ratelimiters: Iterable[Ratelimiter]): self.__ratelimiters = ratelimiters - async def __aenter__(self) -> 'Ratelimiters': + async def __aenter__(self) -> "Ratelimiters": """Delays the request to this endpoint if it could lead to a ratelimit.""" for ratelimiter in self.__ratelimiters: @@ -102,7 +101,7 @@ async def __aexit__( self, exc_type: type[BaseException], exc_val: BaseException, - exc_tb: TracebackType, + exc_tb: "TracebackType", ) -> None: """Stores the previous request's timestamp.""" @@ -111,4 +110,4 @@ async def __aexit__( ratelimiter.__aexit__(exc_type, exc_val, exc_tb) for ratelimiter in self.__ratelimiters ) - ) \ No newline at end of file + ) diff --git a/topgg/types.py b/topgg/types.py index 2da13f95..c211d27e 100644 --- a/topgg/types.py +++ b/topgg/types.py @@ -1,385 +1,493 @@ -# -*- coding: utf-8 -*- - -# The MIT License (MIT) - -# Copyright (c) 2021 Assanali Mukhanov - -# Permission is hereby granted, free of charge, to any person obtaining a -# copy of this software and associated documentation files (the "Software"), -# to deal in the Software without restriction, including without limitation -# the rights to use, copy, modify, merge, publish, distribute, sublicense, -# and/or sell copies of the Software, and to permit persons to whom the -# Software is furnished to do so, subject to the following conditions: - -# The above copyright notice and this permission notice shall be included in -# all copies or substantial portions of the Software. - -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -# DEALINGS IN THE SOFTWARE. - -__all__ = ["WidgetOptions", "StatsWrapper"] +""" +The MIT License (MIT) + +Copyright (c) 2021 Assanali Mukhanov & Top.gg +Copyright (c) 2024-2025 null8626 & Top.gg + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. +""" import dataclasses import typing as t -from datetime import datetime - -KT = t.TypeVar("KT") -VT = t.TypeVar("VT") -Colors = t.Dict[str, int] -Colours = Colors - - -def camel_to_snake(string: str) -> str: - return "".join(["_" + c.lower() if c.isupper() else c for c in string]).lstrip("_") - - -def parse_vote_dict(d: dict) -> dict: - data = d.copy() - - query = data.get("query", "").lstrip("?") - if query: - query_dict = {k: v for k, v in [pair.split("=") for pair in query.split("&")]} - data["query"] = DataDict(**query_dict) - else: - data["query"] = {} - - if "bot" in data: - data["bot"] = int(data["bot"]) - - elif "guild" in data: - data["guild"] = int(data["guild"]) - - for key, value in data.copy().items(): - converted_key = camel_to_snake(key) - if key != converted_key: - del data[key] - data[converted_key] = value - - return data - - -def parse_dict(d: dict) -> dict: - data = d.copy() - - for key, value in data.copy().items(): - if "id" in key.lower(): - if value == "": - value = None - else: - if isinstance(value, str) and value.isdigit(): - value = int(value) - else: - continue - elif value == "": - value = None - - converted_key = camel_to_snake(key) - if key != converted_key: - del data[key] - data[converted_key] = value - - return data - +import warnings -def parse_bot_dict(d: dict) -> dict: - data = parse_dict(d.copy()) - - if data.get("date") and not isinstance(data["date"], datetime): - data["date"] = datetime.strptime(data["date"], "%Y-%m-%dT%H:%M:%S.%fZ") +from urllib.parse import parse_qs +from datetime import datetime +from enum import Enum - if data.get("owners"): - data["owners"] = [int(e) for e in data["owners"]] - if data.get("guilds"): - data["guilds"] = [int(e) for e in data["guilds"]] - for key, value in data.copy().items(): - converted_key = camel_to_snake(key) - if key != converted_key: - del data[key] - data[converted_key] = value +T = t.TypeVar("T") - return data +def truthy_only(value: t.Optional[T]) -> t.Optional[T]: + if value: + return value -def parse_user_dict(d: dict) -> dict: - data = d.copy() - data["social"] = SocialData(**data.get("social", {})) +class WidgetProjectType(Enum): + """A Top.gg widget's project type.""" - return data + __slots__: tuple[str, ...] = () + DISCORD_BOT = "discord/bot" + DISCORD_SERVER = "discord/server" -def parse_bot_stats_dict(d: dict) -> dict: - data = d.copy() - if "server_count" not in data: - data["server_count"] = None - if "shards" not in data: - data["shards"] = [] - if "shard_count" not in data: - data["shard_count"] = None +class WidgetType(Enum): + """A Top.gg widget's type.""" - return data + __slots__: tuple[str, ...] = () + LARGE = "large" + VOTES = "votes" + OWNER = "owner" + SOCIAL = "social" -class DataDict(dict, t.MutableMapping[KT, VT]): - """Base class used to represent received data from the API. - Every data model subclasses this class. - """ +class WidgetOptions: + """Top.gg widget creation options.""" - def __init__(self, **kwargs: VT) -> None: - super().__init__(**parse_dict(kwargs)) - self.__dict__ = self + __slots__: tuple[str, ...] = ("id", "project_type", "type") + id: int + """This widget's project ID.""" -class WidgetOptions(DataDict[str, t.Any]): - """Model that represents widget options that are passed to Top.gg widget URL generated via - :meth:`DBLClient.generate_widget`.""" + project_type: WidgetProjectType + """This widget's project type.""" - id: t.Optional[int] - """ID of a bot to generate the widget for. Must resolve to an ID of a listed bot when converted to a string.""" - colors: Colors - """A dictionary consisting of a parameter as a key and HEX color (type `int`) as value. ``color`` will be - appended to the key in case it doesn't end with ``color``.""" - noavatar: bool - """Indicates whether to exclude the bot's avatar from short widgets. Must be of type ``bool``. Defaults to - ``False``.""" - format: str - """Format to apply to the widget. Must be either ``png`` and ``svg``. Defaults to ``png``.""" - type: str - """Type of a short widget (``status``, ``servers``, ``upvotes``, and ``owner``). For large widget, - must be an empty string.""" + type: WidgetType + """This widget's type.""" def __init__( self, - id: t.Optional[int] = None, - format: t.Optional[str] = None, - type: t.Optional[str] = None, - noavatar: bool = False, - colors: t.Optional[Colors] = None, - colours: t.Optional[Colors] = None, + id: int, + project_type: WidgetProjectType, + type: WidgetType, + *args, + **kwargs, ): - super().__init__( - id=id or None, - format=format or "png", - type=type or "", - noavatar=noavatar or False, - colors=colors or colours or {}, - ) - - @property - def colours(self) -> Colors: - return self.colors - - @colours.setter - def colours(self, value: Colors) -> None: - self.colors = value - - def __setitem__(self, key: str, value: t.Any) -> None: - if key == "colours": - key = "colors" - super().__setitem__(key, value) - - def __getitem__(self, item: str) -> t.Any: - if item == "colours": - item = "colors" - return super().__getitem__(item) - - def get(self, key: str, default: t.Any = None) -> t.Any: - """:meta private:""" - if key == "colours": - key = "colors" - return super().get(key, default) - - -class BotData(DataDict[str, t.Any]): - """Model that contains information about a listed bot on top.gg. The data this model contains can be found `here - `__.""" + self.id = id + self.project_type = project_type + self.type = type + + for arg in args: + warnings.warn(f"Ignored extra argument: {arg!r}", DeprecationWarning) + + for key in kwargs.keys(): + warnings.warn(f"Ignored keyword argument: {key}", DeprecationWarning) + + def __repr__(self) -> str: + return f"<{__class__.__name__} id={self.id} project_type={self.project_type!r} type={self.type!r}>" + + +class BotData: + """A Discord bot listed on Top.gg.""" + + __slots__: tuple[str, ...] = ( + "id", + "topgg_id", + "username", + "discriminator", + "avatar", + "def_avatar", + "prefix", + "shortdesc", + "longdesc", + "tags", + "website", + "support", + "github", + "owners", + "guilds", + "invite", + "date", + "certified_bot", + "vanity", + "points", + "monthly_points", + "donatebotguildid", + "server_count", + "review_score", + "review_count", + ) id: int - """The ID of the bot.""" + """This bot's Discord ID.""" + + topgg_id: int + """This bot's Top.gg ID.""" username: str - """The username of the bot.""" + """This bot's username.""" discriminator: str - """The discriminator of the bot.""" + """This bot's discriminator.""" - avatar: t.Optional[str] - """The avatar hash of the bot.""" + avatar: str + """This bot's avatar URL.""" def_avatar: str - """The avatar hash of the bot's default avatar.""" + """This bot's default avatar hash.""" prefix: str - """The prefix of the bot.""" + """This bot's prefix.""" shortdesc: str - """The brief description of the bot.""" + """This bot's short description.""" longdesc: t.Optional[str] - """The long description of the bot.""" + """This bot's HTML/Markdown long description.""" - tags: t.List[str] - """The tags the bot has.""" + tags: list[str] + """This bot's tags.""" website: t.Optional[str] - """The website of the bot.""" + """This bot's website URL.""" support: t.Optional[str] - """The invite code of the bot's support server.""" + """This bot's support URL.""" github: t.Optional[str] - """The GitHub URL of the repo of the bot.""" + """This bot's GitHub repository URL.""" - owners: t.List[int] - """The IDs of the owners of the bot.""" + owners: list[int] + """This bot's owner IDs.""" - guilds: t.List[int] - """The guilds the bot is in.""" + guilds: list[int] + """This bot's list of servers.""" invite: t.Optional[str] - """The invite URL of the bot.""" + """This bot's invite URL.""" date: datetime - """The time the bot was added.""" + """This bot's submission date.""" certified_bot: bool - """Whether or not the bot is certified.""" + """Whether this bot is certified.""" vanity: t.Optional[str] - """The vanity URL of the bot.""" + """This bot's Top.gg vanity code.""" points: int - """The amount of the votes the bot has.""" + """The amount of votes this bot has.""" monthly_points: int - """The amount of the votes the bot has this month.""" + """The amount of votes this bot has this month.""" donatebotguildid: int - """The guild ID for the donatebot setup.""" + """This bot's donatebot setup server ID.""" + + server_count: t.Optional[int] + """This bot's posted server count.""" + + review_score: float + """This bot's average review score out of 5.""" + + review_count: int + """This bot's review count.""" + + def __init__(self, json: dict): + self.id = int(json["clientid"]) + self.topgg_id = int(json["id"]) + self.username = json["username"] + self.discriminator = "0" + self.avatar = json["avatar"] + self.def_avatar = "" + self.prefix = json["prefix"] + self.shortdesc = json["shortdesc"] + self.longdesc = truthy_only(json.get("longdesc")) + self.tags = json["tags"] + self.website = truthy_only(json.get("website")) + self.support = truthy_only(json.get("support")) + self.github = truthy_only(json.get("github")) + self.owners = [int(id) for id in json["owners"]] + self.guilds = [] + self.invite = truthy_only(json.get("invite")) + self.date = datetime.fromisoformat(json["date"].replace("Z", "+00:00")) + self.certified_bot = False + self.vanity = truthy_only(json.get("vanity")) + self.points = json["points"] + self.monthly_points = json["monthlyPoints"] + self.donatebotguildid = 0 + self.server_count = json.get("server_count") + self.review_score = json["reviews"]["averageScore"] + self.review_count = json["reviews"]["count"] + + def __repr__(self) -> str: + return f"<{__class__.__name__} id={self.id} username={self.username!r} points={self.points} monthly_points={self.monthly_points} server_count={self.server_count}>" + + def __int__(self) -> int: + return self.id + + def __eq__(self, other: "BotData") -> bool: + if isinstance(other, __class__): + return self.id == other.id + + return NotImplemented + + +class BotsData: + """A list of Discord bot's listed on Top.gg.""" + + __slots__: tuple[str, ...] = ("results", "limit", "offset", "count", "total") + + results: list[BotData] + """The list of bots returned.""" + + limit: int + """The maximum amount of bots returned.""" + + offset: int + """The amount of bots skipped.""" + + count: int + """The amount of bots returned. Akin to len(results).""" + + total: int + """The amount of bots that matches the specified query. May be equal or greater than count or len(results).""" - def __init__(self, **kwargs: t.Any): - super().__init__(**parse_bot_dict(kwargs)) + def __init__(self, json: dict): + self.results = [BotData(bot) for bot in json["results"]] + self.limit = json["limit"] + self.offset = json["offset"] + self.count = json["count"] + self.total = json["total"] + def __repr__(self) -> str: + return f"<{__class__.__name__} results={self.results!r} count={self.count} total={self.total}>" -class BotStatsData(DataDict[str, t.Any]): - """Model that contains information about a listed bot's guild and shard count.""" + def __iter__(self) -> t.Iterable[BotData]: + return iter(self.results) + + def __len__(self) -> int: + return self.count + + +class BotStatsData: + """A Discord bot's statistics.""" + + __slots__: tuple[str, ...] = ("server_count", "shards", "shard_count") server_count: t.Optional[int] """The amount of servers the bot is in.""" - shards: t.List[int] + + shards: list[int] """The amount of servers the bot is in per shard.""" + shard_count: t.Optional[int] - """The amount of shards a bot has.""" + """The amount of shards the bot has.""" - def __init__(self, **kwargs: t.Any): - super().__init__(**parse_bot_stats_dict(kwargs)) + def __init__(self, json: dict): + self.server_count = json.get("server_count") + self.shards = [] + self.shard_count = None + def __repr__(self) -> str: + return f"<{__class__.__name__} server_count={self.server_count}>" -class BriefUserData(DataDict[str, t.Any]): - """Model that contains brief information about a Top.gg user.""" + def __int__(self) -> int: + return self.server_count + + def __eq__(self, other: "BotStatsData") -> bool: + if isinstance(other, __class__): + return self.server_count == other.server_count + + return NotImplemented + + +class BriefUserData: + """A Top.gg user's brief information.""" + + __slots__: tuple[str, ...] = ("id", "username", "avatar") id: int - """The Discord ID of the user.""" + """This user's ID.""" + username: str - """The Discord username of the user.""" + """This user's username.""" + avatar: str - """The Discord avatar URL of the user.""" + """This user's avatar URL.""" + + def __init__(self, json: dict): + self.id = int(json["id"]) + self.username = json["username"] + self.avatar = json["avatar"] + + def __repr__(self) -> str: + return f"<{__class__.__name__} id={self.id} username={self.username!r}>" + + def __int__(self) -> int: + return self.id + + def __eq__(self, other: "BriefUserData") -> bool: + if isinstance(other, __class__): + return self.id == other.id + + return NotImplemented - def __init__(self, **kwargs: t.Any): - if kwargs["id"].isdigit(): - kwargs["id"] = int(kwargs["id"]) - super().__init__(**kwargs) +class SocialData: + """A Top.gg user's socials.""" -class SocialData(DataDict[str, str]): - """Model that contains social information about a top.gg user.""" + __slots__: tuple[str, ...] = ("youtube", "reddit", "twitter", "instagram", "github") youtube: str - """The YouTube channel ID of the user.""" + """This user's YouTube channel.""" + reddit: str - """The Reddit username of the user.""" + """This user's Reddit username.""" + twitter: str - """The Twitter username of the user.""" + """This user's Twitter username.""" + instagram: str - """The Instagram username of the user.""" + """This user's Instagram username.""" + github: str - """The GitHub username of the user.""" + """This user's GitHub username.""" + +class UserData: + """A Top.gg user.""" -class UserData(DataDict[str, t.Any]): - """Model that contains information about a top.gg user. The data this model contains can be found `here - `__.""" + __slots__: tuple[str, ...] = ( + "id", + "username", + "discriminator", + "social", + "color", + "supporter", + "certified_dev", + "mod", + "web_mod", + "admin", + ) id: int - """The ID of the user.""" + """This user's ID.""" username: str - """The username of the user.""" + """This user's username.""" discriminator: str - """The discriminator of the user.""" + """This user's discriminator.""" social: SocialData - """The social data of the user.""" + """This user's social links.""" color: str - """The custom hex color of the user.""" + """This user's profile color.""" supporter: bool - """Whether or not the user is a supporter.""" + """Whether this user is a Top.gg supporter.""" certified_dev: bool - """Whether or not the user is a certified dev.""" + """Whether this user is a Top.gg certified developer.""" mod: bool - """Whether or not the user is a Top.gg mod.""" + """Whether this user is a Top.gg moderator.""" web_mod: bool - """Whether or not the user is a Top.gg web mod.""" + """Whether this user is a Top.gg website moderator.""" admin: bool - """Whether or not the user is a Top.gg admin.""" + """Whether this user is a Top.gg website administrator.""" - def __init__(self, **kwargs: t.Any): - super().__init__(**parse_user_dict(kwargs)) +class SortBy(Enum): + """Supported sorting criterias in :meth:`.DBLClient.get_bots`.""" -class VoteDataDict(DataDict[str, t.Any]): - """Base model that represents received information from Top.gg via webhooks.""" + __slots__: tuple[str, ...] = () - type: str - """Type of the action (``upvote`` or ``test``).""" - user: int - """ID of the voter.""" - query: DataDict - """Query parameters in :obj:`~.DataDict`.""" + ID = "id" + """Sorts results based on each bot's ID.""" - def __init__(self, **kwargs: t.Any): - super().__init__(**parse_vote_dict(kwargs)) + SUBMISSION_DATE = "date" + """Sorts results based on each bot's submission date.""" + + MONTHLY_VOTES = "monthlyPoints" + """Sorts results based on each bot's monthly vote count.""" + + +class VoteDataDict: + """A dispatched Top.gg project vote event.""" + + __slots__: tuple[str, ...] = ("type", "user", "query") + + type: t.Optional[str] + """Vote event type. ``upvote`` (invoked from the vote page by a user) or ``test`` (invoked explicitly by the developer for testing.)""" + + user: t.Optional[int] + """The ID of the user who voted.""" + + query: dict + """Query strings found on the vote page.""" + + def __init__(self, json: dict): + self.type = json.get("type") + + user = json.get("user") + self.user = user and int(user) + + self.query = parse_qs(json.get("query", "")) + + def __repr__(self) -> str: + return f"<{__class__.__name__} type={self.type!r} user={self.user} query={self.query!r}>" class BotVoteData(VoteDataDict): - """Model that contains information about a bot vote.""" + """A dispatched Top.gg Discord bot vote event. Extends :class:`.VoteDataDict`.""" + + __slots__: tuple[str, ...] = ("bot", "is_weekend") + + bot: t.Optional[int] + """The ID of the bot that received a vote.""" - bot: int - """ID of the bot the user voted for.""" is_weekend: bool - """Boolean value indicating whether the action was done on a weekend.""" + """Whether the weekend multiplier is active or not, meaning a single vote counts as two.""" + + def __init__(self, json: dict): + super().__init__(json) + + bot = json.get("bot") + self.bot = bot and int(bot) + + self.is_weekend = json.get("isWeekend", False) + + def __repr__(self) -> str: + return f"<{__class__.__name__} type={self.type!r} user={self.user} is_weekend={self.is_weekend}>" class GuildVoteData(VoteDataDict): - """Model that contains information about a guild vote.""" + """ "A dispatched Top.gg Discord server vote event. Extends :class:`.VoteDataDict`.""" - guild: int - """ID of the guild the user voted for.""" + __slots__: tuple[str, ...] = ("guild",) + + guild: t.Optional[int] + """The ID of the server that received a vote.""" + + def __init__(self, json: dict): + super().__init__(json) + + guild = json.get("guild") + self.guild = guild and int(guild) ServerVoteData = GuildVoteData @@ -387,11 +495,11 @@ class GuildVoteData(VoteDataDict): @dataclasses.dataclass class StatsWrapper: - guild_count: int - """The guild count.""" + server_count: t.Optional[int] + """The amount of servers the bot is in.""" shard_count: t.Optional[int] = None - """The shard count.""" + """The amount of shards the bot has.""" shard_id: t.Optional[int] = None """The shard ID the guild count belongs to.""" diff --git a/topgg/version.py b/topgg/version.py index 91c27a95..177b9352 100644 --- a/topgg/version.py +++ b/topgg/version.py @@ -1 +1 @@ -VERSION = "3.0.0" \ No newline at end of file +VERSION = "1.5.0" diff --git a/topgg/webhook.py b/topgg/webhook.py index 4b94ec2b..6e4cb16a 100644 --- a/topgg/webhook.py +++ b/topgg/webhook.py @@ -1,43 +1,33 @@ -# -*- coding: utf-8 -*- - -# The MIT License (MIT) - -# Copyright (c) 2021 Assanali Mukhanov - -# Permission is hereby granted, free of charge, to any person obtaining a -# copy of this software and associated documentation files (the "Software"), -# to deal in the Software without restriction, including without limitation -# the rights to use, copy, modify, merge, publish, distribute, sublicense, -# and/or sell copies of the Software, and to permit persons to whom the -# Software is furnished to do so, subject to the following conditions: - -# The above copyright notice and this permission notice shall be included in -# all copies or substantial portions of the Software. - -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS -# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -# DEALINGS IN THE SOFTWARE. - -__all__ = [ - "endpoint", - "BoundWebhookEndpoint", - "WebhookEndpoint", - "WebhookManager", - "WebhookType", -] +""" +The MIT License (MIT) + +Copyright (c) 2021 Assanali Mukhanov & Top.gg +Copyright (c) 2024-2025 null8626 & Top.gg + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. +""" import enum import typing as t - -import aiohttp from aiohttp import web -from topgg.errors import TopGGException - +from .errors import TopGGException from .data import DataContainerMixin from .types import BotVoteData, GuildVoteData @@ -49,54 +39,52 @@ class WebhookType(enum.Enum): - """An enum that represents the type of an endpoint.""" + """Marks the type of a webhook endpoint.""" + + __slots__: tuple[str, ...] = () BOT = enum.auto() - """Marks the endpoint as a bot webhook.""" + """Marks the endpoint as a Discord bot webhook.""" GUILD = enum.auto() - """Marks the endpoint as a guild webhook.""" + """Marks the endpoint as a Discord server webhook.""" class WebhookManager(DataContainerMixin): - """ - A class for managing Top.gg webhooks. - """ + """A Top.gg webhook manager.""" - __app: web.Application - _webserver: web.TCPSite - _is_closed: bool - __slots__ = ("__app", "_webserver", "_is_running") + __slots__: tuple[str, ...] = ("__app", "_webserver", "_is_running") def __init__(self) -> None: super().__init__() + self.__app = web.Application() self._is_running = False + def __repr__(self) -> str: + return f"<{__class__.__name__} is_running={self.is_running}>" + @t.overload - def endpoint(self, endpoint_: None = None) -> "BoundWebhookEndpoint": - ... + def endpoint(self, endpoint_: None = None) -> "BoundWebhookEndpoint": ... @t.overload - def endpoint(self, endpoint_: "WebhookEndpoint") -> "WebhookManager": - ... + def endpoint(self, endpoint_: "WebhookEndpoint") -> "WebhookManager": ... - def endpoint(self, endpoint_: t.Optional["WebhookEndpoint"] = None) -> t.Any: - """Helper method that returns a WebhookEndpoint object. + def endpoint( + self, endpoint_: t.Optional["WebhookEndpoint"] = None + ) -> t.Union["WebhookManager", "BoundWebhookEndpoint"]: + """ + A helper method that returns a :class:`.WebhookEndpoint` object. - Args: - `endpoint_` (:obj:`typing.Optional` [ :obj:`WebhookEndpoint` ]) - The endpoint to add. + :param endpoint_: The endpoint to add. + :type endpoint_: Optional[:class:`.WebhookEndpoint`] - Returns: - :obj:`typing.Union` [ :obj:`WebhookManager`, :obj:`BoundWebhookEndpoint` ]: - An instance of :obj:`WebhookManager` if endpoint was provided, - otherwise :obj:`BoundWebhookEndpoint`. + :exception TopGGException: If the endpoint is not :py:obj:`None` and is not an instance of :class:`.WebhookEndpoint`. - Raises: - :obj:`~.errors.TopGGException` - If the endpoint is lacking attributes. + :returns: An instance of :class:`.WebhookManager` if an endpoint was provided, otherwise :class:`.BoundWebhookEndpoint`. + :rtype: Union[:class:`.WebhookManager`, :class:`.BoundWebhookEndpoint`] """ + if endpoint_: if not hasattr(endpoint_, "_callback"): raise TopGGException("endpoint missing callback.") @@ -113,56 +101,60 @@ def endpoint(self, endpoint_: t.Optional["WebhookEndpoint"] = None) -> t.Any: endpoint_._type, endpoint_._auth, endpoint_._callback ), ) + return self - return BoundWebhookEndpoint(manager=self) + return BoundWebhookEndpoint(self) async def start(self, port: int) -> None: - """Runs the webhook. + """ + Runs the webhook. - Args: - port (int) - The port to run the webhook on. + :param port: The port to use. + :type port: :py:class:`int` """ + runner = web.AppRunner(self.__app) await runner.setup() + self._webserver = web.TCPSite(runner, "0.0.0.0", port) await self._webserver.start() + self._is_running = True @property def is_running(self) -> bool: - """Returns whether or not the webserver is running.""" + """Whether the webserver is running.""" + return self._is_running @property def app(self) -> web.Application: - """Returns the internal web application that handles webhook requests. + """The internal :class:`~aiohttp.web.Application` that handles web requests.""" - Returns: - :class:`aiohttp.web.Application`: - The internal web application. - """ return self.__app async def close(self) -> None: """Stops the webhook.""" + await self._webserver.stop() self._is_running = False def _get_handler( self, type_: WebhookType, auth: str, callback: t.Callable[..., t.Any] ) -> _HandlerT: - async def _handler(request: aiohttp.web.Request) -> web.Response: + async def _handler(request: web.Request) -> web.Response: if request.headers.get("Authorization", "") != auth: return web.Response(status=401, text="Unauthorized") data = await request.json() + await self._invoke_callback( callback, - (BotVoteData if type_ is WebhookType.BOT else GuildVoteData)(**data), + (BotVoteData if type_ is WebhookType.BOT else GuildVoteData)(data), ) - return web.Response(status=200, text="OK") + + return web.Response(status=204, text="") return _handler @@ -171,11 +163,9 @@ async def _handler(request: aiohttp.web.Request) -> web.Response: class WebhookEndpoint: - """ - A helper class to setup webhook endpoint. - """ + """A helper class to setup a Top.gg webhook endpoint.""" - __slots__ = ("_callback", "_auth", "_route", "_type") + __slots__: tuple[str, ...] = ("_callback", "_auth", "_route", "_type") def __init__(self) -> None: self._auth = "" @@ -183,92 +173,105 @@ def __init__(self) -> None: def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Any: return self._callback(*args, **kwargs) + def __repr__(self) -> str: + return f"<{__class__.__name__} type={self._type!r} route={self._route!r}>" + def type(self: T, type_: WebhookType) -> T: - """Sets the type of this endpoint. + """ + Sets the type of this endpoint. - Args: - `type_` (:obj:`WebhookType`) - The type of the endpoint. + :param type_: The endpoint's new type. + :type type_: :class:`.WebhookType` - Returns: - :obj:`WebhookEndpoint` + :returns: The object itself. + :rtype: :class:`.WebhookEndpoint` """ + self._type = type_ + return self def route(self: T, route_: str) -> T: """ Sets the route of this endpoint. - Args: - `route_` (str) - The route of this endpoint. + :param route_: The endpoint's new route. + :type route_: :py:class:`str` - Returns: - :obj:`WebhookEndpoint` + :returns: The object itself. + :rtype: :class:`.WebhookEndpoint` """ + self._route = route_ + return self def auth(self: T, auth_: str) -> T: """ - Sets the auth of this endpoint. + Sets the password of this endpoint. - Args: - `auth_` (str) - The auth of this endpoint. + :param auth_: The endpoint's new password. + :type auth_: :py:class:`str` - Returns: - :obj:`WebhookEndpoint` + :returns: The object itself. + :rtype: :class:`.WebhookEndpoint` """ + self._auth = auth_ + return self @t.overload - def callback(self, callback_: None) -> t.Callable[[CallbackT], CallbackT]: - ... + def callback(self, callback_: None) -> t.Callable[[CallbackT], CallbackT]: ... @t.overload - def callback(self: T, callback_: CallbackT) -> T: - ... + def callback(self: T, callback_: CallbackT) -> T: ... - def callback(self, callback_: t.Any = None) -> t.Any: + def callback(self, callback_: t.Any = None) -> t.Union[t.Any, "WebhookEndpoint"]: """ - Registers a vote callback, called whenever this endpoint receives POST requests. + Registers a vote callback that gets called whenever this endpoint receives POST requests. The callback can be either sync or async. - The callback can be either sync or async. This method can be used as a decorator or a decorator factory. - :Example: - .. code-block:: python + .. code-block:: python + + webhook_manager = topgg.WebhookManager() + + endpoint = ( + topgg.WebhookEndpoint() + .type(topgg.WebhookType.BOT) + .route("/dblwebhook") + .auth("youshallnotpass") + ) + + # The following are valid. + endpoint.callback(lambda vote: print(f"Got a vote: {vote!r}")) - import topgg - webhook_manager = topgg.WebhookManager() - endpoint = ( - topgg.WebhookEndpoint() - .type(topgg.WebhookType.BOT) - .route("/dblwebhook") - .auth("youshallnotpass") - ) + # Used as decorator, the decorated function will become the WebhookEndpoint object. + @endpoint.callback + def on_vote(vote: topgg.BotVoteData) -> None: ... + + + # Used as decorator factory, the decorated function will still be the function itself. + @endpoint.callback() + def on_vote(vote: topgg.BotVoteData) -> None: ... - # The following are valid. - endpoint.callback(lambda vote_data: print("Receives a vote!", vote_data)) - # Used as decorator, the decorated function will become the WebhookEndpoint object. - @endpoint.callback - def endpoint(vote_data: topgg.BotVoteData): - ... + webhook_manager.endpoint(endpoint) - # Used as decorator factory, the decorated function will still be the function itself. - @endpoint.callback() - def on_vote(vote_data: topgg.BotVoteData): - ... + await webhook_manager.start(8080) - webhook_manager.endpoint(endpoint) + :param callback_: The endpoint's new vote callback. + :type callback_: Any + + :returns: The object itself if ``callback`` is not :py:obj:`None`, otherwise the object's own callback. + :rtype: Union[Any, :class:`.WebhookEndpoint`] """ + if callback_ is not None: self._callback = callback_ + return self return self.callback @@ -276,57 +279,64 @@ def on_vote(vote_data: topgg.BotVoteData): class BoundWebhookEndpoint(WebhookEndpoint): """ - A WebhookEndpoint with a WebhookManager bound to it. + A :class:`.WebhookEndpoint` with a :class:`.WebhookManager` bound to it. - You can instantiate this object using the :meth:`WebhookManager.endpoint` method. + You can instantiate this object using the :meth:`.WebhookManager.endpoint` method. - :Example: - .. code-block:: python + .. code-block:: python - import topgg + endpoint = ( + topgg.WebhookManager() + .endpoint() + .type(topgg.WebhookType.BOT) + .route("/dblwebhook") + .auth("youshallnotpass") + ) - webhook_manager = ( - topgg.WebhookManager() - .endpoint() - .type(topgg.WebhookType.BOT) - .route("/dblwebhook") - .auth("youshallnotpass") - ) + # The following are valid. + endpoint.callback(lambda vote: print(f"Got a vote: {vote!r}")) - # The following are valid. - endpoint.callback(lambda vote_data: print("Receives a vote!", vote_data)) - # Used as decorator, the decorated function will become the BoundWebhookEndpoint object. - @endpoint.callback - def endpoint(vote_data: topgg.BotVoteData): - ... + # Used as decorator, the decorated function will become the BoundWebhookEndpoint object. + @endpoint.callback + def on_vote(vote: topgg.BotVoteData) -> None: ... - # Used as decorator factory, the decorated function will still be the function itself. - @endpoint.callback() - def on_vote(vote_data: topgg.BotVoteData): - ... - endpoint.add_to_manager() + # Used as decorator factory, the decorated function will still be the function itself. + @endpoint.callback() + def on_vote(vote: topgg.BotVoteData) -> None: ... + + + endpoint.add_to_manager() + + await endpoint.manager.start(8080) + + :param manager: The webhook manager to use. + :type manager: :class:`.WebhookManager` """ - __slots__ = ("manager",) + __slots__: tuple[str, ...] = ("manager",) def __init__(self, manager: WebhookManager): super().__init__() + self.manager = manager + def __repr__(self) -> str: + return f"<{__class__.__name__} manager={self.manager!r}>" + def add_to_manager(self) -> WebhookManager: """ Adds this endpoint to the webhook manager. - Returns: - :obj:`WebhookManager` + :exception TopGGException: If the webhook manager is not :py:obj:`None` and is not an instance of :class:`.WebhookEndpoint`. - Raises: - :obj:`errors.TopGGException`: - If the object lacks attributes. + :returns: The webhook manager used. + :rtype: :class:`WebhookManager` """ + self.manager.endpoint(self) + return self.manager @@ -334,32 +344,30 @@ def endpoint( route: str, type: WebhookType, auth: str = "" ) -> t.Callable[[t.Callable[..., t.Any]], WebhookEndpoint]: """ - A decorator factory for instantiating WebhookEndpoint. + A decorator factory for instantiating a :class:`.WebhookEndpoint`. - Args: - route (str) - The route for the endpoint. - type (WebhookType) - The type of the endpoint. - auth (str) - The auth for the endpoint. + .. code-block:: python - Returns: - :obj:`typing.Callable` [[ :obj:`typing.Callable` [..., :obj:`typing.Any` ]], :obj:`WebhookEndpoint` ]: - The actual decorator. + manager = topgg.WebhookManager() - :Example: - .. code-block:: python - import topgg + @topgg.endpoint("/dblwebhook", WebhookType.BOT, "youshallnotpass") + async def on_vote(vote: topgg.BotVoteData): ... + + + manager.endpoint(on_vote) + + await manager.start(8080) + + :param route: The endpoint's route. + :type route: :py:class:`str` + :param type: The endpoint's type. + :type type: :class:`.WebhookType` + :param auth: The endpoint's password. + :type auth: :py:class:`str` - @topgg.endpoint("/dblwebhook", WebhookType.BOT, "youshallnotpass") - async def on_vote( - vote_data: topgg.BotVoteData, - # database here is an injected data - database: Database = topgg.data(Database), - ): - ... + :returns: The actual decorator. + :rtype: Callable[[Callable[..., Any]], :class:`.WebhookEndpoint`] """ def decorator(func: t.Callable[..., t.Any]) -> WebhookEndpoint: