diff --git a/requirements.txt b/requirements.txt index 9ad05803..ff685f61 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1 @@ -aiohttp>=3.6.0,<3.9.0 \ No newline at end of file +aiohttp>3.6.0 diff --git a/setup.py b/setup.py index 36733afb..d7b7c5ff 100644 --- a/setup.py +++ b/setup.py @@ -34,7 +34,7 @@ author=f"{metadata.author}, Top.gg", author_email="shivaco.osu@gmail.com", maintainer=f"{metadata.maintainer}, Top.gg", - url="https://github.com/top-gg/python-sdk", + url="https://github.com/RealYuri001/python-sdk", version=metadata.version, packages=find_packages(), license=metadata.license, @@ -42,7 +42,7 @@ long_description=readme, package_data={"topgg": ["py.typed"]}, include_package_data=True, - python_requires=">= 3.6", + python_requires=">= 3.8", install_requires=requirements, keywords="discord bot server list discordservers serverlist discordbots botlist topgg top.gg", classifiers=[ @@ -52,10 +52,9 @@ "Natural Language :: English", "Operating System :: OS Independent", "Programming Language :: Python :: 3 :: Only", - "Programming Language :: Python :: 3.6", - "Programming Language :: Python :: 3.7", "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", "Topic :: Internet", "Topic :: Software Development :: Libraries", "Topic :: Software Development :: Libraries :: Python Modules", diff --git a/topgg/__init__.py b/topgg/__init__.py index 1a9025eb..d9f766a1 100644 --- a/topgg/__init__.py +++ b/topgg/__init__.py @@ -8,11 +8,11 @@ :license: MIT, see LICENSE for more details. """ -__title__ = "topggpy" +__title__ = "topgg.py" __author__ = "Assanali Mukhanov" -__maintainer__ = "Norizon" +__maintainer__ = "RealKiller666" __license__ = "MIT" -__version__ = "2.0.0a1" +__version__ = "2.0.0a3" from .autopost import * from .client import * diff --git a/topgg/autopost.py b/topgg/autopost.py index 3bfe4afa..0714c52f 100644 --- a/topgg/autopost.py +++ b/topgg/autopost.py @@ -22,23 +22,32 @@ __all__ = ["AutoPoster"] +from __future__ import annotations + import asyncio import datetime import sys import traceback -import typing as t +from typing import ( + Any, + Callable, + Optional, + overload, + TYPE_CHECKING, + Union +) from topgg import errors from .types import StatsWrapper -if t.TYPE_CHECKING: +if TYPE_CHECKING: import asyncio from .client import DBLClient -CallbackT = t.Callable[..., t.Any] -StatsCallbackT = t.Callable[[], StatsWrapper] +CallbackT = Callable[..., Any] +StatsCallbackT = Callable[[], StatsWrapper] class AutoPoster: @@ -67,9 +76,9 @@ class AutoPoster: _success: CallbackT _stats: CallbackT _interval: float - _task: t.Optional["asyncio.Task[None]"] + _task: Optional[asyncio.Task[None]] - def __init__(self, client: "DBLClient") -> None: + def __init__(self, client: DBLClient) -> None: super().__init__() self.client = client self._interval: float = 900 @@ -82,15 +91,15 @@ def _default_error_handler(self, exception: Exception) -> None: type(exception), exception, exception.__traceback__, file=sys.stderr ) - @t.overload - def on_success(self, callback: None) -> t.Callable[[CallbackT], CallbackT]: + @overload + def on_success(self, callback: None) -> Callable[[CallbackT], CallbackT]: ... - @t.overload - def on_success(self, callback: CallbackT) -> "AutoPoster": + @overload + def on_success(self, callback: CallbackT) -> AutoPoster: ... - def on_success(self, callback: t.Any = None) -> t.Any: + def on_success(self, callback: Any = None) -> Any: """ Registers an autopost success callback. The callback can be either sync or async. @@ -123,15 +132,15 @@ def decorator(callback: CallbackT) -> CallbackT: return decorator - @t.overload - def on_error(self, callback: None) -> t.Callable[[CallbackT], CallbackT]: + @overload + def on_error(self, callback: None) -> Callable[[CallbackT], CallbackT]: ... - @t.overload - def on_error(self, callback: CallbackT) -> "AutoPoster": + @overload + def on_error(self, callback: CallbackT) -> AutoPoster: ... - def on_error(self, callback: t.Any = None) -> t.Any: + def on_error(self, callback: Any = None) -> Any: """ Registers an autopost error callback. The callback can be either sync or async. @@ -168,15 +177,15 @@ def decorator(callback: CallbackT) -> CallbackT: return decorator - @t.overload - def stats(self, callback: None) -> t.Callable[[StatsCallbackT], StatsCallbackT]: + @overload + def stats(self, callback: None) -> Callable[[StatsCallbackT], StatsCallbackT]: ... - @t.overload - def stats(self, callback: StatsCallbackT) -> "AutoPoster": + @overload + def stats(self, callback: StatsCallbackT) -> AutoPoster: ... - def stats(self, callback: t.Any = None) -> t.Any: + def stats(self, callback: Any = None) -> Any: """ Registers a function that returns an instance of :obj:`~.types.StatsWrapper`. @@ -221,11 +230,11 @@ def interval(self) -> float: return self._interval @interval.setter - def interval(self, seconds: t.Union[float, datetime.timedelta]) -> None: + def interval(self, seconds: Union[float, datetime.timedelta]) -> None: """Alias to :meth:`~.autopost.AutoPoster.set_interval`.""" self.set_interval(seconds) - def set_interval(self, seconds: t.Union[float, datetime.timedelta]) -> "AutoPoster": + def set_interval(self, seconds: Union[float, datetime.timedelta]) -> AutoPoster: """ Sets the interval between posting stats. @@ -255,7 +264,7 @@ def _refresh_state(self) -> None: self._task = None self._stopping = False - def _fut_done_callback(self, future: "asyncio.Future") -> None: + def _fut_done_callback(self, future: asyncio.Future) -> None: self._refresh_state() if future.cancelled(): return @@ -272,8 +281,7 @@ async def _internal_loop(self) -> None: if isinstance(err, errors.Unauthorized): 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: @@ -283,7 +291,7 @@ async def _internal_loop(self) -> None: finally: self._refresh_state() - def start(self) -> "asyncio.Task[None]": + def start(self) -> asyncio.Task[None]: """ Starts the autoposting loop. @@ -332,4 +340,4 @@ def cancel(self) -> None: self._task.cancel() self._refresh_state() - return None + return None \ No newline at end of file diff --git a/topgg/client.py b/topgg/client.py index 0f1a72db..7bc3e107 100644 --- a/topgg/client.py +++ b/topgg/client.py @@ -24,7 +24,7 @@ __all__ = ["DBLClient"] -import typing as t +from typing import overload, Optional, Any, Union import aiohttp @@ -60,9 +60,9 @@ def __init__( self, token: str, *, - default_bot_id: t.Optional[int] = None, - session: t.Optional[aiohttp.ClientSession] = None, - **kwargs: t.Any, + default_bot_id: Optional[int] = None, + session: Optional[aiohttp.ClientSession] = None, + **kwargs: Any, ) -> None: super().__init__() self._token = token @@ -70,7 +70,7 @@ def __init__( self._is_closed = False if session is not None: self.http = HTTPClient(token, session=session) - self._autopost: t.Optional[AutoPoster] = None + self._autopost: Optional[AutoPoster] = None @property def is_closed(self) -> bool: @@ -83,7 +83,7 @@ async def _ensure_session(self) -> None: if not hasattr(self, "http"): self.http = HTTPClient(self._token, session=None) - def _validate_and_get_bot_id(self, bot_id: t.Optional[int]) -> int: + def _validate_and_get_bot_id(self, bot_id: Optional[int]) -> int: bot_id = bot_id or self.default_bot_id if bot_id is None: raise errors.ClientException("bot_id or default_bot_id is unset.") @@ -104,27 +104,27 @@ async def get_weekend_status(self) -> bool: data = await self.http.get_weekend_status() return data["is_weekend"] - @t.overload + @overload async def post_guild_count(self, stats: types.StatsWrapper) -> None: ... - @t.overload + @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, + 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, + stats: Any = None, *, - guild_count: t.Any = None, - shard_count: t.Any = None, - shard_id: t.Any = None, + guild_count: Any = None, + shard_count: Any = None, + shard_id: Any = None, ) -> None: """Posts your bot's guild count and shards info to Top.gg. @@ -162,7 +162,7 @@ async def post_guild_count( await self.http.post_guild_count(guild_count, shard_count, shard_id) async def get_guild_count( - self, bot_id: t.Optional[int] = None + self, bot_id: Optional[int] = None ) -> types.BotStatsData: """Gets a bot's guild count and shard info from Top.gg. @@ -185,7 +185,7 @@ async def get_guild_count( response = await self.http.get_guild_count(bot_id) return types.BotStatsData(**response) - async def get_bot_votes(self) -> t.List[types.BriefUserData]: + async def get_bot_votes(self) -> list[types.BriefUserData]: """Gets information about last 1000 votes for your bot on Top.gg. Note: @@ -209,7 +209,7 @@ async def get_bot_votes(self) -> t.List[types.BriefUserData]: response = await self.http.get_bot_votes(self.default_bot_id) return [types.BriefUserData(**user) for user in response] - async def get_bot_info(self, bot_id: t.Optional[int] = None) -> types.BotData: + async def get_bot_info(self, bot_id: Optional[int] = None) -> types.BotData: """This function is a coroutine. Gets information about a bot from Top.gg. @@ -238,10 +238,10 @@ async def get_bots( self, limit: int = 50, offset: int = 0, - sort: t.Optional[str] = None, - search: t.Optional[t.Dict[str, t.Any]] = None, - fields: t.Optional[t.List[str]] = None, - ) -> types.DataDict[str, t.Any]: + sort: Optional[str] = None, + search: Optional[dict[str, Any]] = None, + fields: Optional[list[str]] = None, + ) -> types.DataDict[str, Any]: """This function is a coroutine. Gets information about listed bots on Top.gg. @@ -354,8 +354,7 @@ def generate_widget(self, *, options: types.WidgetOptions) -> str: widget_format = options.format widget_type = f"/{options.type}" if options.type else "" - url = f"""https://top.gg/api/widget{widget_type}/{bot_id}.{widget_format}?{widget_query}""" - return url + return f"""https://top.gg/api/widget{widget_type}/{bot_id}.{widget_format}?{widget_query}""" async def close(self) -> None: """Closes all connections.""" diff --git a/topgg/data.py b/topgg/data.py index 7126d3bf..42e5231a 100644 --- a/topgg/data.py +++ b/topgg/data.py @@ -23,15 +23,28 @@ __all__ = ["data", "DataContainerMixin"] import inspect -import typing as t +from typing import ( + TypeVar, + Any, + Type, + cast, + Generic, + Mapping, + overload, + Callable, + Optional, + Union +) + +from typing_extensions import Self from topgg.errors import TopGGException -T = t.TypeVar("T") -DataContainerT = t.TypeVar("DataContainerT", bound="DataContainerMixin") +T = TypeVar("T") +DataContainerT = TypeVar("DataContainerT", bound="DataContainerMixin") -def data(type_: t.Type[T]) -> T: +def data(type_: Type[T]) -> T: """ Represents the injected data. This should be set as the parameter's default value. @@ -56,14 +69,14 @@ def data(type_: t.Type[T]) -> T: def get_stats(client: Client = topgg.data(Client)): return topgg.StatsWrapper(guild_count=len(client.guilds), shard_count=len(client.shards)) """ - return t.cast(T, Data(type_)) + return cast(T, Data(type_)) -class Data(t.Generic[T]): +class Data(Generic[T]): __slots__ = ("type",) - def __init__(self, type_: t.Type[T]) -> None: - self.type: t.Type[T] = type_ + def __init__(self, type_: Type[T]) -> None: + self.type: Type[T] = type_ class DataContainerMixin: @@ -77,10 +90,10 @@ class DataContainerMixin: __slots__ = ("_data",) def __init__(self) -> None: - self._data: t.Dict[t.Type, t.Any] = {type(self): self} + self._data: dict[Type, Any] = {type(self): Self} def set_data( - self: DataContainerT, data_: t.Any, *, override: bool = False + self: DataContainerT, data_: Any, *, override: bool = False ) -> DataContainerT: """ Sets data to be available in your functions. @@ -104,28 +117,28 @@ def set_data( self._data[type_] = data_ return self - @t.overload - def get_data(self, type_: t.Type[T]) -> t.Optional[T]: + @overload + def get_data(self, type_: Type[T]) -> Optional[T]: ... - @t.overload - def get_data(self, type_: t.Type[T], default: t.Any = None) -> t.Any: + @overload + def get_data(self, type_: Type[T], default: Any = None) -> Any: ... - def get_data(self, type_: t.Any, default: t.Any = None) -> t.Any: + def get_data(self, type_: Any, default: Any = None) -> Any: """Gets the injected data.""" return self._data.get(type_, default) async def _invoke_callback( - self, callback: t.Callable[..., T], *args: t.Any, **kwargs: t.Any + self, callback: Callable[..., T], *args: Any, **kwargs: Any ) -> T: - parameters: t.Mapping[str, inspect.Parameter] + parameters: 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 @@ -135,11 +148,11 @@ async def _invoke_callback( for k, v in signatures.items(): signatures[k] = self._resolve_data(v.type) - res = callback(*args, **{**signatures, **kwargs}) + res = callback(*args, Union[signatures, kwargs]) if inspect.isawaitable(res): return await res return res - def _resolve_data(self, type_: t.Type[T]) -> T: + def _resolve_data(self, type_: Type[T]) -> T: return self._data[type_] diff --git a/topgg/http.py b/topgg/http.py index 08160d67..469d8856 100644 --- a/topgg/http.py +++ b/topgg/http.py @@ -238,7 +238,7 @@ def get_user_vote(self, bot_id: int, user_id: int) -> Coroutine[Any, Any, dict]: async def _rate_limit_handler(until: float) -> None: """Handles the displayed message when we are ratelimited.""" - duration = round(until - datetime.utcnow().timestamp()) + duration = round(until - datetime.now().timestamp()) mins = duration / 60 fmt = ( "We have exhausted a ratelimit quota. Retrying in %.2f seconds (%.3f minutes)." diff --git a/topgg/ratelimiter.py b/topgg/ratelimiter.py index 028a98ee..6c36c8f8 100644 --- a/topgg/ratelimiter.py +++ b/topgg/ratelimiter.py @@ -60,10 +60,10 @@ def __init__( async def __aenter__(self) -> "AsyncRateLimiter": async with self.__lock: if len(self.calls) >= self.max_calls: - until = datetime.utcnow().timestamp() + self.period - self._timespan + until = datetime.now().timestamp() + self.period - self._timespan if self.callback: asyncio.ensure_future(self.callback(until)) - sleep_time = until - datetime.utcnow().timestamp() + sleep_time = until - datetime.now().timestamp() if sleep_time > 0: await asyncio.sleep(sleep_time) return self @@ -76,7 +76,7 @@ async def __aexit__( ) -> None: async with self.__lock: # Store the last operation timestamp. - self.calls.append(datetime.utcnow().timestamp()) + self.calls.append(datetime.now().timestamp()) while self._timespan >= self.period: self.calls.popleft() diff --git a/topgg/types.py b/topgg/types.py index 2da13f95..9dbbb2c1 100644 --- a/topgg/types.py +++ b/topgg/types.py @@ -16,7 +16,7 @@ # 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 +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMEN 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 @@ -25,25 +25,32 @@ __all__ = ["WidgetOptions", "StatsWrapper"] import dataclasses -import typing as t +from typing import ( + TypeVar, + MutableMapping, + Optional, + Any +) + from datetime import datetime -KT = t.TypeVar("KT") -VT = t.TypeVar("VT") -Colors = t.Dict[str, int] +KT = TypeVar("KT") +VT = TypeVar("VT") +Colors = 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("_") + return "".join( + [f"_{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("&")]} + if query := data.get("query", "").lstrip("?"): + query_dict = dict([pair.split("=") for pair in query.split("&")]) data["query"] = DataDict(**query_dict) else: data["query"] = {} @@ -70,11 +77,10 @@ def parse_dict(d: dict) -> dict: if "id" in key.lower(): if value == "": value = None + elif isinstance(value, str) and value.isdigit(): + value = int(value) else: - if isinstance(value, str) and value.isdigit(): - value = int(value) - else: - continue + continue elif value == "": value = None @@ -127,7 +133,7 @@ def parse_bot_stats_dict(d: dict) -> dict: return data -class DataDict(dict, t.MutableMapping[KT, VT]): +class DataDict(dict, MutableMapping[KT, VT]): """Base class used to represent received data from the API. Every data model subclasses this class. @@ -138,11 +144,11 @@ def __init__(self, **kwargs: VT) -> None: self.__dict__ = self -class WidgetOptions(DataDict[str, t.Any]): +class WidgetOptions(DataDict[str, Any]): """Model that represents widget options that are passed to Top.gg widget URL generated via - :meth:`DBLClient.generate_widget`.""" + :meth:`DBLCliengenerate_widget`.""" - id: t.Optional[int] + id: 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 @@ -151,19 +157,19 @@ class WidgetOptions(DataDict[str, t.Any]): """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``.""" + """Format to apply to the widge 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.""" def __init__( self, - id: t.Optional[int] = None, - format: t.Optional[str] = None, - type: t.Optional[str] = None, + id: Optional[int] = None, + format: Optional[str] = None, + type: Optional[str] = None, noavatar: bool = False, - colors: t.Optional[Colors] = None, - colours: t.Optional[Colors] = None, + colors: Optional[Colors] = None, + colours: Optional[Colors] = None, ): super().__init__( id=id or None, @@ -181,70 +187,70 @@ def colours(self) -> Colors: def colours(self, value: Colors) -> None: self.colors = value - def __setitem__(self, key: str, value: t.Any) -> None: + def __setitem__(self, key: str, value: Any) -> None: if key == "colours": key = "colors" super().__setitem__(key, value) - def __getitem__(self, item: str) -> t.Any: + def __getitem__(self, item: str) -> Any: if item == "colours": item = "colors" return super().__getitem__(item) - def get(self, key: str, default: t.Any = None) -> t.Any: + def get(self, key: str, default: Any = None) -> Any: """:meta private:""" if key == "colours": key = "colors" return super().get(key, default) -class BotData(DataDict[str, t.Any]): +class BotData(DataDict[str, Any]): """Model that contains information about a listed bot on top.gg. The data this model contains can be found `here `__.""" id: int - """The ID of the bot.""" + """The ID of the bo""" username: str - """The username of the bot.""" + """The username of the bo""" discriminator: str - """The discriminator of the bot.""" + """The discriminator of the bo""" - avatar: t.Optional[str] - """The avatar hash of the bot.""" + avatar: Optional[str] + """The avatar hash of the bo""" def_avatar: str """The avatar hash of the bot's default avatar.""" prefix: str - """The prefix of the bot.""" + """The prefix of the bo""" shortdesc: str - """The brief description of the bot.""" + """The brief description of the bo""" - longdesc: t.Optional[str] - """The long description of the bot.""" + longdesc: Optional[str] + """The long description of the bo""" - tags: t.List[str] + tags: list[str] """The tags the bot has.""" - website: t.Optional[str] - """The website of the bot.""" + website: Optional[str] + """The website of the bo""" - support: t.Optional[str] + support: Optional[str] """The invite code of the bot's support server.""" - github: t.Optional[str] + github: Optional[str] """The GitHub URL of the repo of the bot.""" - owners: t.List[int] + owners: list[int] """The IDs of the owners of the bot.""" - guilds: t.List[int] + guilds: list[int] """The guilds the bot is in.""" - invite: t.Optional[str] + invite: Optional[str] """The invite URL of the bot.""" date: datetime @@ -253,7 +259,7 @@ class BotData(DataDict[str, t.Any]): certified_bot: bool """Whether or not the bot is certified.""" - vanity: t.Optional[str] + vanity: Optional[str] """The vanity URL of the bot.""" points: int @@ -265,25 +271,25 @@ class BotData(DataDict[str, t.Any]): donatebotguildid: int """The guild ID for the donatebot setup.""" - def __init__(self, **kwargs: t.Any): + def __init__(self, **kwargs: Any): super().__init__(**parse_bot_dict(kwargs)) -class BotStatsData(DataDict[str, t.Any]): - """Model that contains information about a listed bot's guild and shard count.""" +class BotStatsData(DataDict[str, Any]): + """Model that contains information about a listed bot's guild and shard coun""" - server_count: t.Optional[int] + server_count: 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] + shard_count: Optional[int] """The amount of shards a bot has.""" - def __init__(self, **kwargs: t.Any): + def __init__(self, **kwargs: Any): super().__init__(**parse_bot_stats_dict(kwargs)) -class BriefUserData(DataDict[str, t.Any]): +class BriefUserData(DataDict[str, Any]): """Model that contains brief information about a Top.gg user.""" id: int @@ -293,7 +299,7 @@ class BriefUserData(DataDict[str, t.Any]): avatar: str """The Discord avatar URL of the user.""" - def __init__(self, **kwargs: t.Any): + def __init__(self, **kwargs: Any): if kwargs["id"].isdigit(): kwargs["id"] = int(kwargs["id"]) super().__init__(**kwargs) @@ -314,7 +320,7 @@ class SocialData(DataDict[str, str]): """The GitHub username of the user.""" -class UserData(DataDict[str, t.Any]): +class UserData(DataDict[str, Any]): """Model that contains information about a top.gg user. The data this model contains can be found `here `__.""" @@ -348,11 +354,11 @@ class UserData(DataDict[str, t.Any]): admin: bool """Whether or not the user is a Top.gg admin.""" - def __init__(self, **kwargs: t.Any): + def __init__(self, **kwargs: Any): super().__init__(**parse_user_dict(kwargs)) -class VoteDataDict(DataDict[str, t.Any]): +class VoteDataDict(DataDict[str, Any]): """Base model that represents received information from Top.gg via webhooks.""" type: str @@ -362,7 +368,7 @@ class VoteDataDict(DataDict[str, t.Any]): query: DataDict """Query parameters in :obj:`~.DataDict`.""" - def __init__(self, **kwargs: t.Any): + def __init__(self, **kwargs: Any): super().__init__(**parse_vote_dict(kwargs)) @@ -388,10 +394,10 @@ class GuildVoteData(VoteDataDict): @dataclasses.dataclass class StatsWrapper: guild_count: int - """The guild count.""" + """The guild coun""" - shard_count: t.Optional[int] = None - """The shard count.""" + shard_count: Optional[int] = None + """The shard coun""" - shard_id: t.Optional[int] = None + shard_id: Optional[int] = None """The shard ID the guild count belongs to.""" diff --git a/topgg/webhook.py b/topgg/webhook.py index 4b94ec2b..c7eb3038 100644 --- a/topgg/webhook.py +++ b/topgg/webhook.py @@ -30,8 +30,18 @@ "WebhookType", ] +from __future__ import annotations + import enum -import typing as t +from typing import ( + Callable, + Optional, + TypeVar, + Awaitable, + TYPE_CHECKING, + Any, + overload +) import aiohttp from aiohttp import web @@ -41,11 +51,11 @@ from .data import DataContainerMixin from .types import BotVoteData, GuildVoteData -if t.TYPE_CHECKING: +if TYPE_CHECKING: from aiohttp.web import Request, StreamResponse -T = t.TypeVar("T", bound="WebhookEndpoint") -_HandlerT = t.Callable[["Request"], t.Awaitable["StreamResponse"]] +T = TypeVar("T", bound="WebhookEndpoint") +_HandlerT = Callable[["Request"], Awaitable["StreamResponse"]] class WebhookType(enum.Enum): @@ -73,15 +83,15 @@ def __init__(self) -> None: self.__app = web.Application() self._is_running = False - @t.overload - def endpoint(self, endpoint_: None = None) -> "BoundWebhookEndpoint": + @overload + def endpoint(self, endpoint_: None = None) -> BoundWebhookEndpoint: ... - @t.overload - def endpoint(self, endpoint_: "WebhookEndpoint") -> "WebhookManager": + @overload + def endpoint(self, endpoint_: WebhookEndpoint) -> WebhookManager: ... - def endpoint(self, endpoint_: t.Optional["WebhookEndpoint"] = None) -> t.Any: + def endpoint(self, endpoint_: Optional["WebhookEndpoint"] = None) -> Any: """Helper method that returns a WebhookEndpoint object. Args: @@ -97,25 +107,24 @@ def endpoint(self, endpoint_: t.Optional["WebhookEndpoint"] = None) -> t.Any: :obj:`~.errors.TopGGException` If the endpoint is lacking attributes. """ - if endpoint_: - if not hasattr(endpoint_, "_callback"): - raise TopGGException("endpoint missing callback.") - - if not hasattr(endpoint_, "_type"): - raise TopGGException("endpoint missing type.") - - if not hasattr(endpoint_, "_route"): - raise TopGGException("endpoint missing route.") - - self.app.router.add_post( - endpoint_._route, - self._get_handler( - endpoint_._type, endpoint_._auth, endpoint_._callback - ), - ) - return self - - return BoundWebhookEndpoint(manager=self) + if not endpoint_: + return BoundWebhookEndpoint(manager=self) + if not hasattr(endpoint_, "_callback"): + raise TopGGException("endpoint missing callback.") + + if not hasattr(endpoint_, "_type"): + raise TopGGException("endpoint missing type.") + + if not hasattr(endpoint_, "_route"): + raise TopGGException("endpoint missing route.") + + self.app.router.add_post( + endpoint_._route, + self._get_handler( + endpoint_._type, endpoint_._auth, endpoint_._callback + ), + ) + return self async def start(self, port: int) -> None: """Runs the webhook. @@ -151,7 +160,7 @@ async def close(self) -> None: self._is_running = False def _get_handler( - self, type_: WebhookType, auth: str, callback: t.Callable[..., t.Any] + self, type_: WebhookType, auth: str, callback: Callable[..., Any] ) -> _HandlerT: async def _handler(request: aiohttp.web.Request) -> web.Response: if request.headers.get("Authorization", "") != auth: @@ -167,7 +176,7 @@ async def _handler(request: aiohttp.web.Request) -> web.Response: return _handler -CallbackT = t.Callable[..., t.Any] +CallbackT = Callable[..., Any] class WebhookEndpoint: @@ -180,7 +189,7 @@ class WebhookEndpoint: def __init__(self) -> None: self._auth = "" - def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Any: + def __call__(self, *args: Any, **kwargs: Any) -> Any: return self._callback(*args, **kwargs) def type(self: T, type_: WebhookType) -> T: @@ -224,15 +233,15 @@ def auth(self: T, auth_: str) -> T: self._auth = auth_ return self - @t.overload - def callback(self, callback_: None) -> t.Callable[[CallbackT], CallbackT]: + @overload + def callback(self, callback_: None) -> Callable[[CallbackT], CallbackT]: ... - @t.overload + @overload def callback(self: T, callback_: CallbackT) -> T: ... - def callback(self, callback_: t.Any = None) -> t.Any: + def callback(self, callback_: Any = None) -> Any: """ Registers a vote callback, called whenever this endpoint receives POST requests. @@ -332,7 +341,7 @@ def add_to_manager(self) -> WebhookManager: def endpoint( route: str, type: WebhookType, auth: str = "" -) -> t.Callable[[t.Callable[..., t.Any]], WebhookEndpoint]: +) -> Callable[[Callable[..., Any]], WebhookEndpoint]: """ A decorator factory for instantiating WebhookEndpoint. @@ -362,7 +371,7 @@ async def on_vote( ... """ - def decorator(func: t.Callable[..., t.Any]) -> WebhookEndpoint: + def decorator(func: Callable[..., Any]) -> WebhookEndpoint: return WebhookEndpoint().route(route).type(type).auth(auth).callback(func) return decorator