Implement custom asyncio event loop based on Qt

asyncio is an established library for Python applications with
concurrency and asynchronous I/O, and the de facto standard that
multiple other async frameworks build upon. Like Qt, it is based on an
event loop, so to this end, it offers an extensive API to implement
custom event loops that applications using asyncio can then leverage.

Task-number: PYSIDE-769
Change-Id: I3daf5d631e2fa0d44fd8c3c272ac5cce96f58653
Reviewed-by: Cristian Maureira-Fredes <cristian.maureira-fredes@qt.io>

13 files changed, 907 insertions, 0 deletions

diff --git a/build_history/blacklist.txt b/build_history/blacklist.txt
index 475e5d40e..cb9b1c500 100644
--- a/build_history/blacklist.txt
+++ b/build_history/blacklist.txt
@@ -41,6 +41,8 @@
     darwin ci
 [QtWidgets::qpicture_test]
     darwin ci
+[QtAsyncio::qasyncio_test_chain]
+    win32
 
 # PYSIDE-535: These errors are still present. Please try to remove one :)
 [sample::mixed_mi]
diff --git a/build_scripts/platforms/unix.py b/build_scripts/platforms/unix.py
index 570c4d570..a8510f7ed 100644
--- a/build_scripts/platforms/unix.py
+++ b/build_scripts/platforms/unix.py
@@ -209,6 +209,13 @@ def prepare_packages_posix(pyside_build, _vars, cross_build=False):
                 "{st_build_dir}/{st_package_name}/support",
                 _vars=_vars)
 
+        # <source>/pyside6/{st_package_name}/QtAsyncio/* ->
+        #   <setup>/{st_package_name}/QtAsyncio/*
+        copydir(
+            f"{{build_dir}}/{PYSIDE}/{{st_package_name}}/QtAsyncio",
+            "{st_build_dir}/{st_package_name}/QtAsyncio",
+            _vars=_vars)
+
         # <source>/pyside6/{st_package_name}/*.pyi ->
         #   <setup>/{st_package_name}/*.pyi
         copydir(
diff --git a/build_scripts/platforms/windows_desktop.py b/build_scripts/platforms/windows_desktop.py
index ee293aca2..f8ad4c8c8 100644
--- a/build_scripts/platforms/windows_desktop.py
+++ b/build_scripts/platforms/windows_desktop.py
@@ -163,6 +163,13 @@ def prepare_packages_win32(pyside_build, _vars):
             destination_dir / "support",
             _vars=_vars)
 
+        # <source>/pyside6/{st_package_name}/QtAsyncio/* ->
+        #   <setup>/{st_package_name}/QtAsyncio/*
+        copydir(
+            f"{{build_dir}}/{PYSIDE}/{{st_package_name}}/QtAsyncio",
+            "{st_build_dir}/{st_package_name}/QtAsyncio",
+            _vars=_vars)
+
         # <source>/pyside6/{st_package_name}/*.pyi ->
         #   <setup>/{st_package_name}/*.pyi
         copydir(
diff --git a/build_scripts/wheel_files.py b/build_scripts/wheel_files.py
index 71a2cbf68..8636f471f 100644
--- a/build_scripts/wheel_files.py
+++ b/build_scripts/wheel_files.py
@@ -221,6 +221,7 @@ def wheel_files_pyside_addons() -> List[ModuleData]:
         module_QtWebSockets(),
         module_QtHttpServer(),
         module_QtLocation(),
+        module_QtAsyncio(),
     ]
     return files
 
@@ -983,3 +984,8 @@ def module_QtLocation() -> ModuleData:
     data.plugins = get_module_plugins(json_data)
     data.translations.append("qtlocation_*")
     return data
+
+
+def module_QtAsyncio() -> ModuleData:
+    data = ModuleData("Asyncio")
+    return data
diff --git a/sources/pyside6/PySide6/CMakeLists.txt b/sources/pyside6/PySide6/CMakeLists.txt
index cf913e797..e24fdb682 100644
--- a/sources/pyside6/PySide6/CMakeLists.txt
+++ b/sources/pyside6/PySide6/CMakeLists.txt
@@ -109,3 +109,6 @@ install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/templates/opengl_common.xml
         DESTINATION share/PySide6${pyside_SUFFIX}/typesystems)
 install(FILES ${CMAKE_CURRENT_BINARY_DIR}/pyside6_global.h
         DESTINATION include/${BINDING_NAME}${pyside6_SUFFIX})
+
+install(DIRECTORY "${CMAKE_CURRENT_SOURCE_DIR}/QtAsyncio"
+        DESTINATION "${CMAKE_CURRENT_BINARY_DIR}")
diff --git a/sources/pyside6/PySide6/QtAsyncio/__init__.py b/sources/pyside6/PySide6/QtAsyncio/__init__.py
new file mode 100644
index 000000000..2c673e405
--- /dev/null
+++ b/sources/pyside6/PySide6/QtAsyncio/__init__.py
@@ -0,0 +1,14 @@
+# Copyright (C) 2023 The Qt Company Ltd.
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
+
+from .events import (
+    QAsyncioEventLoopPolicy, QAsyncioEventLoop, QAsyncioHandle, QAsyncioTimerHandle
+)
+from .futures import QAsyncioFuture
+from .tasks import QAsyncioTask
+
+__all__ = [
+    "QAsyncioEventLoopPolicy", "QAsyncioEventLoop",
+    "QAsyncioHandle", "QAsyncioTimerHandle",
+    "QAsyncioFuture", "QAsyncioTask"
+]
diff --git a/sources/pyside6/PySide6/QtAsyncio/events.py b/sources/pyside6/PySide6/QtAsyncio/events.py
new file mode 100644
index 000000000..ebae09a73
--- /dev/null
+++ b/sources/pyside6/PySide6/QtAsyncio/events.py
@@ -0,0 +1,512 @@
+# Copyright (C) 2023 The Qt Company Ltd.
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
+
+from PySide6.QtCore import QDateTime, QCoreApplication, QTimer, QThread, Slot
+
+from . import futures
+from . import tasks
+
+import asyncio
+import collections.abc
+import concurrent.futures
+import contextvars
+import enum
+import os
+import socket
+import subprocess
+import typing
+
+__all__ = [
+    "QAsyncioEventLoopPolicy", "QAsyncioEventLoop",
+    "QAsyncioHandle", "QAsyncioTimerHandle",
+]
+
+
+class QAsyncioEventLoopPolicy(asyncio.AbstractEventLoopPolicy):
+    def __init__(self, application: typing.Optional[QCoreApplication] = None) -> None:
+        super().__init__()
+        if application is None:
+            if QCoreApplication.instance() is None:
+                application = QCoreApplication()
+            else:
+                application = QCoreApplication.instance()
+        self._application: QCoreApplication = application  # type: ignore[assignment]
+        self._event_loop: typing.Optional[asyncio.AbstractEventLoop] = None
+
+    def get_event_loop(self) -> asyncio.AbstractEventLoop:
+        if self._event_loop is None:
+            self._event_loop = QAsyncioEventLoop(self._application)
+        return self._event_loop
+
+    def set_event_loop(self, loop: typing.Optional[asyncio.AbstractEventLoop]) -> None:
+        self._event_loop = loop
+
+    def new_event_loop(self) -> asyncio.AbstractEventLoop:
+        return QAsyncioEventLoop(self._application)
+
+
+class QAsyncioEventLoop(asyncio.BaseEventLoop):
+    """
+    Implements the asyncio API:
+    https://docs.python.org/3/library/asyncio-eventloop.html
+    """
+
+    class ShutDownThread(QThread):
+        def __init__(self, future: futures.QAsyncioFuture, loop: "QAsyncioEventLoop") -> None:
+            super().__init__()
+            self._future = future
+            self._loop = loop
+            self.started.connect(self.shutdown)
+
+        def run(self) -> None:
+            pass
+
+        def shutdown(self) -> None:
+            try:
+                self._loop._default_executor.shutdown(wait=True)
+                if not self._loop.is_closed():
+                    self._loop.call_soon_threadsafe(self._future.set_result, None)
+            except Exception as e:
+                if not self._loop.is_closed():
+                    self._loop.call_soon_threadsafe(self._future.set_exception, e)
+
+    def __init__(self, application: QCoreApplication) -> None:
+        super().__init__()
+
+        self._application: QCoreApplication = application
+        self._thread = QThread.currentThread()
+
+        self._closed = False
+        self._asyncgens: typing.Set[collections.abc.AsyncGenerator] = set()
+
+        # Starting with Python 3.11, this must be an instance of
+        # ThreadPoolExecutor.
+        self._default_executor = concurrent.futures.ThreadPoolExecutor()
+
+        self._exception_handler: typing.Optional[typing.Callable] = self.default_exception_handler
+        self._task_factory: typing.Optional[typing.Callable] = None
+        self._future_to_complete: typing.Optional[futures.QAsyncioFuture] = None
+
+        self._debug = bool(os.getenv("PYTHONASYNCIODEBUG", False))
+
+    # Running and stopping the loop
+
+    def _run_until_complete_cb(self, future: futures.QAsyncioFuture) -> None:
+        if not future.cancelled():
+            if isinstance(future.exception(), (SystemExit, KeyboardInterrupt)):
+                return
+        future.get_loop().stop()
+
+    def run_until_complete(self, future: futures.QAsyncioFuture) -> typing.Any:  # type: ignore[override]
+        if self.is_closed():
+            raise RuntimeError("Event loop is closed")
+        if self.is_running():
+            raise RuntimeError("Event loop is already running")
+
+        arg_was_coro = not asyncio.futures.isfuture(future)
+        future = asyncio.tasks.ensure_future(future, loop=self)  # type: ignore[assignment]
+        future.add_done_callback(self._run_until_complete_cb)
+        self._future_to_complete = future
+
+        try:
+            self.run_forever()
+        except Exception as e:
+            if arg_was_coro and future.done() and not future.cancelled():
+                future.exception()
+            raise e
+        finally:
+            future.remove_done_callback(self._run_until_complete_cb)
+        if not future.done():
+            raise RuntimeError("Event loop stopped before Future completed")
+
+        return future.result()
+
+    def run_forever(self) -> None:
+        if self.is_closed():
+            raise RuntimeError("Event loop is closed")
+        if self.is_running():
+            raise RuntimeError("Event loop is already running")
+        asyncio.events._set_running_loop(self)
+        self._application.exec()
+        asyncio.events._set_running_loop(None)
+
+    def stop(self) -> None:
+        if self._future_to_complete is not None:
+            if self._future_to_complete.done():
+                self._future_to_complete = None
+            else:
+                return
+        self._application.quit()
+
+    def is_running(self) -> bool:
+        return self._thread.loopLevel() > 0
+
+    def is_closed(self) -> bool:
+        return self._closed
+
+    def close(self) -> None:
+        if self.is_running():
+            raise RuntimeError("Cannot close a running event loop")
+        if self.is_closed():
+            return
+        if self._default_executor is not None:
+            self._default_executor.shutdown(wait=False)
+        self._application.shutdown()
+        self._closed = True
+
+    async def shutdown_asyncgens(self) -> None:
+        if not len(self._asyncgens):
+            return
+
+        results = await asyncio.tasks.gather(
+            *[asyncgen.aclose() for asyncgen in self._asyncgens],
+            return_exceptions=True)
+
+        for result, asyncgen in zip(results, self._asyncgens):
+            if isinstance(result, Exception):
+                self.call_exception_handler({
+                    "message": f"Closing asynchronous generator {asyncgen}"
+                               f"raised an exception",
+                    "exception": result,
+                    "asyncgen": asyncgen})
+
+        self._asyncgens.clear()
+
+    async def shutdown_default_executor(self) -> None:
+        if self._default_executor is None:
+            return
+        future = self.create_future()
+        thread = QAsyncioEventLoop.ShutDownThread(future, self)
+        thread.start()
+        try:
+            await future
+        finally:
+            thread.wait()
+
+    # Scheduling callbacks
+
+    def call_soon(self, callback: typing.Callable, *args: typing.Any,
+                  context: typing.Optional[contextvars.Context] = None):
+        return self.call_later(0, callback, *args, context=context)
+
+    def call_soon_threadsafe(self, callback: typing.Callable,  # type: ignore[override]
+                             *args: typing.Any,
+                             context:
+                             typing.Optional[contextvars.Context] = None) -> "QAsyncioHandle":
+        if self.is_closed():
+            raise RuntimeError("Event loop is closed")
+        return self.call_soon(callback, *args, context=context)
+
+    def call_later(self, delay: typing.Union[int, float],  # type: ignore[override]
+                   callback: typing.Callable, *args: typing.Any,
+                   context: typing.Optional[contextvars.Context] = None) -> "QAsyncioHandle":
+        if not isinstance(delay, (int, float)):
+            raise TypeError("delay must be an int or float")
+        return self.call_at(self.time() + delay * 1000, callback, *args,
+                            context=context)
+
+    def call_at(self, when: typing.Union[int, float],  # type: ignore[override]
+                callback: typing.Callable, *args: typing.Any,
+                context: typing.Optional[contextvars.Context] = None) -> "QAsyncioHandle":
+        if not isinstance(when, (int, float)):
+            raise TypeError("when must be an int or float")
+        if self.is_closed():
+            raise RuntimeError("Event loop is closed")
+        return QAsyncioTimerHandle(int(when), callback, args, self, context)
+
+    def time(self) -> int:
+        return QDateTime.currentMSecsSinceEpoch()
+
+    # Creating Futures and Tasks
+
+    def create_future(self) -> futures.QAsyncioFuture:  # type: ignore[override]
+        return futures.QAsyncioFuture(loop=self)
+
+    def create_task(self,  # type: ignore[override]
+                    coro: typing.Union[collections.abc.Generator, collections.abc.Coroutine],
+                    *, name: typing.Optional[str] = None,
+                    context: typing.Optional[contextvars.Context] = None) -> tasks.QAsyncioTask:
+        if self.is_closed():
+            raise RuntimeError("Event loop is closed")
+        if self._task_factory is None:
+            task = tasks.QAsyncioTask(coro, loop=self, name=name, context=context)
+        else:
+            task = self._task_factory(self, coro, context=context)
+            task.set_name(name)
+
+        return task
+
+    def set_task_factory(self, factory: typing.Optional[typing.Callable]) -> None:
+        if factory is not None and not callable(factory):
+            raise TypeError("The task factory must be a callable or None")
+        self._task_factory = factory
+
+    def get_task_factory(self) -> typing.Optional[typing.Callable]:
+        return self._task_factory
+
+    # Opening network connections
+
+    async def create_connection(
+            self, protocol_factory, host=None, port=None,
+            *, ssl=None, family=0, proto=0,
+            flags=0, sock=None, local_addr=None,
+            server_hostname=None,
+            ssl_handshake_timeout=None,
+            ssl_shutdown_timeout=None,
+            happy_eyeballs_delay=None, interleave=None):
+        raise NotImplementedError
+
+    async def create_datagram_endpoint(self, protocol_factory,
+                                       local_addr=None, remote_addr=None, *,
+                                       family=0, proto=0, flags=0,
+                                       reuse_address=None, reuse_port=None,
+                                       allow_broadcast=None, sock=None):
+        raise NotImplementedError
+
+    async def create_unix_connection(
+            self, protocol_factory, path=None, *,
+            ssl=None, sock=None,
+            server_hostname=None,
+            ssl_handshake_timeout=None,
+            ssl_shutdown_timeout=None):
+        raise NotImplementedError
+
+    # Creating network servers
+
+    async def create_server(
+            self, protocol_factory, host=None, port=None,
+            *, family=socket.AF_UNSPEC,
+            flags=socket.AI_PASSIVE, sock=None, backlog=100,
+            ssl=None, reuse_address=None, reuse_port=None,
+            ssl_handshake_timeout=None,
+            ssl_shutdown_timeout=None,
+            start_serving=True):
+        raise NotImplementedError
+
+    async def create_unix_server(
+            self, protocol_factory, path=None, *,
+            sock=None, backlog=100, ssl=None,
+            ssl_handshake_timeout=None,
+            ssl_shutdown_timeout=None,
+            start_serving=True):
+        raise NotImplementedError
+
+    async def connect_accepted_socket(
+            self, protocol_factory, sock,
+            *, ssl=None,
+            ssl_handshake_timeout=None,
+            ssl_shutdown_timeout=None):
+        raise NotImplementedError
+
+    # Transferring files
+
+    async def sendfile(self, transport, file, offset=0, count=None,
+                       *, fallback=True):
+        raise NotImplementedError
+
+    # TLS Upgrade
+
+    async def start_tls(self, transport, protocol, sslcontext, *,
+                        server_side=False,
+                        server_hostname=None,
+                        ssl_handshake_timeout=None,
+                        ssl_shutdown_timeout=None):
+        raise NotImplementedError
+
+    # Watching file descriptors
+
+    def add_reader(self, fd, callback, *args):
+        raise NotImplementedError
+
+    def remove_reader(self, fd):
+        raise NotImplementedError
+
+    def add_writer(self, fd, callback, *args):
+        raise NotImplementedError
+
+    def remove_writer(self, fd):
+        raise NotImplementedError
+
+    # Working with socket objects directly
+
+    async def sock_recv(self, sock, nbytes):
+        raise NotImplementedError
+
+    async def sock_recv_into(self, sock, buf):
+        raise NotImplementedError
+
+    async def sock_recvfrom(self, sock, bufsize):
+        raise NotImplementedError
+
+    async def sock_recvfrom_into(self, sock, buf, nbytes=0):
+        raise NotImplementedError
+
+    async def sock_sendall(self, sock, data):
+        raise NotImplementedError
+
+    async def sock_sendto(self, sock, data, address):
+        raise NotImplementedError
+
+    async def sock_connect(self, sock, address):
+        raise NotImplementedError
+
+    async def sock_accept(self, sock):
+        raise NotImplementedError
+
+    async def sock_sendfile(self, sock, file, offset=0, count=None, *,
+                            fallback=None):
+        raise NotImplementedError
+
+    # DNS
+
+    async def getaddrinfo(self, host, port, *,
+                          family=0, type=0, proto=0, flags=0):
+        raise NotImplementedError
+
+    async def getnameinfo(self, sockaddr, flags=0):
+        raise NotImplementedError
+
+    # Working with pipes
+
+    async def connect_read_pipe(self, protocol_factory, pipe):
+        raise NotImplementedError
+
+    async def connect_write_pipe(self, protocol_factory, pipe):
+        raise NotImplementedError
+
+    # Unix signals
+
+    def add_signal_handler(self, sig, callback, *args):
+        raise NotImplementedError
+
+    def remove_signal_handler(self, sig):
+        raise NotImplementedError
+
+    # Executing code in thread or process pools
+
+    def run_in_executor(self,
+                        executor: typing.Optional[concurrent.futures.ThreadPoolExecutor],
+                        func: typing.Callable, *args: typing.Tuple) -> asyncio.futures.Future:
+        if self.is_closed():
+            raise RuntimeError("Event loop is closed")
+        if executor is None:
+            executor = self._default_executor
+        return asyncio.futures.wrap_future(
+            executor.submit(func, *args), loop=self
+        )
+
+    def set_default_executor(self,
+                             executor: typing.Optional[
+                                 concurrent.futures.ThreadPoolExecutor]) -> None:
+        if not isinstance(executor, concurrent.futures.ThreadPoolExecutor):
+            raise TypeError("The executor must be a ThreadPoolExecutor")
+        self._default_executor = executor
+
+    # Error Handling API
+
+    def set_exception_handler(self, handler: typing.Optional[typing.Callable]) -> None:
+        if handler is not None and not callable(handler):
+            raise TypeError("The handler must be a callable or None")
+        self._exception_handler = handler
+
+    def get_exception_handler(self) -> typing.Optional[typing.Callable]:
+        return self._exception_handler
+
+    def default_exception_handler(self, context: typing.Dict[str, typing.Any]) -> None:
+        # TODO
+        print(context["message"])
+
+    def call_exception_handler(self, context: typing.Dict[str, typing.Any]) -> None:
+        if self._exception_handler is not None:
+            self._exception_handler(context)
+
+    # Enabling debug mode
+
+    def get_debug(self) -> bool:
+        # TODO: Part of the asyncio API but currently unused. More details:
+        # https://docs.python.org/3/library/asyncio-dev.html#asyncio-debug-mode
+        return self._debug
+
+    def set_debug(self, enabled: bool) -> None:
+        self._debug = enabled
+
+    # Running subprocesses
+
+    async def subprocess_exec(self, protocol_factory, *args,
+                              stdin=subprocess.PIPE,
+                              stdout=subprocess.PIPE,
+                              stderr=subprocess.PIPE,
+                              **kwargs):
+        raise NotImplementedError
+
+    async def subprocess_shell(self, protocol_factory, cmd, *,
+                               stdin=subprocess.PIPE,
+                               stdout=subprocess.PIPE,
+                               stderr=subprocess.PIPE,
+                               **kwargs):
+        raise NotImplementedError
+
+
+class QAsyncioHandle():
+    class HandleState(enum.Enum):
+        PENDING = enum.auto()
+        CANCELLED = enum.auto()
+        DONE = enum.auto()
+
+    def __init__(self, callback: typing.Callable, args: typing.Tuple,
+                 loop: QAsyncioEventLoop, context: typing.Optional[contextvars.Context]) -> None:
+        self._cancel_exception_msg: typing.Optional[str] = None
+
+        self._callback = callback
+        self._args = args
+        self._loop = loop
+        self._context = context
+
+        self._timeout = 0
+
+        self._state = QAsyncioHandle.HandleState.PENDING
+        self._start()
+
+    def _start(self) -> None:
+        QTimer.singleShot(self._timeout, lambda: self._cb())
+
+    @Slot()
+    def _cb(self) -> None:
+        if self._state == QAsyncioHandle.HandleState.PENDING:
+            if self._context is not None:
+                self._context.run(self._callback, *self._args)
+            else:
+                self._callback(*self._args)
+            self._state = QAsyncioHandle.HandleState.DONE
+
+    @Slot()
+    def _cancel_exception(self) -> None:
+        raise asyncio.CancelledError(self._cancel_exception_msg)
+
+    def cancel(self) -> None:
+        if self._state == QAsyncioHandle.HandleState.PENDING:
+            self._state = QAsyncioHandle.HandleState.CANCELLED
+            # The old timer that was created in _start will still trigger but _cb won't do anything.
+            QTimer.singleShot(0, lambda: self._cancel_exception())
+
+    def cancelled(self) -> bool:
+        return self._state == QAsyncioHandle.HandleState.CANCELLED
+
+
+class QAsyncioTimerHandle(QAsyncioHandle):
+    def __init__(self, when: int, callback: typing.Callable, args: typing.Tuple,
+                 loop: QAsyncioEventLoop, context: typing.Optional[contextvars.Context]) -> None:
+        super().__init__(callback, args, loop, context)
+
+        self._when = when
+        self._timeout = max(self._when - self._loop.time(), 0)
+
+        super()._start()
+
+    # Override this so that timer.start() is only called once at the end
+    # of the constructor for both QtHandle and QtTimerHandle.
+    def _start(self) -> None:
+        pass
+
+    def when(self) -> int:
+        return self._when
diff --git a/sources/pyside6/PySide6/QtAsyncio/futures.py b/sources/pyside6/PySide6/QtAsyncio/futures.py
new file mode 100644
index 000000000..abd01db8f
--- /dev/null
+++ b/sources/pyside6/PySide6/QtAsyncio/futures.py
@@ -0,0 +1,107 @@
+# Copyright (C) 2023 The Qt Company Ltd.
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
+
+from . import events
+
+import asyncio
+import contextvars
+import enum
+import typing
+
+
+class QAsyncioFuture():
+    """ https://docs.python.org/3/library/asyncio-future.html """
+
+    # Declare that this class implements the Future protocol. The field must
+    # exist and be boolean - True indicates 'await' or 'yield from', False
+    # indicates 'yield'.
+    _asyncio_future_blocking = False
+
+    class FutureState(enum.Enum):
+        PENDING = enum.auto()
+        CANCELLED = enum.auto()
+        DONE_WITH_RESULT = enum.auto()
+        DONE_WITH_EXCEPTION = enum.auto()
+
+    def __init__(self, *, loop: typing.Optional["events.QAsyncioEventLoop"] = None,
+                 context: typing.Optional[contextvars.Context] = None) -> None:
+        if loop is None:
+            self._loop = asyncio.events.get_event_loop()
+        else:
+            self._loop = loop
+        self._context = context
+
+        self._state = QAsyncioFuture.FutureState.PENDING
+        self._result: typing.Any = None
+        self._exception: typing.Optional[Exception] = None
+
+        self._callbacks: typing.List[typing.Callable] = list()
+
+    def __await__(self):
+        if not self.done():
+            self._asyncio_future_blocking = True
+            yield self
+            if not self.done():
+                raise RuntimeError("await was not used with a Future")
+        return self.result()
+
+    __iter__ = __await__
+
+    def _schedule_callbacks(self, context: typing.Optional[contextvars.Context] = None):
+        for cb in self._callbacks:
+            self._loop.call_soon(
+                cb, self, context=context if context else self._context)
+
+    def result(self) -> typing.Union[typing.Any, Exception]:
+        if self._state == QAsyncioFuture.FutureState.DONE_WITH_RESULT:
+            return self._result
+        if self._state == QAsyncioFuture.FutureState.DONE_WITH_EXCEPTION and self._exception:
+            raise self._exception
+        if self._state == QAsyncioFuture.FutureState.CANCELLED:
+            raise asyncio.CancelledError
+        raise asyncio.InvalidStateError
+
+    def set_result(self, result: typing.Any) -> None:
+        self._result = result
+        self._state = QAsyncioFuture.FutureState.DONE_WITH_RESULT
+        self._schedule_callbacks()
+
+    def set_exception(self, exception: Exception) -> None:
+        self._exception = exception
+        self._state = QAsyncioFuture.FutureState.DONE_WITH_EXCEPTION
+        self._schedule_callbacks()
+
+    def done(self) -> bool:
+        return self._state != QAsyncioFuture.FutureState.PENDING
+
+    def cancelled(self) -> bool:
+        return self._state == QAsyncioFuture.FutureState.CANCELLED
+
+    def add_done_callback(self, cb: typing.Callable, *,
+                          context: typing.Optional[contextvars.Context] = None) -> None:
+        if self.done():
+            self._loop.call_soon(
+                cb, self, context=context if context else self._context)
+        else:
+            self._callbacks.append(cb)
+
+    def remove_done_callback(self, cb: typing.Callable) -> int:
+        original_len = len(self._callbacks)
+        self._callbacks = [_cb for _cb in self._callbacks if _cb != cb]
+        return original_len - len(self._callbacks)
+
+    def cancel(self) -> None:
+        if self.done():
+            return
+        self._state = QAsyncioFuture.FutureState.CANCELLED
+        self._schedule_callbacks()
+
+    def exception(self) -> typing.Optional[Exception]:
+        if self._state == QAsyncioFuture.FutureState.CANCELLED:
+            raise asyncio.CancelledError
+        if self.done():
+            return self._exception
+        raise asyncio.InvalidStateError
+
+    def get_loop(self) -> asyncio.AbstractEventLoop:
+        return self._loop
diff --git a/sources/pyside6/PySide6/QtAsyncio/tasks.py b/sources/pyside6/PySide6/QtAsyncio/tasks.py
new file mode 100644
index 000000000..ca3d08d24
--- /dev/null
+++ b/sources/pyside6/PySide6/QtAsyncio/tasks.py
@@ -0,0 +1,136 @@
+# Copyright (C) 2023 The Qt Company Ltd.
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
+
+from . import events
+from . import futures
+
+import asyncio
+import collections.abc
+import concurrent.futures
+import contextvars
+import typing
+
+
+class QAsyncioTask(futures.QAsyncioFuture):
+    """ https://docs.python.org/3/library/asyncio-task.html """
+
+    def __init__(self, coro: typing.Union[collections.abc.Generator, collections.abc.Coroutine], *,
+                 loop: typing.Optional["events.QAsyncioEventLoop"] = None,
+                 name: typing.Optional[str] = None,
+                 context: typing.Optional[contextvars.Context] = None) -> None:
+        super().__init__(loop=loop, context=context)
+
+        self._coro = coro
+        self._name = name if name else "QtTask"
+
+        self._handle = self._loop.call_soon(self._step, context=self._context)
+
+        self._cancellation_requests = 0
+
+    def __repr__(self) -> str:
+        if self._state == futures.QAsyncioFuture.FutureState.PENDING:
+            state = "Pending"
+        elif self._state == futures.QAsyncioFuture.FutureState.DONE_WITH_RESULT:
+            state = "Done"
+        elif self._state == futures.QAsyncioFuture.FutureState.DONE_WITH_EXCEPTION:
+            state = f"Done with exception ({repr(self._exception)})"
+        elif self._state == futures.QAsyncioFuture.FutureState.CANCELLED:
+            state = "Cancelled"
+
+        return f"Task '{self.get_name()}' with state: {state}"
+
+    class QtTaskApiMisuseError(Exception):
+        pass
+
+    def __await__(self) -> None:  # type: ignore[override]
+        # This function is not inherited from the Future APIs.
+        raise QAsyncioTask.QtTaskApiMisuseError("Tasks cannot be awaited")
+
+    def __iter__(self) -> None:  # type: ignore[override]
+        # This function is not inherited from the Future APIs.
+        raise QAsyncioTask.QtTaskApiMisuseError("Tasks cannot be iterated over")
+
+    def set_result(self, result: typing.Any) -> None:  # type: ignore[override]
+        # This function is not inherited from the Future APIs.
+        raise QAsyncioTask.QtTaskApiMisuseError("Tasks cannot set results")
+
+    def set_exception(self, exception: typing.Any) -> None:  # type: ignore[override]
+        # This function is not inherited from the Future APIs.
+        raise QAsyncioTask.QtTaskApiMisuseError("Tasks cannot set exceptions")
+
+    def _step(self,
+              exception_or_future: typing.Union[
+                  Exception, futures.QAsyncioFuture, None] = None):
+        if self.done():
+            raise asyncio.exceptions.InvalidStateError("Task is already done")
+        result = None
+        try:
+            if exception_or_future is None:
+                result = self._coro.send(None)
+            elif asyncio.futures.isfuture(exception_or_future):
+                exception_or_future.result()
+                exception_or_future = None
+                result = self._coro.send(None)
+            elif isinstance(exception_or_future, Exception):
+                result = self._coro.throw(exception_or_future)
+        except StopIteration as e:
+            self._state = futures.QAsyncioFuture.FutureState.DONE_WITH_RESULT
+            self._result = e.value
+        except concurrent.futures.CancelledError as e:
+            self._state = futures.QAsyncioFuture.FutureState.CANCELLED
+            self._exception = e
+        except Exception as e:
+            self._state = futures.QAsyncioFuture.FutureState.DONE_WITH_EXCEPTION
+            self._exception = e  # type: ignore[assignment]
+        else:
+            if asyncio.futures.isfuture(result):
+                result.add_done_callback(
+                    self._step, context=self._context)  # type: ignore[arg-type]
+            else:
+                self._loop.call_soon(self._step, exception_or_future, context=self._context)
+        finally:
+            if self._exception:
+                self._loop.call_exception_handler({
+                    "message": (str(self._exception) if self._exception
+                                else "An exception occurred during task "
+                                "execution"),
+                    "exception": self._exception,
+                    "task": self,
+                    "future": (exception_or_future
+                               if asyncio.futures.isfuture(exception_or_future)
+                               else None)
+                })
+            if self.done():
+                self._schedule_callbacks()
+                self._loop.stop()
+
+    def get_stack(self, *, limit=None) -> typing.List[typing.Any]:
+        # TODO
+        raise NotImplementedError("QtTask.get_stack is not implemented")
+
+    def print_stack(self, *, limit=None, file=None) -> None:
+        # TODO
+        raise NotImplementedError("QtTask.print_stack is not implemented")
+
+    def get_coro(self) -> typing.Union[collections.abc.Generator, collections.abc.Coroutine]:
+        return self._coro
+
+    def get_name(self) -> str:
+        return self._name
+
+    def set_name(self, value) -> None:
+        self._name = str(value)
+
+    def cancel(self, msg: typing.Optional[str] = None) -> None:
+        if (isinstance(self._handle, events.QAsyncioHandle)):
+            self._handle._cancel_exception_msg = msg
+        self._handle.cancel()
+        self._state = futures.QAsyncioFuture.FutureState.CANCELLED
+
+    def uncancel(self) -> None:
+        # TODO
+        raise NotImplementedError("QtTask.uncancel is not implemented")
+
+    def cancelling(self) -> bool:
+        # TODO
+        raise NotImplementedError("QtTask.cancelling is not implemented")
diff --git a/sources/pyside6/tests/QtAsyncio/CMakeLists.txt b/sources/pyside6/tests/QtAsyncio/CMakeLists.txt
new file mode 100644
index 000000000..935e0d90a
--- /dev/null
+++ b/sources/pyside6/tests/QtAsyncio/CMakeLists.txt
@@ -0,0 +1,2 @@
+PYSIDE_TEST(qasyncio_test.py)
+PYSIDE_TEST(qasyncio_test_chain.py)
diff --git a/sources/pyside6/tests/QtAsyncio/QtAsyncio.pyproject b/sources/pyside6/tests/QtAsyncio/QtAsyncio.pyproject
new file mode 100644
index 000000000..33f514695
--- /dev/null
+++ b/sources/pyside6/tests/QtAsyncio/QtAsyncio.pyproject
@@ -0,0 +1,3 @@
+{
+    "files": ["qt_asyncio_test.py", "qt_asyncio_test_chain.py"]
+}
diff --git a/sources/pyside6/tests/QtAsyncio/qasyncio_test.py b/sources/pyside6/tests/QtAsyncio/qasyncio_test.py
new file mode 100644
index 000000000..953483818
--- /dev/null
+++ b/sources/pyside6/tests/QtAsyncio/qasyncio_test.py
@@ -0,0 +1,51 @@
+# Copyright (C) 2023 The Qt Company Ltd.
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only WITH Qt-GPL-exception-1.0
+
+'''Test cases for QtAsyncio'''
+
+import unittest
+import asyncio
+
+from PySide6.QtAsyncio import QAsyncioEventLoopPolicy
+
+
+class QAsyncioTestCase(unittest.TestCase):
+    async def sleep(self, output):
+        output += "Hello"
+        await asyncio.sleep(1)
+        output += "World"
+
+    async def gather(self, output):
+        await asyncio.gather(self.sleep(output), self.sleep(output), self.sleep(output))
+
+    def test_sleep(self):
+        outputs_expected = []
+        outputs_real = []
+
+        # Run the code without QAsyncioEventLoopPolicy
+        asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
+        asyncio.run(self.sleep(outputs_expected))
+
+        # Run the code with QAsyncioEventLoopPolicy and QtEventLoop
+        asyncio.set_event_loop_policy(QAsyncioEventLoopPolicy())
+        asyncio.run(self.sleep(outputs_real))
+
+        self.assertEqual(outputs_expected, outputs_real)
+
+    def test_gather(self):
+        outputs_expected = []
+        outputs_real = []
+
+        # Run the code without QAsyncioEventLoopPolicy
+        asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
+        asyncio.run(self.gather(outputs_expected))
+
+        # Run the code with QAsyncioEventLoopPolicy and QtEventLoop
+        asyncio.set_event_loop_policy(QAsyncioEventLoopPolicy())
+        asyncio.run(self.gather(outputs_real))
+
+        self.assertEqual(outputs_expected, outputs_real)
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/sources/pyside6/tests/QtAsyncio/qasyncio_test_chain.py b/sources/pyside6/tests/QtAsyncio/qasyncio_test_chain.py
new file mode 100644
index 000000000..f45b51a71
--- /dev/null
+++ b/sources/pyside6/tests/QtAsyncio/qasyncio_test_chain.py
@@ -0,0 +1,57 @@
+# Copyright (C) 2023 The Qt Company Ltd.
+# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only WITH Qt-GPL-exception-1.0
+
+'''Test cases for QtAsyncio'''
+
+import unittest
+import asyncio
+import random
+import time
+
+from PySide6.QtAsyncio import QAsyncioEventLoopPolicy
+
+
+class QAsyncioTestCaseChain(unittest.TestCase):
+
+    async def link(self, output, n, i):
+        t = random.randint(0, 5)
+        output += f"link {i}({n}): {t}s "
+        await asyncio.sleep(i)
+        result = f"result {n}-{i}"
+        output += f"link {i}({n}) finished with {result} "
+        return result
+
+    async def chain(self, output, n):
+        link1 = await self.link(output, n, 1)
+        link2 = await self.link(output, n, 2)
+        output += f"chain {n}: {link1} -> {link2} "
+
+    async def gather(self, output, *args):
+        await asyncio.gather(*(self.chain(output, n) for n in args))
+
+    def test_chain(self):
+        args = [1, 2, 3]
+
+        outputs_expected = []
+        outputs_real = []
+
+        # Run the code without QAsyncioEventLoopPolicy
+        random.seed(17)
+        asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
+        start = time.perf_counter()
+        asyncio.run(self.gather(outputs_expected, *args))
+        end_expected = time.perf_counter() - start
+
+        # Run the code with QAsyncioEventLoopPolicy and QtEventLoop
+        random.seed(17)
+        asyncio.set_event_loop_policy(QAsyncioEventLoopPolicy())
+        start = time.perf_counter()
+        asyncio.run(self.gather(outputs_real, *args))
+        end_real = time.perf_counter() - start
+
+        self.assertEqual(outputs_expected, outputs_real)
+        self.assertAlmostEqual(end_expected, end_real, delta=0.5)
+
+
+if __name__ == '__main__':
+    unittest.main()