Enabled Event and CapacityLimiter to be instantiated outside an event loop

docs/versionhistory.rst

@@ -8,6 +8,8 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 - Add support for ``byte``-based paths in ``connect_unix``, ``create_unix_listeners``,
   ``create_unix_datagram_socket``, and ``create_connected_unix_datagram_socket``. (PR by
   Lura Skye)
+- Enabled the ``Event`` and ``CapacityLimiter`` classes to be instantiated outside an
+  event loop thread
 - Fixed adjusting the total number of tokens in a ``CapacityLimiter`` on asyncio failing
   to wake up tasks waiting to acquire the limiter in certain edge cases (fixed with help
   from Egor Blagov)

src/anyio/_core/_synchronization.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import math
 from collections import deque
 from dataclasses import dataclass
 from types import TracebackType
@@ -76,7 +77,7 @@ class SemaphoreStatistics:
 
 class Event:
     def __new__(cls) -> Event:
-        return get_async_backend().create_event()
+        return EventAdapter()
 
     def set(self) -> None:
         """Set the flag, notifying all listeners."""
@@ -101,6 +102,32 @@ def statistics(self) -> EventStatistics:
         raise NotImplementedError
 
 
+class EventAdapter(Event):
+    _internal_event: Event | None = None
+
+    def __new__(cls) -> EventAdapter:
+        return object.__new__(cls)
+
+    @property
+    def _event(self) -> Event:
+        if self._internal_event is None:
+            self._internal_event = get_async_backend().create_event()
+
+        return self._internal_event
+
+    def set(self) -> None:
+        self._event.set()
+
+    def is_set(self) -> bool:
+        return self._internal_event is not None and self._internal_event.is_set()
+
+    async def wait(self) -> None:
+        await self._event.wait()
+
+    def statistics(self) -> EventStatistics:
+        return self._event.statistics()
+
+
 class Lock:
     _owner_task: TaskInfo | None = None
 
@@ -372,8 +399,8 @@ def statistics(self) -> SemaphoreStatistics:
 
 
 class CapacityLimiter:
-    def __new__(cls, total_tokens: float) -> CapacityLimiter:
-        return get_async_backend().create_capacity_limiter(total_tokens)
+    def __new__(cls, total_tokens: int) -> CapacityLimiter:
+        return CapacityLimiterAdapter(total_tokens)
 
     async def __aenter__(self) -> None:
         raise NotImplementedError
@@ -482,6 +509,78 @@ def statistics(self) -> CapacityLimiterStatistics:
         raise NotImplementedError
 
 
+class CapacityLimiterAdapter(CapacityLimiter):
+    _internal_limiter: CapacityLimiter | None = None
+
+    def __new__(cls, total_tokens: int) -> CapacityLimiterAdapter:
+        return object.__new__(cls)
+
+    def __init__(self, total_tokens: int) -> None:
+        if not isinstance(total_tokens, int) and total_tokens is not math.inf:
+            raise TypeError("total_tokens must be an int or math.inf")
+        elif total_tokens < 1:
+            raise ValueError("total_tokens must be >= 1")
+
+        self._total_tokens = total_tokens
+
+    @property
+    def _limiter(self) -> CapacityLimiter:
+        if self._internal_limiter is None:
+            self._internal_limiter = get_async_backend().create_capacity_limiter(
+                self._total_tokens
+            )
+
+        return self._internal_limiter
+
+    async def __aenter__(self) -> None:
+        await self._limiter.__aenter__()
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> bool | None:
+        return await self._limiter.__aexit__(exc_type, exc_val, exc_tb)
+
+    @property
+    def total_tokens(self) -> float:
+        return self._limiter.total_tokens
+
+    @total_tokens.setter
+    def total_tokens(self, value: float) -> None:
+        self._limiter.total_tokens = value
+
+    @property
+    def borrowed_tokens(self) -> int:
+        return self._limiter.borrowed_tokens
+
+    @property
+    def available_tokens(self) -> float:
+        return self._limiter.available_tokens
+
+    def acquire_nowait(self) -> None:
+        self._limiter.acquire_nowait()
+
+    def acquire_on_behalf_of_nowait(self, borrower: object) -> None:
+        self._limiter.acquire_on_behalf_of_nowait(borrower)
+
+    async def acquire(self) -> None:
+        await self._limiter.acquire()
+
+    async def acquire_on_behalf_of(self, borrower: object) -> None:
+        await self._limiter.acquire_on_behalf_of(borrower)
+
+    def release(self) -> None:
+        self._limiter.release()
+
+    def release_on_behalf_of(self, borrower: object) -> None:
+        self._limiter.release_on_behalf_of(borrower)
+
+    def statistics(self) -> CapacityLimiterStatistics:
+        return self._limiter.statistics()
+
+
 class ResourceGuard:
     """
     A context manager for ensuring that a resource is only used by a single task at a

tests/test_synchronization.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+from typing import Any
 
 import pytest
 
@@ -13,6 +14,7 @@
     WouldBlock,
     create_task_group,
     fail_after,
+    run,
     to_thread,
     wait_all_tasks_blocked,
 )
@@ -208,6 +210,18 @@ async def waiter() -> None:
 
         assert event.statistics().tasks_waiting == 0
 
+    def test_instantiate_outside_event_loop(
+        self, anyio_backend_name: str, anyio_backend_options: dict[str, Any]
+    ) -> None:
+        async def use_event() -> None:
+            event.set()
+            await event.wait()
+
+        event = Event()
+        run(
+            use_event, backend=anyio_backend_name, backend_options=anyio_backend_options
+        )
+
 
 class TestCondition:
     async def test_contextmanager(self) -> None:
@@ -595,3 +609,17 @@ async def worker(entered_event: Event) -> None:
 
             # Allow all tasks to exit
             continue_event.set()
+
+    def test_instantiate_outside_event_loop(
+        self, anyio_backend_name: str, anyio_backend_options: dict[str, Any]
+    ) -> None:
+        async def use_limiter() -> None:
+            async with limiter:
+                pass
+
+        limiter = CapacityLimiter(1)
+        run(
+            use_limiter,
+            backend=anyio_backend_name,
+            backend_options=anyio_backend_options,
+        )