modify anyio_service.py

Author: Khwahish29

libp2p/tools/anyio_service/README.md

@@ -55,9 +55,9 @@ The implementation maintains the same public API as the previous async service i
 To migrate from the previous async service implementation:
 
 1. Update imports to use `libp2p.tools.anyio_service` instead of `libp2p.tools.async_service`
-2. No other code changes required - the API is fully compatible
+1. No other code changes required - the API is fully compatible
 
 ## Requirements
 
 - Python 3.7+
-- anyio library 
+- anyio library

libp2p/tools/anyio_service/__init__.py

@@ -1,6 +1,10 @@
 from .abc import (
     ServiceAPI,
 )
+from .anyio_service import (
+    AnyioManager,
+    background_anyio_service,
+)
 from .base import (
     Service,
     as_service,
@@ -9,10 +13,6 @@
     DaemonTaskExit,
     LifecycleError,
 )
-from .anyio_service import (
-    AnyioManager,
-    background_anyio_service,
-)
 
 __all__ = [
     "ServiceAPI",
@@ -22,4 +22,4 @@
     "LifecycleError",
     "AnyioManager",
     "background_anyio_service",
-] 
+]

libp2p/tools/anyio_service/abc.py

@@ -205,7 +205,11 @@ class InternalManagerAPI(ManagerAPI):
 
     @abstractmethod
     def run_task(
-        self, async_fn: AsyncFn, *args: Any, daemon: bool = False, name: Optional[str] = None
+        self,
+        async_fn: AsyncFn,
+        *args: Any,
+        daemon: bool = False,
+        name: Optional[str] = None
     ) -> None:
         """
         Run a task in the background.  If the function throws an exception it
@@ -217,7 +221,9 @@ def run_task(
         ...
 
     @abstractmethod
-    def run_daemon_task(self, async_fn: AsyncFn, *args: Any, name: Optional[str] = None) -> None:
+    def run_daemon_task(
+        self, async_fn: AsyncFn, *args: Any, name: Optional[str] = None
+    ) -> None:
         """
         Run a daemon task in the background.
 
@@ -247,4 +253,4 @@ def run_daemon_child_service(
 
         Equivalent to `run_child_service(..., daemon=True)`.
         """
-        ... 
+        ...

libp2p/tools/anyio_service/anyio_service.py

@@ -1,7 +1,5 @@
-import anyio
-import logging
-import sys
 from collections.abc import (
+    AsyncIterator,
     Awaitable,
     Callable,
 )
@@ -11,15 +9,17 @@
 from functools import (
     wraps,
 )
+import logging
+import sys
 from typing import (
     Any,
-    AsyncIterator,
     Optional,
     TypeVar,
     cast,
-    Union,
 )
 
+import anyio
+
 if sys.version_info >= (3, 11):
     from builtins import (
         ExceptionGroup,
@@ -52,18 +52,16 @@
 
 T = TypeVar("T", bound=Callable[..., Any])
 
+
 def external_api(func: T) -> T:
-    """
-    Decorator to mark a method as an external API that can be called from outside the service.
-    This ensures that the service is in the correct state before allowing the method to be called.
-    """
     @wraps(func)
     def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
         if not hasattr(self, "manager"):
             raise LifecycleError("Service has no manager")
         if not self.manager.is_running:
             raise LifecycleError("Service is not running")
         return func(self, *args, **kwargs)
+
     return cast(T, wrapper)
 
 
@@ -204,11 +202,11 @@ def __init__(self, service: ServiceAPI) -> None:
         self._total_task_count = 0
         self._done_task_count = 0
 
-        self._started = anyio.create_event()
-        self._cancelled = anyio.create_event()
-        self._finished = anyio.create_event()
+        self._started = anyio.Event()
+        self._cancelled = anyio.Event()
+        self._finished = anyio.Event()
 
-        self._run_lock = anyio.create_lock()
+        self._run_lock = anyio.Lock()
         self._task_group: Optional[anyio.abc.TaskGroup] = None
 
     def __str__(self) -> str:
@@ -238,7 +236,10 @@ async def stop(self) -> None:
             await self.wait_finished()
 
     def run_daemon_task(
-        self, async_fn: Callable[..., Awaitable[Any]], *args: Any, name: Optional[str] = None
+        self,
+        async_fn: Callable[..., Awaitable[Any]],
+        *args: Any,
+        name: Optional[str] = None,
     ) -> None:
         self.run_task(async_fn, *args, daemon=True, name=name)
 
@@ -255,7 +256,9 @@ def stats(self) -> Stats:
             tasks=TaskStats(total_count=total_count, finished_count=finished_count)
         )
 
-    def _add_child_task(self, parent: Optional[TaskWithChildrenAPI], task: TaskAPI) -> None:
+    def _add_child_task(
+        self, parent: Optional[TaskWithChildrenAPI], task: TaskAPI
+    ) -> None:
         if parent is not None:
             parent.add_child(task)
 
@@ -355,7 +358,7 @@ async def run(self) -> None:
                     if not isinstance(exc_value, DaemonTaskExit):
                         exceptions.append(exc_value.with_traceback(exc_tb))
                         error_messages.append(f"{exc_type.__name__}: {str(exc_value)}")
-            
+
             if len(exceptions) == 1:
                 raise exceptions[0]
             elif len(exceptions) > 1:
@@ -440,8 +443,9 @@ def run_child_service(
 
 @asynccontextmanager
 async def background_anyio_service(service: ServiceAPI) -> AsyncIterator[ManagerAPI]:
-    """Run a service in the background and yield its manager.
-    
+    """
+    Run a service in the background and yield its manager.
+
     The service will be stopped when the context exits.
     """
     async with anyio.create_task_group() as tg:

libp2p/tools/anyio_service/base.py

@@ -1,14 +1,9 @@
-from abc import (
-    abstractmethod,
-)
 from collections.abc import (
     Awaitable,
 )
 from typing import (
     Any,
     Callable,
-    TypeVar,
-    cast,
 )
 
 from .abc import (
@@ -63,4 +58,4 @@ async def run(self) -> None:
 
     _Service.__name__ = service_fn.__name__
     _Service.__doc__ = service_fn.__doc__
-    return _Service 
+    return _Service

libp2p/tools/anyio_service/exceptions.py

@@ -13,4 +13,4 @@ class LifecycleError(ServiceException):
 class DaemonTaskExit(ServiceException):
     """
     Raised when a daemon task exits unexpectedly.
-    """ 
+    """

libp2p/tools/anyio_service/stats.py

@@ -1,4 +1,6 @@
-from dataclasses import dataclass
+from dataclasses import (
+    dataclass,
+)
 
 
 @dataclass
@@ -9,4 +11,4 @@ class TaskStats:
 
 @dataclass
 class Stats:
-    tasks: TaskStats 
+    tasks: TaskStats

libp2p/tools/anyio_service/typing.py

@@ -11,4 +11,4 @@
 
 AsyncFn = Callable[..., Awaitable[Any]]
 
-TFunc = TypeVar("TFunc", bound=Callable[..., Awaitable[Any]]) 
+TFunc = TypeVar("TFunc", bound=Callable[..., Awaitable[Any]])

newsfragments/604.feature.rst

@@ -1 +1 @@
-Transitioned from `async_service` to `anyio_service`, leveraging AnyIO's robust async primitives for improved task management and exception handling. This change enhances compatibility with modern async patterns and improves overall service reliability. 
+Transitioned from `async_service` to `anyio_service`, leveraging AnyIO's robust async primitives for improved task management and exception handling. This change enhances compatibility with modern async patterns and improves overall service reliability.

tests/core/tools/anyio_service/test_anyio_service.py

@@ -1,6 +1,7 @@
+import sys
+
 import pytest
 import anyio
-import sys
 
 if sys.version_info >= (3, 11):
     from builtins import (
@@ -10,12 +11,13 @@
     from exceptiongroup import ExceptionGroup
 
 from libp2p.tools.anyio_service import (
-    Service,
     AnyioManager,
+    Service,
     as_service,
     background_anyio_service,
 )
 
+
 @pytest.mark.anyio
 async def test_service_lifecycle():
     class SimpleService(Service):
@@ -28,6 +30,7 @@ async def run(self):
         assert manager.is_running
     assert manager.is_finished
 
+
 @pytest.mark.anyio
 async def test_exception_handling():
     class ErrorService(Service):
@@ -44,6 +47,7 @@ async def run(self):
         for e in exc_info.value.exceptions
     )
 
+
 @pytest.mark.anyio
 async def test_task_management():
     task_event = anyio.Event()
@@ -60,6 +64,7 @@ async def task_fn():
         with anyio.fail_after(0.1):
             await task_event.wait()
 
+
 @pytest.mark.anyio
 async def test_cancellation_and_cleanup():
     class CancellableService(Service):
@@ -71,4 +76,4 @@ async def run(self):
         assert manager.is_running
         manager.cancel()
     assert manager.is_cancelled
-    assert manager.is_finished 
+    assert manager.is_finished