Actually ensure the attribute access is late...

Author: mikeshardmind

src/async_utils/_paramkey.py

@@ -20,10 +20,10 @@
 
 from collections.abc import Hashable
 
-from ._typings import Any, Final, final
+from . import _typings as t
 
 
-@final
+@t.final
 class _HK:
     __slots__ = ("_hashvalue", "_tup")
 
@@ -40,11 +40,11 @@ def __eq__(self, other: object) -> bool:
         return self._tup == other._tup
 
 
-_marker: Final[tuple[object]] = (object(),)
+_marker: t.Final[tuple[object]] = (object(),)
 
 
-def make_key(args: tuple[Any, ...], kwds: dict[Any, Any]) -> Hashable:
-    key: tuple[Any, ...] = args
+def make_key(args: tuple[t.Any, ...], kwds: dict[t.Any, t.Any]) -> Hashable:
+    key: tuple[t.Any, ...] = args
     if kwds:
         key += _marker
         for item in kwds.items():

src/async_utils/_typings.py

@@ -12,6 +12,17 @@
 #   See the License for the specific language governing permissions and
 #   limitations under the License.
 
+"""Shim for typing- and annotation-related symbols to avoid runtime dependencies on `typing` or `typing-extensions`.
+
+A warning for annotation-related symbols: Do not directly import them from this module
+(e.g. `from ._typing_compat import Any`)! Doing so will trigger the module-level `__getattr__`, causing `typing` to
+get imported. Instead, import the module and use symbols via attribute access as needed
+(e.g. `from . import _typing_compat [as _t]`). To avoid those symbols being evaluated at runtime, which would also cause
+`typing` to get imported, make sure to put `from __future__ import annotations` at the top of the module.
+"""
+# comment above
+# taken verbatim from quote from https://github.com/Sachaa-Thanasius in discord
+# cause I forget otherwise too.
 
 from __future__ import annotations
 
@@ -26,7 +37,6 @@ def final(f):  # noqa: ANN001
 
 
 def __getattr__(name: str):
-
     if name in {"Any", "Final", "Literal", "Self"}:
         import typing
 

src/async_utils/bg_tasks.py

@@ -18,9 +18,9 @@
 from collections.abc import Coroutine
 from contextvars import Context
 
-from ._typings import Any, Self
+from . import _typings as t
 
-type _CoroutineLike[T] = Coroutine[Any, Any, T]
+type _CoroutineLike[T] = Coroutine[t.Any, t.Any, T]
 
 
 __all__ = ["BGTasks"]
@@ -61,7 +61,7 @@ def create_task[T](
         t.add_done_callback(self._tasks.discard)
         return t
 
-    async def __aenter__(self: Self) -> Self:
+    async def __aenter__(self: t.Self) -> t.Self:
         return self
 
     async def __aexit__(self, *_dont_care: object) -> None:

src/async_utils/corofunc_cache.py

@@ -18,14 +18,14 @@
 from collections.abc import Awaitable, Callable, Coroutine, Hashable
 from functools import partial, wraps
 
+from . import _typings as t
 from ._paramkey import make_key
-from ._typings import Any
 from .lru import LRU
 
 __all__ = ("corocache", "lrucorocache")
 
 
-type CoroFunc[**P, R] = Callable[P, Coroutine[Any, Any, R]]
+type CoroFunc[**P, R] = Callable[P, Coroutine[t.Any, t.Any, R]]
 type CoroLike[**P, R] = Callable[P, Awaitable[R]]
 
 type Deco[**P, R] = Callable[[CoroLike[P, R]], CoroFunc[P, R]]

src/async_utils/priority_sem.py

@@ -22,7 +22,7 @@
 from collections.abc import Callable, Generator
 from contextlib import contextmanager
 
-from ._typings import Any, Self
+from . import _typings as t
 
 __all__ = ["PrioritySemaphore", "priority_context"]
 
@@ -34,7 +34,7 @@
 class PriorityWaiter(tuple[int, float, asyncio.Future[None]]):
     __slots__ = ()
 
-    def __new__(cls, priority: int, ts: float, future: asyncio.Future[None]) -> Self:
+    def __new__(cls, priority: int, ts: float, future: asyncio.Future[None]) -> t.Self:
         return super().__new__(cls, (priority, ts, future))
 
     @property
@@ -57,10 +57,10 @@ def cancelled(self) -> Callable[[], bool]:
     def done(self) -> Callable[[], bool]:
         return self.future.done
 
-    def __await__(self) -> Generator[Any, Any, None]:
+    def __await__(self) -> Generator[t.Any, t.Any, None]:
         return self.future.__await__()
 
-    def __lt__(self, other: Any) -> bool:
+    def __lt__(self, other: t.Any) -> bool:
         if not isinstance(other, PriorityWaiter):
             return NotImplemented
         return self[:2] < other[:2]
@@ -82,7 +82,7 @@ def priority_context(priority: int, /) -> Generator[None, None, None]:
         _priority.reset(token)
 
 
-_default: Any = object()
+_default: t.Any = object()
 
 
 class PrioritySemaphore:

src/async_utils/scheduler.py

@@ -18,11 +18,11 @@
 from functools import total_ordering
 from time import time
 
-from ._typings import Any, Self
+from . import _typings as t
 
 __all__ = ("Scheduler",)
 
-MISSING: Any = object()
+MISSING: t.Any = object()
 
 
 class CancellationToken:
@@ -76,7 +76,7 @@ def __init__(self, granularity: float, /) -> None:
         self.__tqueue = MISSING
         self.__l = MISSING
 
-    async def __aenter__(self) -> Self:
+    async def __aenter__(self) -> t.Self:
         self.__closed = False
         asyncio.get_running_loop()
 
@@ -94,7 +94,7 @@ async def __aenter__(self) -> Self:
     async def __aexit__(self, *_dont_care: object) -> None:
         self.__closed = True
 
-    def __aiter__(self) -> Self:
+    def __aiter__(self) -> t.Self:
         return self
 
     async def __anext__(self) -> T:

src/async_utils/sig_service.py

@@ -22,19 +22,19 @@
 from collections.abc import Callable
 from types import FrameType
 
-from ._typings import Any, Final, Literal
+from . import _typings as t
 
 __all__ = ["SignalService", "SpecialExit"]
 
-type SignalCallback = Callable[[signal.Signals | SpecialExit], Any]
-type StartStopCall = Callable[[], Any]
-type _HTC = Callable[[int, FrameType | None], Any]
+type SignalCallback = Callable[[signal.Signals | SpecialExit], t.Any]
+type StartStopCall = Callable[[], t.Any]
+type _HTC = Callable[[int, FrameType | None], t.Any]
 type _HANDLER = _HTC | int | signal.Handlers | None
 
-type HandleableSignals = Literal["SIGINT", "SIGTERM", "SIGBREAK", "SIGHUP"]
+type HandleableSignals = t.Literal["SIGINT", "SIGTERM", "SIGBREAK", "SIGHUP"]
 type SignalTuple = tuple[HandleableSignals, ...]
 
-default_handled: Final = "SIGINT", "SIGTERM", "SIGBREAK"
+default_handled: t.Final = "SIGINT", "SIGTERM", "SIGBREAK"
 
 
 class SpecialExit(enum.IntEnum):

src/async_utils/task_cache.py

@@ -19,15 +19,15 @@
 from collections.abc import Callable, Coroutine, Hashable
 from functools import partial, wraps
 
+from . import _typings as t
 from ._paramkey import make_key
-from ._typings import Any
 from .lru import LRU
 
 __all__ = ("lrutaskcache", "taskcache")
 
 
 # Use below doesn't accept non-task Futures, so can't accept general awaitables
-type CoroFunc[**P, R] = Callable[P, Coroutine[Any, Any, R]]
+type CoroFunc[**P, R] = Callable[P, Coroutine[t.Any, t.Any, R]]
 type TaskFunc[**P, R] = CoroFunc[P, R] | Callable[P, asyncio.Task[R]]
 type TaskCoroFunc[**P, R] = CoroFunc[P, R] | TaskFunc[P, R]
 

src/async_utils/waterfall.py

@@ -19,12 +19,12 @@
 import time
 from collections.abc import Callable, Coroutine, Sequence
 
-from ._typings import Any
+from . import _typings as t
 
 __all__ = ("Waterfall",)
 
-type CBT[T] = Callable[[Sequence[T]], Coroutine[Any, Any, Any]]
-type NC = Coroutine[Any, Any, None]
+type CBT[T] = Callable[[Sequence[T]], Coroutine[t.Any, t.Any, t.Any]]
+type NC = Coroutine[t.Any, t.Any, None]
 
 
 class Waterfall[T]: