Ensure typing module is only imported if annotations are introspected

Author: mikeshardmind

pyproject.toml

@@ -36,6 +36,8 @@ select = [
     "TRY", "UP", "YTT"
 ]
 
+typing-modules = ["async_utils._typings"]
+
 ignore = [
     "ANN202", # implied return fine sometimes
     "ANN401",  # Any is the correct type in some cases

src/async_utils/__init__.py

@@ -9,7 +9,7 @@
 __author__ = "Michael Hall"
 __license__ = "Apache-2.0"
 __copyright__ = "Copyright 2020-Present Michael Hall"
-__version__ = "2024.12.27"
+__version__ = "2025.01.06"
 
 import os
 import sys

src/async_utils/_paramkey.py

@@ -1,11 +1,26 @@
+#   Copyright 2020-present Michael Hall
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+
 # This used to include CPython code, some minor performance losses have been
 # taken to not tightly include upstream code
 
 
 from __future__ import annotations
 
 from collections.abc import Hashable
-from typing import Any, Final, final
+
+from ._typings import Any, Final, final
 
 
 @final

src/async_utils/_typings.py

@@ -0,0 +1,41 @@
+#   Copyright 2020-present Michael Hall
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+
+
+from __future__ import annotations
+
+TYPE_CHECKING = False
+
+if TYPE_CHECKING:
+    from typing import Any, Final, Literal, Self, final
+else:
+
+    def final(f):  # noqa: ANN001
+        return f
+
+
+def __getattr__(name: str):
+    if name == "final":  # this one actually executes at runtime
+        return final
+
+    if name in {"Any", "Final", "Literal", "Self"}:
+        import typing
+
+        return getattr(typing, name)
+
+    msg = f"module {__name__!r} has no attribute {name!r}"
+    raise AttributeError(msg)
+
+
+__all__ = ["Any", "Final", "Literal", "Self", "final"]

src/async_utils/bg_tasks.py

@@ -17,7 +17,8 @@
 import asyncio
 from collections.abc import Coroutine
 from contextvars import Context
-from typing import Any, Self
+
+from ._typings import Any, Self
 
 type _CoroutineLike[T] = Coroutine[Any, Any, T]
 

src/async_utils/corofunc_cache.py

@@ -17,9 +17,9 @@
 import asyncio
 from collections.abc import Awaitable, Callable, Coroutine, Hashable
 from functools import partial, wraps
-from typing import Any
 
 from ._paramkey import make_key
+from ._typings import Any
 from .lru import LRU
 
 __all__ = ("corocache", "lrucorocache")

src/async_utils/priority_sem.py

@@ -21,7 +21,8 @@
 import threading
 from collections.abc import Callable, Generator
 from contextlib import contextmanager
-from typing import Any, NamedTuple
+
+from ._typings import Any, Self
 
 __all__ = ["PrioritySemaphore", "priority_context"]
 
@@ -30,10 +31,23 @@
 _priority: contextvars.ContextVar[int] = contextvars.ContextVar("_priority", default=0)
 
 
-class PriorityWaiter(NamedTuple):
-    priority: int
-    ts: float
-    future: asyncio.Future[None]
+class PriorityWaiter(tuple[int, float, asyncio.Future[None]]):
+    __slots__ = ()
+
+    def __new__(cls, priority: int, ts: float, future: asyncio.Future[None]) -> Self:
+        return super().__new__(cls, (priority, ts, future))
+
+    @property
+    def priority(self) -> int:
+        return self[0]
+
+    @property
+    def ts(self) -> float:
+        return self[1]
+
+    @property
+    def future(self) -> asyncio.Future[None]:
+        return self[2]
 
     @property
     def cancelled(self) -> Callable[[], bool]:
@@ -49,7 +63,7 @@ def __await__(self) -> Generator[Any, Any, None]:
     def __lt__(self, other: Any) -> bool:
         if not isinstance(other, PriorityWaiter):
             return NotImplemented
-        return (self.priority, self.ts) < (other.priority, other.ts)
+        return self[:2] < other[:2]
 
 
 @contextmanager

src/async_utils/scheduler.py

@@ -17,7 +17,8 @@
 import asyncio
 from functools import total_ordering
 from time import time
-from typing import Any, Self
+
+from ._typings import Any, Self
 
 __all__ = ("Scheduler",)
 

src/async_utils/sig_service.py

@@ -21,7 +21,8 @@
 import sys
 from collections.abc import Callable
 from types import FrameType
-from typing import Any, Final, Literal
+
+from ._typings import Any, Final, Literal
 
 __all__ = ["SignalService", "SpecialExit"]
 

src/async_utils/task_cache.py

@@ -18,9 +18,9 @@
 import inspect
 from collections.abc import Callable, Coroutine, Hashable
 from functools import partial, wraps
-from typing import Any
 
 from ._paramkey import make_key
+from ._typings import Any
 from .lru import LRU
 
 __all__ = ("lrutaskcache", "taskcache")