Switch to exterior union for maybe awaitables to disallow dynamic shifts

tanjun/abc.py

@@ -78,29 +78,23 @@
 if typing.TYPE_CHECKING:
  _T = typing.TypeVar("_T")
 
- __CheckSig = collections.Callable[
- typing_extensions.Concatenate[_ContextT_contra, _P], typing.Union[collections.Awaitable[bool], bool]
- ]
+ _MaybeAwaitableSig = typing.Union[collections.Callable[_P, _T], collections.Callable[_P, collections.Awaitable[_T]]]
+
+ __CheckSig = _MaybeAwaitableSig[typing_extensions.Concatenate[_ContextT_contra, _P], bool]
  _CheckSig = __CheckSig[_ContextT_contra, ...]
 
- __HookSig = collections.Callable[
- typing_extensions.Concatenate[_ContextT_contra, _P], typing.Union[collections.Awaitable[None], None]
- ]
+ __HookSig = _MaybeAwaitableSig[typing_extensions.Concatenate[_ContextT_contra, _P], None]
  _HookSig = __HookSig[_ContextT_contra, ...]
  _MessageHookSigT = typing.TypeVar("_MessageHookSigT", bound="_HookSig[MessageContext]")
  _SlashHookSigT = typing.TypeVar("_SlashHookSigT", bound="_HookSig[SlashContext]")
 
- __ParserHookSig = collections.Callable[
- typing_extensions.Concatenate[_ContextT_contra, errors.ParserError, _P],
- typing.Union[collections.Awaitable[None], None],
- ]
+ __ParserHookSig = _MaybeAwaitableSig[typing_extensions.Concatenate[_ContextT_contra, errors.ParserError, _P], None]
  _ParserHookSig = __ParserHookSig[_ContextT_contra, ...]
  _MessageParserHookSigT = typing.TypeVar("_MessageParserHookSigT", bound="_ParserHookSig[MessageContext]")
  _SlashParserHookSigT = typing.TypeVar("_SlashParserHookSigT", bound="_ParserHookSig[SlashContext]")
 
