From ebd0d7ecc9725513b6faa8ad62e7e0eb31f64d86 Mon Sep 17 00:00:00 2001 From: FasterSpeeding Date: Sat, 16 Nov 2024 22:52:15 +0000 Subject: [PATCH] Move away from runtime_checkable (#968) Co-authored-by: always-on-duty[bot] <120557446+always-on-duty[bot]@users.noreply.github.com> --- CHANGELOG.md | 4 ++++ tanjun/schedules.py | 16 +++++++++++++--- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a39c12e5..f398bbf11 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Fixed +- Moved away from using `typing.runtime_checkable` as this is unreliable in + newer Python versions. + ## [2.17.6] - 2024-10-07 ### Changed - Support Python 3.13. diff --git a/tanjun/schedules.py b/tanjun/schedules.py index 7a20db546..268025840 100644 --- a/tanjun/schedules.py +++ b/tanjun/schedules.py @@ -47,6 +47,7 @@ from . import components if typing.TYPE_CHECKING: + import typing_extensions from alluka import abc as alluka from typing_extensions import Self @@ -132,12 +133,21 @@ async def stop(self) -> None: """ -@typing.runtime_checkable class _ComponentProto(typing.Protocol): def add_schedule(self, schedule: AbstractSchedule, /) -> typing.Any: raise NotImplementedError +def _is_component_proto(value: typing.Any, /) -> typing_extensions.TypeGuard[_ComponentProto]: + try: + value.add_schedule + + except AttributeError: + return False + + return True + + def as_interval( interval: typing.Union[int, float, datetime.timedelta], /, @@ -298,7 +308,7 @@ def copy(self) -> Self: def load_into_component(self, component: tanjun.Component, /) -> None: # <>. - if isinstance(component, _ComponentProto): + if _is_component_proto(component): component.add_schedule(self) def set_start_callback(self, callback: _CallbackSig, /) -> Self: @@ -1083,7 +1093,7 @@ async def _loop(self, client: alluka.Client, /) -> None: def load_into_component(self, component: tanjun.Component, /) -> None: # <>. - if isinstance(component, _ComponentProto): + if _is_component_proto(component): component.add_schedule(self) def start(self, client: alluka.Client, /, *, loop: typing.Optional[asyncio.AbstractEventLoop] = None) -> None: