From 745cd1d6d3d6227d2a1e82cf25611d76221c40cd Mon Sep 17 00:00:00 2001 From: decorator-factory <42166884+decorator-factory@users.noreply.github.com> Date: Sat, 7 Aug 2021 05:23:03 +0300 Subject: Fix type annotations --- bot/utils/decorators.py | 37 +++++++++++++++++++------------------ 1 file changed, 19 insertions(+), 18 deletions(-) (limited to 'bot/utils/decorators.py') diff --git a/bot/utils/decorators.py b/bot/utils/decorators.py index c0783144..2612ff81 100644 --- a/bot/utils/decorators.py +++ b/bot/utils/decorators.py @@ -2,9 +2,10 @@ import asyncio import functools import logging import random -import typing as t from asyncio import Lock +from collections.abc import Container from functools import wraps +from typing import Callable, Union from weakref import WeakValueDictionary from discord import Colour, Embed @@ -32,7 +33,7 @@ class InMonthCheckFailure(CheckFailure): pass -def seasonal_task(*allowed_months: Month, sleep_time: t.Union[float, int] = ONE_DAY) -> t.Callable: +def seasonal_task(*allowed_months: Month, sleep_time: Union[float, int] = ONE_DAY) -> Callable: """ Perform the decorated method periodically in `allowed_months`. @@ -44,7 +45,7 @@ def seasonal_task(*allowed_months: Month, sleep_time: t.Union[float, int] = ONE_ The wrapped task is responsible for waiting for the bot to be ready, if necessary. """ - def decorator(task_body: t.Callable) -> t.Callable: + def decorator(task_body: Callable) -> Callable: @functools.wraps(task_body) async def decorated_task(*args, **kwargs) -> None: """Call `task_body` once every `sleep_time` seconds in `allowed_months`.""" @@ -63,13 +64,13 @@ def seasonal_task(*allowed_months: Month, sleep_time: t.Union[float, int] = ONE_ return decorator -def in_month_listener(*allowed_months: Month) -> t.Callable: +def in_month_listener(*allowed_months: Month) -> Callable: """ Shield a listener from being invoked outside of `allowed_months`. The check is performed against current UTC month. """ - def decorator(listener: t.Callable) -> t.Callable: + def decorator(listener: Callable) -> Callable: @functools.wraps(listener) async def guarded_listener(*args, **kwargs) -> None: """Wrapped listener will abort if not in allowed month.""" @@ -84,7 +85,7 @@ def in_month_listener(*allowed_months: Month) -> t.Callable: return decorator -def in_month_command(*allowed_months: Month) -> t.Callable: +def in_month_command(*allowed_months: Month) -> Callable: """ Check whether the command was invoked in one of `enabled_months`. @@ -106,7 +107,7 @@ def in_month_command(*allowed_months: Month) -> t.Callable: return commands.check(predicate) -def in_month(*allowed_months: Month) -> t.Callable: +def in_month(*allowed_months: Month) -> Callable: """ Universal decorator for season-locking commands and listeners alike. @@ -124,7 +125,7 @@ def in_month(*allowed_months: Month) -> t.Callable: manually set to True - this causes a circumvention of the group's callback and the seasonal check applied to it. """ - def decorator(callable_: t.Callable) -> t.Callable: + def decorator(callable_: Callable) -> Callable: # Functions decorated as commands are turned into instances of `Command` if isinstance(callable_, Command): logging.debug(f"Command {callable_.qualified_name} will be locked to {human_months(allowed_months)}") @@ -144,7 +145,7 @@ def in_month(*allowed_months: Month) -> t.Callable: return decorator -def with_role(*role_ids: int) -> t.Callable: +def with_role(*role_ids: int) -> Callable: """Check to see whether the invoking user has any of the roles specified in role_ids.""" async def predicate(ctx: Context) -> bool: if not ctx.guild: # Return False in a DM @@ -167,7 +168,7 @@ def with_role(*role_ids: int) -> t.Callable: return commands.check(predicate) -def without_role(*role_ids: int) -> t.Callable: +def without_role(*role_ids: int) -> Callable: """Check whether the invoking user does not have all of the roles specified in role_ids.""" async def predicate(ctx: Context) -> bool: if not ctx.guild: # Return False in a DM @@ -187,7 +188,7 @@ def without_role(*role_ids: int) -> t.Callable: return commands.check(predicate) -def whitelist_check(**default_kwargs: t.Container[int]) -> t.Callable[[Context], bool]: +def whitelist_check(**default_kwargs: Container[int]) -> Callable[[Context], bool]: """ Checks if a message is sent in a whitelisted context. @@ -222,8 +223,8 @@ def whitelist_check(**default_kwargs: t.Container[int]) -> t.Callable[[Context], kwargs[arg] = new_value # Merge containers - elif isinstance(default_value, t.Container): - if isinstance(new_value, t.Container): + elif isinstance(default_value, Container): + if isinstance(new_value, Container): kwargs[arg] = (*default_value, *new_value) else: kwargs[arg] = new_value @@ -279,7 +280,7 @@ def whitelist_check(**default_kwargs: t.Container[int]) -> t.Callable[[Context], return predicate -def whitelist_override(bypass_defaults: bool = False, **kwargs: t.Container[int]) -> t.Callable: +def whitelist_override(bypass_defaults: bool = False, **kwargs: Container[int]) -> Callable: """ Override global whitelist context, with the kwargs specified. @@ -288,7 +289,7 @@ def whitelist_override(bypass_defaults: bool = False, **kwargs: t.Container[int] This decorator has to go before (below) below the `command` decorator. """ - def inner(func: t.Callable) -> t.Callable: + def inner(func: Callable) -> Callable: func.override = kwargs func.override_reset = bypass_defaults return func @@ -296,7 +297,7 @@ def whitelist_override(bypass_defaults: bool = False, **kwargs: t.Container[int] return inner -def locked() -> t.Union[t.Callable, None]: +def locked() -> Union[Callable, None]: """ Allows the user to only run one instance of the decorated command at a time. @@ -304,11 +305,11 @@ def locked() -> t.Union[t.Callable, None]: This decorator has to go before (below) the `command` decorator. """ - def wrap(func: t.Callable) -> t.Union[t.Callable, None]: + def wrap(func: Callable) -> Union[Callable, None]: func.__locks = WeakValueDictionary() @wraps(func) - async def inner(self: t.Callable, ctx: Context, *args, **kwargs) -> t.Union[t.Callable, None]: + async def inner(self: Callable, ctx: Context, *args, **kwargs) -> Union[Callable, None]: lock = func.__locks.setdefault(ctx.author.id, Lock()) if lock.locked(): embed = Embed() -- cgit v1.2.3 From f4e5e3850c9a030e0125a908dc5a8dbbcd6659dd Mon Sep 17 00:00:00 2001 From: Xithrius Date: Wed, 1 Sep 2021 18:32:55 -0700 Subject: Union item with None to Optional with item. --- bot/exts/easter/egg_decorating.py | 6 +++--- bot/exts/evergreen/fun.py | 2 +- bot/exts/halloween/spookynamerate.py | 4 ++-- bot/utils/decorators.py | 8 ++++---- 4 files changed, 10 insertions(+), 10 deletions(-) (limited to 'bot/utils/decorators.py') diff --git a/bot/exts/easter/egg_decorating.py b/bot/exts/easter/egg_decorating.py index 6201d424..fb5701c4 100644 --- a/bot/exts/easter/egg_decorating.py +++ b/bot/exts/easter/egg_decorating.py @@ -4,7 +4,7 @@ import random from contextlib import suppress from io import BytesIO from pathlib import Path -from typing import Union +from typing import Optional, Union import discord from PIL import Image @@ -33,7 +33,7 @@ class EggDecorating(commands.Cog): """Decorate some easter eggs!""" @staticmethod - def replace_invalid(colour: str) -> Union[int, None]: + def replace_invalid(colour: str) -> Optional[int]: """Attempts to match with HTML or XKCD colour names, returning the int value.""" with suppress(KeyError): return int(HTML_COLOURS[colour], 16) @@ -44,7 +44,7 @@ class EggDecorating(commands.Cog): @commands.command(aliases=("decorateegg",)) async def eggdecorate( self, ctx: commands.Context, *colours: Union[discord.Colour, str] - ) -> Union[Image.Image, None]: + ) -> Optional[Image.Image]: """ Picks a random egg design and decorates it using the given colours. diff --git a/bot/exts/evergreen/fun.py b/bot/exts/evergreen/fun.py index 1783d7b4..fd17a691 100644 --- a/bot/exts/evergreen/fun.py +++ b/bot/exts/evergreen/fun.py @@ -192,7 +192,7 @@ class Fun(Cog): Returns a tuple of: str: If `text` is a valid discord Message, the contents of the message, else `text`. - Union[Embed, None]: The embed if found in the valid Message, else None + Optional[Embed]: The embed if found in the valid Message, else None """ embed = None diff --git a/bot/exts/halloween/spookynamerate.py b/bot/exts/halloween/spookynamerate.py index 3d6d95fa..bab79e56 100644 --- a/bot/exts/halloween/spookynamerate.py +++ b/bot/exts/halloween/spookynamerate.py @@ -6,7 +6,7 @@ from datetime import datetime, timedelta from logging import getLogger from os import getenv from pathlib import Path -from typing import Union +from typing import Optional from async_rediscache import RedisCache from discord import Embed, Reaction, TextChannel, User @@ -355,7 +355,7 @@ class SpookyNameRate(Cog): return embed - async def get_channel(self) -> Union[TextChannel, None]: + async def get_channel(self) -> Optional[TextChannel]: """Gets the sir-lancebot-channel after waiting until ready.""" await self.bot.wait_until_ready() channel = self.bot.get_channel( diff --git a/bot/utils/decorators.py b/bot/utils/decorators.py index 2612ff81..132aaa87 100644 --- a/bot/utils/decorators.py +++ b/bot/utils/decorators.py @@ -5,7 +5,7 @@ import random from asyncio import Lock from collections.abc import Container from functools import wraps -from typing import Callable, Union +from typing import Callable, Optional, Union from weakref import WeakValueDictionary from discord import Colour, Embed @@ -297,7 +297,7 @@ def whitelist_override(bypass_defaults: bool = False, **kwargs: Container[int]) return inner -def locked() -> Union[Callable, None]: +def locked() -> Optional[Callable]: """ Allows the user to only run one instance of the decorated command at a time. @@ -305,11 +305,11 @@ def locked() -> Union[Callable, None]: This decorator has to go before (below) the `command` decorator. """ - def wrap(func: Callable) -> Union[Callable, None]: + def wrap(func: Callable) -> Optional[Callable]: func.__locks = WeakValueDictionary() @wraps(func) - async def inner(self: Callable, ctx: Context, *args, **kwargs) -> Union[Callable, None]: + async def inner(self: Callable, ctx: Context, *args, **kwargs) -> Optional[Callable]: lock = func.__locks.setdefault(ctx.author.id, Lock()) if lock.locked(): embed = Embed() -- cgit v1.2.3