From 646e21964c71e7a1de7f7bcccadebefa821c1e8d Mon Sep 17 00:00:00 2001 From: Numerlor Date: Mon, 13 Jun 2022 14:56:27 +0200 Subject: port command_wraps/update_wrapper_globals from bot --- botcore/utils/function.py | 116 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 116 insertions(+) create mode 100644 botcore/utils/function.py (limited to 'botcore/utils/function.py') diff --git a/botcore/utils/function.py b/botcore/utils/function.py new file mode 100644 index 00000000..1cde5cd9 --- /dev/null +++ b/botcore/utils/function.py @@ -0,0 +1,116 @@ +"""Utils for manipulating functions.""" + +from __future__ import annotations + +import functools +import types +import typing +from collections.abc import Sequence, Set +from typing import Callable # sphinx-autodoc-typehints breaks with collections.abc.Callable + +__all__ = ["command_wraps", "GlobalNameConflictError", "update_wrapper_globals"] + + +if typing.TYPE_CHECKING: + import typing_extensions + _P = typing_extensions.ParamSpec("_P") + _R = typing.TypeVar("_R") + + +class GlobalNameConflictError(Exception): + """Raised on a conflict between the globals used to resolve annotations of a wrapped function and its wrapper.""" + + +def update_wrapper_globals( + wrapper: Callable[_P, _R], + wrapped: Callable[_P, _R], + *, + ignored_conflict_names: Set[str] = frozenset(), +) -> Callable[_P, _R]: + r""" + Update globals of the ``wrapper`` function with the globals from the ``wrapped`` function. + + For forwardrefs in command annotations, discord.py uses the ``__global__`` attribute of the function + to resolve their values, with decorators that replace the function this breaks because they have + their own globals. + + This function creates a new function functionally identical to ``wrapper``\, which has the globals replaced with + a merge of ``wrapped``\s globals and the ``wrapper``\s globals. + + .. warning:: + This function captures the state of ``wrapped``\'s module's globals when it's called, + changes won't be reflected in the new function's globals. + + Args: + wrapper: The function to wrap. + wrapped: The function to wrap with. + ignored_conflict_names: A set of names to ignore if a conflict between them is found. + + Raises: + :exc:`GlobalNameConflictError`: + If ``wrapper`` and ``wrapped`` share a global name that's also used in ``wrapped``\'s typehints, + and is not in ``ignored_conflict_names``. + """ + wrapped = typing.cast(types.FunctionType, wrapped) + wrapper = typing.cast(types.FunctionType, wrapper) + + annotation_global_names = ( + ann.split(".", maxsplit=1)[0] for ann in wrapped.__annotations__.values() if isinstance(ann, str) + ) + # Conflicting globals from both functions' modules that are also used in the wrapper and in wrapped's annotations. + shared_globals = ( + set(wrapper.__code__.co_names) + & set(annotation_global_names) + & set(wrapped.__globals__) + & set(wrapper.__globals__) + - ignored_conflict_names + ) + if shared_globals: + raise GlobalNameConflictError( + f"wrapper and the wrapped function share the following " + f"global names used by annotations: {', '.join(shared_globals)}. Resolve the conflicts or add " + f"the name to the `ignored_conflict_names` set to suppress this error if this is intentional." + ) + + new_globals = wrapper.__globals__.copy() + new_globals.update((k, v) for k, v in wrapped.__globals__.items() if k not in wrapper.__code__.co_names) + return types.FunctionType( + code=wrapper.__code__, + globals=new_globals, + name=wrapper.__name__, + argdefs=wrapper.__defaults__, + closure=wrapper.__closure__, + ) + + +def command_wraps( + wrapped: Callable[_P, _R], + assigned: Sequence[str] = functools.WRAPPER_ASSIGNMENTS, + updated: Sequence[str] = functools.WRAPPER_UPDATES, + *, + ignored_conflict_names: Set[str] = frozenset(), +) -> Callable[[Callable[_P, _R]], Callable[_P, _R]]: + r""" + Update the decorated function to look like ``wrapped``\, and update globals for discord.py forwardref evaluation. + + See :func:`update_wrapper_globals` for more details on how the globals are updated. + + Args: + wrapped: The function to wrap with. + assigned: Sequence of attribute names that are directly assigned from ``wrapped`` to ``wrapper``. + updated: Sequence of attribute names that are ``.update``d on ``wrapper`` from the attributes on ``wrapped``. + ignored_conflict_names: A set of names to ignore if a conflict between them is found. + + Returns: + A decorator that behaves like :func:`functools.wraps`, + with the wrapper replaced with the function :func:`update_wrapper_globals` returned. + """ # noqa: D200 + def decorator(wrapper: Callable[_P, _R]) -> Callable[_P, _R]: + return functools.update_wrapper( + update_wrapper_globals(wrapper, wrapped, ignored_conflict_names=ignored_conflict_names), + wrapped, + assigned, + updated, + ) + + return decorator -- cgit v1.2.3 From c8b23bbbd25372a55d6e3640b68cd96828922af0 Mon Sep 17 00:00:00 2001 From: Numerlor Date: Tue, 21 Jun 2022 15:09:57 +0200 Subject: reword docstrings Co-authored-by: MarkKoz --- botcore/utils/cooldown.py | 14 +++++++------- botcore/utils/function.py | 9 +++------ docs/utils.py | 2 +- 3 files changed, 11 insertions(+), 14 deletions(-) (limited to 'botcore/utils/function.py') diff --git a/botcore/utils/cooldown.py b/botcore/utils/cooldown.py index a06dce46..9e79e48a 100644 --- a/botcore/utils/cooldown.py +++ b/botcore/utils/cooldown.py @@ -70,9 +70,9 @@ class _CommandCooldownManager: """ Manage invocation cooldowns for a command through the arguments the command is called with. - A cooldown is set through `set_cooldown` for a channel with the given `call_arguments`, - if `is_on_cooldown` is checked within `cooldown_duration` seconds - of the call to `set_cooldown` with the same arguments, True is returned. + Use `set_cooldown` to set a cooldown, + and `is_on_cooldown` to check for a cooldown for a channel with the given arguments. + A cooldown lasts for `cooldown_duration` seconds. """ def __init__(self, *, cooldown_duration: float): @@ -99,7 +99,7 @@ class _CommandCooldownManager: cooldowns_list.append(_CooldownItem(call_arguments, timeout_timestamp)) def is_on_cooldown(self, channel: Hashable, call_arguments: _ArgsTuple) -> bool: - """Check whether ``call_arguments`` is on cooldown in ``channel``.""" + """Check whether `call_arguments` is on cooldown in `channel`.""" current_time = time.monotonic() try: return self._cooldowns.get((channel, call_arguments), -math.inf) > current_time @@ -115,9 +115,9 @@ class _CommandCooldownManager: async def _periodical_cleanup(self, initial_delay: float) -> None: """ - Wait for `initial_delay`, after that delete stale items every hour. + Delete stale items every hour after waiting for `initial_delay`. - The `initial_delay` ensures we're not running cleanups for every command at the same time. + The `initial_delay` ensures cleanups are not running for every command at the same time. """ await asyncio.sleep(initial_delay) while True: @@ -151,7 +151,7 @@ def block_duplicate_invocations( Args: cooldown_duration: Length of the cooldown in seconds. - send_notice: If True, the user is notified of the cooldown with a reply. + send_notice: If :obj:`True`, notify the user about the cooldown with a reply. Returns: A decorator that adds a wrapper which applies the cooldowns. diff --git a/botcore/utils/function.py b/botcore/utils/function.py index 1cde5cd9..e8d24e90 100644 --- a/botcore/utils/function.py +++ b/botcore/utils/function.py @@ -28,17 +28,14 @@ def update_wrapper_globals( ignored_conflict_names: Set[str] = frozenset(), ) -> Callable[_P, _R]: r""" - Update globals of the ``wrapper`` function with the globals from the ``wrapped`` function. + Create a copy of ``wrapper``\, the copy's globals are updated with ``wrapped``\'s globals. For forwardrefs in command annotations, discord.py uses the ``__global__`` attribute of the function - to resolve their values, with decorators that replace the function this breaks because they have + to resolve their values. This breaks for decorators that replace the function because they have their own globals. - This function creates a new function functionally identical to ``wrapper``\, which has the globals replaced with - a merge of ``wrapped``\s globals and the ``wrapper``\s globals. - .. warning:: - This function captures the state of ``wrapped``\'s module's globals when it's called, + This function captures the state of ``wrapped``\'s module's globals when it's called; changes won't be reflected in the new function's globals. Args: diff --git a/docs/utils.py b/docs/utils.py index a4662ba4..9d299ebf 100644 --- a/docs/utils.py +++ b/docs/utils.py @@ -107,7 +107,7 @@ def _global_assign_pos(ast_: NodeWithBody, name: str) -> typing.Union[tuple[int, """ Find the first instance where the `name` global is defined in `ast_`. - Top level assignments, and assignments nested in top level ifs are checked. + Check top-level assignments and assignments nested in top-level if blocks. """ for ast_obj in ast_.body: if isinstance(ast_obj, ast.Assign): -- cgit v1.2.3 From ed9890abd8c07d6f414e273139e8715f3917b7fc Mon Sep 17 00:00:00 2001 From: Numerlor Date: Sun, 18 Sep 2022 20:41:47 +0200 Subject: use paramspec from typing the package now requires python 3.10 --- botcore/utils/cooldown.py | 9 +++------ botcore/utils/function.py | 3 +-- 2 files changed, 4 insertions(+), 8 deletions(-) (limited to 'botcore/utils/function.py') diff --git a/botcore/utils/cooldown.py b/botcore/utils/cooldown.py index ee65033d..5fb974e2 100644 --- a/botcore/utils/cooldown.py +++ b/botcore/utils/cooldown.py @@ -25,14 +25,11 @@ _ArgsList = list[object] _HashableArgsTuple = tuple[Hashable, ...] if typing.TYPE_CHECKING: - from botcore import BotBase import typing_extensions - P = typing_extensions.ParamSpec("P") - P.__constraints__ = () -else: - P = typing.TypeVar("P") - """The command's signature.""" + from botcore import BotBase +P = typing.ParamSpec("P") +"""The command's signature.""" R = typing.TypeVar("R") """The command's return value.""" diff --git a/botcore/utils/function.py b/botcore/utils/function.py index 0e90d4c5..d89163ec 100644 --- a/botcore/utils/function.py +++ b/botcore/utils/function.py @@ -11,8 +11,7 @@ __all__ = ["command_wraps", "GlobalNameConflictError", "update_wrapper_globals"] if typing.TYPE_CHECKING: - import typing_extensions - _P = typing_extensions.ParamSpec("_P") + _P = typing.ParamSpec("_P") _R = typing.TypeVar("_R") -- cgit v1.2.3