- __ErrorHookSig = collections.Callable[
- typing_extensions.Concatenate[_ContextT_contra, Exception, _P],
- typing.Union[collections.Awaitable[typing.Optional[bool]], typing.Optional[bool]],
+ __ErrorHookSig = _MaybeAwaitableSig[
+ typing_extensions.Concatenate[_ContextT_contra, Exception, _P], typing.Optional[bool]
  ]
  _ErrorHookSig = __ErrorHookSig[_ContextT_contra, ...]
  _MessageErrorHookSigT = typing.TypeVar("_MessageErrorHookSigT", bound="_ErrorHookSig[MessageContext]")
@@ -116,7 +110,7 @@
  _BaseSlashCommandT = typing.TypeVar("_BaseSlashCommandT", bound="BaseSlashCommand")
  _MessageCommandT = typing.TypeVar("_MessageCommandT", bound="MessageCommand[typing.Any]")
 
- _MetaEventSig = collections.Callable[..., typing.Union[collections.Awaitable[None], None]]
+ _MetaEventSig = _MaybeAwaitableSig[..., None]
  _MetaEventSigT = typing.TypeVar("_MetaEventSigT", bound="_MetaEventSig")
 
  __CommandSig = collections.Callable[

tanjun/checks.py

@@ -74,10 +74,10 @@
 
 if typing.TYPE_CHECKING:
  _P = typing_extensions.ParamSpec("_P")
+ _T = typing.TypeVar("_V")
 
- __CheckSig = collections.Callable[
- typing_extensions.Concatenate[_ContextT_contra, _P], typing.Union[collections.Awaitable[bool], bool]
- ]
+ _MaybeAwaitableSig = typing.Union[collections.Callable[_P, _T], collections.Callable[_P, collections.Awaitable[_T]]]
+ __CheckSig = _MaybeAwaitableSig[typing_extensions.Concatenate[_ContextT_contra, _P], bool]
  _CheckSig = __CheckSig[_ContextT_contra, ...]
 
  _CommandT = typing.TypeVar("_CommandT", bound="tanjun_abc.ExecutableCommand[typing.Any]")

tanjun/clients.py

@@ -97,10 +97,11 @@ def __call__(
  raise NotImplementedError
 
  _P = typing_extensions.ParamSpec("_P")
+ _T = typing.TypeVar("_T")
 
- __CheckSig = collections.Callable[
- typing_extensions.Concatenate[tanjun_abc.Context, _P], typing.Union[collections.Awaitable[bool], bool]
- ]
+ _MaybeAwaitableSig = typing.Union[collections.Callable[_P, _T], collections.Callable[_P, collections.Awaitable[_T]]]
+
+ __CheckSig = _MaybeAwaitableSig[typing_extensions.Concatenate[tanjun_abc.Context, _P], bool]
  _CheckSig = __CheckSig[...]
  _CheckSigT = typing.TypeVar("_CheckSigT", bound=_CheckSig)
 

tanjun/commands.py

@@ -89,13 +89,13 @@
  from . import parsing
 
  _T = typing.TypeVar("_T")
+ _MaybeAwaitableSig = typing.Union[collections.Callable[_P, _T], collections.Callable[_P, collections.Awaitable[_T]]]
+
  _AnyMessageCommandT = typing.TypeVar("_AnyMessageCommandT", bound=abc.MessageCommand[typing.Any])
  _AnyBaseSlashCommandT = typing.TypeVar("_AnyBaseSlashCommandT", bound="abc.BaseSlashCommand")
  _SlashCommandT = typing.TypeVar("_SlashCommandT", bound="SlashCommand[typing.Any]")
 
- __CheckSig = collections.Callable[
- typing_extensions.Concatenate[_ContextT_contra, _P], typing.Union[collections.Awaitable[bool], bool]
- ]
+ __CheckSig = _MaybeAwaitableSig[typing_extensions.Concatenate[_ContextT_contra, _P], bool]
  _CheckSig = __CheckSig[_ContextT_contra, ...]
  _MessageCheckSigT = typing.TypeVar("_MessageCheckSigT", bound=_CheckSig[abc.MessageContext])
  _SlashCheckSigT = typing.TypeVar("_SlashCheckSigT", bound=_CheckSig[abc.SlashContext])
@@ -107,9 +107,7 @@
  abc.SlashCommand[_CallbackSigT],
  ]
 
- __ConverterSig = collections.Callable[
- typing_extensions.Concatenate[_T, _P], typing.Union[collections.Awaitable[typing.Any], typing.Any]
- ]
+ __ConverterSig = _MaybeAwaitableSig[typing_extensions.Concatenate[_T, _P], typing.Any]
  _ConverterSig = __ConverterSig[_T, ...]
 
 

tanjun/components.py

@@ -58,14 +58,15 @@
  from . import schedules
 
  _P = typing_extensions.ParamSpec("_P")
+ _T = typing.TypeVar("_T")
+
+ _MaybeAwaitableSig = typing.Union[collections.Callable[_P, _T], collections.Callable[_P, collections.Awaitable[_T]]]
 
  _ContextT_contra = typing.TypeVar("_ContextT_contra", bound="tanjun_abc.Context", contravariant=True)
  _CommandT = typing.TypeVar("_CommandT", bound="tanjun_abc.ExecutableCommand[typing.Any]")
  _ScheduleT = typing.TypeVar("_ScheduleT", bound=schedules.AbstractSchedule)
 
- __CheckSig = collections.Callable[
- typing_extensions.Concatenate[_ContextT_contra, _P], typing.Union[collections.Awaitable[bool], bool]
- ]
+ __CheckSig = _MaybeAwaitableSig[typing_extensions.Concatenate[_ContextT_contra, _P], bool]
  _CheckSig = __CheckSig[_ContextT_contra, ...]
  _CheckSigT = typing.TypeVar("_CheckSigT", bound=_CheckSig[typing.Any])
 
@@ -79,13 +80,13 @@
  _BaseSlashCommandT = typing.TypeVar("_BaseSlashCommandT", bound="tanjun_abc.BaseSlashCommand")
  _MessageCommandT = typing.TypeVar("_MessageCommandT", bound="tanjun_abc.MessageCommand[typing.Any]")
 
- _OnCallbackSig = collections.Callable[..., typing.Union[typing.Awaitable[None], None]]
+ _OnCallbackSig = _MaybeAwaitableSig[..., None]
  _OnCallbackSigT = typing.TypeVar("_OnCallbackSigT", bound=_OnCallbackSig)
 
  # This errors on earlier 3.9 releases when not quotes cause dumb handling of the [_CommandT] list
  _WithCommandReturnSig = typing.Union[_CommandT, "collections.Callable[[_CommandT], _CommandT]"]
 
- _MetaEventSig = collections.Callable[..., typing.Union[collections.Awaitable[None], None]]
+ _MetaEventSig = _MaybeAwaitableSig[..., None]
  _MetaEventSigT = typing.TypeVar("_MetaEventSigT", bound="_MetaEventSig")
 
 

tanjun/conversion.py

@@ -90,10 +90,11 @@
  import typing_extensions
 
  _P = typing_extensions.ParamSpec("_P")
-
- __ConverterSig = collections.Callable[
- typing_extensions.Concatenate[str, _P], typing.Union[collections.Awaitable[typing.Any], typing.Any]
+ _MaybeAwaitableSig = typing.Union[
+ collections.Callable[_P, "_ValueT"], collections.Callable[_P, collections.Awaitable["_ValueT"]]
  ]
+
+ __ConverterSig = _MaybeAwaitableSig[typing_extensions.Concatenate[str, _P], typing.Any]
  _ConverterSig = __ConverterSig[...]
 
 

tanjun/dependencies/data.py

@@ -47,6 +47,12 @@
 
  import typing_extensions
 
+ _P = typing_extensions.ParamSpec("_P")
+ _MaybeAwaitableSig = typing.Union[
+ collections.Callable[_P, "_T"], collections.Callable[_P, collections.Awaitable["_T"]]
+ ]
+
+
 _T = typing.TypeVar("_T")
 
 
@@ -59,7 +65,7 @@ class LazyConstant(typing.Generic[_T]):
 
  __slots__ = ("_callback", "_lock", "_value")
 
- def __init__(self, callback: collections.Callable[..., typing.Union[collections.Awaitable[_T], _T]], /) -> None:
+ def __init__(self, callback: _MaybeAwaitableSig[..., _T], /) -> None:
  """Initiate a new lazy constant.
 
  Parameters
@@ -218,7 +224,7 @@ class _CacheCallback(typing.Generic[_T]):
 
  def __init__(
  self,
- callback: collections.Callable[..., typing.Union[collections.Awaitable[_T], _T]],
+ callback: _MaybeAwaitableSig[..., _T],
  /,
  *,
  expire_after: typing.Union[int, float, datetime.timedelta, None],
@@ -271,7 +277,7 @@ async def __call__(
 
 
 def cache_callback(
- callback: collections.Callable[..., typing.Union[collections.Awaitable[_T], _T]],
+ callback: _MaybeAwaitableSig[..., _T],
  /,
  *,
  expire_after: typing.Union[int, float, datetime.timedelta, None] = None,
@@ -309,7 +315,7 @@ def cache_callback(
 
 
 def cached_inject(
- callback: collections.Callable[..., typing.Union[collections.Awaitable[_T], _T]],
+ callback: _MaybeAwaitableSig[..., _T],
  /,
  *,
  expire_after: typing.Union[float, int, datetime.timedelta, None] = None,

tanjun/hooks.py

@@ -49,30 +49,26 @@
 if typing.TypeVar:
  import typing_extensions
 
+ _T = typing.TypeVar("_T")
  _CommandT = typing.TypeVar("_CommandT", bound=abc.ExecutableCommand[typing.Any])
  _BaseSlashCommandT = typing.TypeVar("_BaseSlashCommandT", bound=abc.BaseSlashCommand)
  _MessageCommandT = typing.TypeVar("_MessageCommandT", bound=abc.MessageCommand[typing.Any])
 
  _P = typing_extensions.ParamSpec("_P")
 
- __HookSig = collections.Callable[
- typing_extensions.Concatenate[_ContextT_contra, _P], typing.Union[collections.Awaitable[None], None]
- ]
+ _MaybeAwaitableSig = typing.Union[collections.Callable[_P, _T], collections.Callable[_P, collections.Awaitable[_T]]]
+ __HookSig = _MaybeAwaitableSig[typing_extensions.Concatenate[_ContextT_contra, _P], None]
  _HookSig = __HookSig[_ContextT_contra, ...]
  _MessageHookSigT = typing.TypeVar("_MessageHookSigT", bound="_HookSig[abc.MessageContext]")
  _SlashHookSigT = typing.TypeVar("_SlashHookSigT", bound="_HookSig[abc.SlashContext]")
 
- __ParserHookSig = collections.Callable[
- typing_extensions.Concatenate[_ContextT_contra, errors.ParserError, _P],
- typing.Union[collections.Awaitable[None], None],
- ]
+ __ParserHookSig = _MaybeAwaitableSig[typing_extensions.Concatenate[_ContextT_contra, errors.ParserError, _P], None]
  _ParserHookSig = __ParserHookSig[_ContextT_contra, ...]
  _MessageParserHookSigT = typing.TypeVar("_MessageParserHookSigT", bound="_ParserHookSig[abc.MessageContext]")
  _SlashParserHookSigT = typing.TypeVar("_SlashParserHookSigT", bound="_ParserHookSig[abc.SlashContext]")
 
- __ErrorHookSig = collections.Callable[
- typing_extensions.Concatenate[_ContextT_contra, Exception, _P],
- typing.Union[collections.Awaitable[typing.Optional[bool]], typing.Optional[bool]],
+ __ErrorHookSig = _MaybeAwaitableSig[
+ typing_extensions.Concatenate[_ContextT_contra, Exception, _P], typing.Optional[bool]
  ]
  _ErrorHookSig = __ErrorHookSig[_ContextT_contra, ...]
  _MessageErrorHookSigT = typing.TypeVar("_MessageErrorHookSigT", bound="_ErrorHookSig[abc.MessageContext]")

tanjun/injecting.py

@@ -66,7 +66,10 @@
 
  from . import abc as tanjun_abc
 
- _CallbackSig = collections.Callable[..., typing.Union[collections.Awaitable[_T], _T]]
+ _P = typing_extensions.ParamSpec("_P")
+
+ _MaybeAwaitableSig = typing.Union[collections.Callable[_P, _T], collections.Callable[_P, collections.Awaitable[_T]]]
+ _CallbackSig = _MaybeAwaitableSig[..., _T]
 
 
 class Undefined:

tanjun/injecting.pyi

@@ -57,7 +57,9 @@ import typing_extensions
 from . import abc as tanjun_abc
 
 _T = typing.TypeVar("_T")
-_CallbackSig = collections.Callable[..., typing.Union[typing.Awaitable[_T], _T]]
+_P = typing_extensions.ParamSpec("_P")
+_MaybeAwaitableSig = typing.Union[collections.Callable[_P, _T], collections.Callable[_P, collections.Awaitable[_T]]]
+_CallbackSig = _MaybeAwaitableSig[..., _T]
 
 class Undefined:
  def __bool__(self) -> typing.Literal[False]: ...

tanjun/parsing.py

@@ -69,14 +69,13 @@
  _ParseableProtoT = typing.TypeVar("_ParseableProtoT", bound="ParseableProto")
 
  _P = typing_extensions.ParamSpec("_P")
+ _MaybeAwaitableSig = typing.Union[collections.Callable[_P, _T], collections.Callable[_P, collections.Awaitable[_T]]]
  __CommandCallbackSig = typing.Callable[
  typing_extensions.Concatenate[tanjun_abc.MessageContext, _P], collections.Awaitable[None]
  ]
  _CommandCallbackSig = __CommandCallbackSig[...]
 
- __ConverterSig = collections.Callable[
- typing_extensions.Concatenate[str, _P], typing.Union[collections.Awaitable[typing.Any], typing.Any]
- ]
+ __ConverterSig = _MaybeAwaitableSig[typing_extensions.Concatenate[str, _P], typing.Any]
  _ConverterSig = __ConverterSig[...]