Add a new TaskCreator protocol

This will be used in the future to allow overriding where to create
tasks (for example the default `asyncio` loop, a custom `loop` or even
a `TaskGroup`).

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/core/asyncio.py

@@ -12,14 +12,51 @@
   task and waits for it to finish, handling `CancelledError` exceptions.
 - [Service][frequenz.core.asyncio.Service]: A base class for
   implementing services running in the background that can be started and stopped.
+- [TaskCreator][frequenz.core.asyncio.TaskCreator]: A protocol for creating tasks.
 """
 
 
 import abc
 import asyncio
 import collections.abc
+import contextvars
 from types import TracebackType
-from typing import Any, Self
+from typing import Any, Protocol, Self, TypeVar, runtime_checkable
+
+TaskReturnT = TypeVar("TaskReturnT")
+"""The type of the return value of a task."""
+
+
+@runtime_checkable
+class TaskCreator(Protocol):
+    """A protocol for creating tasks.
+
+    Built-in asyncio functions and classes implementing this protocol:
+
+    - [`asyncio`][]
+    - [`asyncio.AbstractEventLoop`][] (returned by [`asyncio.get_event_loop`][] for
+      example)
+    - [`asyncio.TaskGroup`][]
+    """
+
+    def create_task(
+        self,
+        coro: collections.abc.Coroutine[Any, Any, TaskReturnT],
+        *,
+        name: str | None = None,
+        context: contextvars.Context | None = None,
+    ) -> asyncio.Task[TaskReturnT]:
+        """Create a task.
+
+        Args:
+            coro: The coroutine to be executed.
+            name: The name of the task.
+            context: The context to be used for the task.
+
+        Returns:
+            The new task.
+        """
+        ...  # pylint: disable=unnecessary-ellipsis
 
 
 async def cancel_and_await(task: asyncio.Task[Any]) -> None:

tests/test_asyncio.py

@@ -9,7 +9,7 @@
 import async_solipsism
 import pytest
 
-from frequenz.core.asyncio import Service
+from frequenz.core.asyncio import Service, TaskCreator
 
 
 # This method replaces the event loop for all tests in the file.
@@ -152,3 +152,18 @@ async def test_async_context_manager() -> None:
         assert fake_service.is_running is True
 
     assert fake_service.is_running is False
+
+
+def test_task_creator_asyncio() -> None:
+    """Test that the asyncio module is a TaskCreator."""
+    assert isinstance(asyncio, TaskCreator)
+
+
+async def test_task_creator_loop() -> None:
+    """Test that the asyncio event loop is a TaskCreator."""
+    assert isinstance(asyncio.get_event_loop(), TaskCreator)
+
+
+def test_task_creator_task_group() -> None:
+    """Test that the asyncio task group is a TaskCreator."""
+    assert isinstance(asyncio.TaskGroup(), TaskCreator)