remove 2 uses of Any as sentinel value type

Author: mikeshardmind

src/async_utils/__init__.py

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

src/async_utils/priority_sem.py

@@ -80,9 +80,6 @@ def priority_context(priority: int, /) -> Generator[None, None, None]:
         _priority.reset(token)
 
 
-_default: t.Any = object()
-
-
 class PrioritySemaphore:
     """A Semaphore with priority-based aquisition ordering.
 
@@ -154,9 +151,7 @@ async def __aenter__(self) -> None:
     async def __aexit__(self, *dont_care: object) -> None:
         self.__release()
 
-    async def __acquire(self, priority: int = _default) -> bool:
-        if priority is _default:
-            priority = _priority.get()
+    async def __acquire(self, priority: int, /) -> bool:
         if not self.__locked():
             self._value -= 1
             return True

src/async_utils/scheduler.py

@@ -22,8 +22,6 @@
 
 __all__ = ("CancellationToken", "Scheduler")
 
-MISSING: t.Any = object()
-
 
 class CancellationToken:
     """An object to use for cancelation of a task.
@@ -72,34 +70,17 @@ def __init_subclass__(cls) -> t.Never:
 
     __final__ = True
 
-    __tasks: dict[CancellationToken, _Task[T]]
-    __tqueue: asyncio.PriorityQueue[_Task[T]]
-    __closed: bool
-    __l: asyncio.Lock
-    __granularity: float
-
     __slots__ = ("__closed", "__granularity", "__l", "__tasks", "__tqueue")
 
     def __init__(self, granularity: float, /) -> None:
-        self.__granularity = granularity
-        self.__closed = MISSING
-        self.__tasks = MISSING
-        self.__tqueue = MISSING
-        self.__l = MISSING
-
-    async def __aenter__(self) -> t.Self:
-        self.__closed = False
-        asyncio.get_running_loop()
-
-        # lock is only needeed on modifying or removing tasks
-        # insertion is not guarded and only racy in the order of emitted events
-        # when inserting a task that is scheduled in the past
-        # or within 1 full iteration of pending tasks on the event loop
-        # (generally, ms range (subsecond), depending on application)
+        self.__granularity: float = granularity
+        self.__closed: bool = False
+        self.__tasks: dict[CancellationToken, _Task[T]] = {}
+        # PYUPGRADE: check: 3.15; relies on asyncio.Lock & Queues not eagerly binding to an event loop.
         self.__l = asyncio.Lock()
-        self.__tasks = {}
-        self.__tqueue = asyncio.PriorityQueue()
+        self.__tqueue: asyncio.PriorityQueue[_Task[T]] = asyncio.PriorityQueue()
 
+    async def __aenter__(self) -> t.Self:
         return self
 
     async def __aexit__(self, *_dont_care: object) -> None: