How to set up a qasync event loop in another thread?

Question

I am developing an asynchronous PyQt application - meaning, I need both asyncio and PyQt. Since both asyncio and PyQt use their own event loops, this would be impossible without qasync, which is created to solve this problem. Here is how you use it:

app = QApplication(sys.argv)
asyncio.set_event_loop(qasync.QEventLoop(app))
exit(app.exec_())

Now, I want to move all my logic into a separate "worker" thread. To do this, I create a QThread, which automatically sets up a new Qt event loop for this thread. But I need to be able to use asyncio in that thread, too, so I have to do the same qasync trick for the worker thread.

But how do I do it? To my understanding, qasync.QEventLoop(app) would get the application's main event loop, not the thread's loop. And you can't do qasync.QEventLoop(thread).

Here is a code example.

import logging
import sys
import datetime
import asyncio
import qasync

from PyQt5.QtCore import QObject, pyqtSlot, QThread
from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QWidget, QVBoxLayout


class Logic(QObject):  # Logic to run in a separate thread
    enabled = False

    @pyqtSlot()
    def start(self):
        logger.debug('Start called')
        self.enabled = True
        asyncio.create_task(self.process())

    @pyqtSlot()
    def stop(self):
        logger.debug('Stop called')
        self.enabled = False

    async def process(self):
        while self.enabled:
            logger.debug(f'Processing ({datetime.datetime.now()})...')
            await asyncio.sleep(0.5)


if __name__ == '__main__':
    logging.basicConfig(format='%(levelname)s:%(name)s: %(message)s')
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)

    app = QApplication(sys.argv)
    asyncio.set_event_loop(qasync.QEventLoop(app))

    logic = Logic()

    # Move logic to another thread
    # thread = QThread()
    # logic.moveToThread(thread)
    # thread.start()

    window = QMainWindow()
    window.setCentralWidget(QWidget())
    window.centralWidget().setLayout(QVBoxLayout())

    window.centralWidget().layout().addWidget(QPushButton(text='Start', clicked=logic.start))
    window.centralWidget().layout().addWidget(QPushButton(text='Stop', clicked=logic.stop))

    window.show()

    logger.debug('Launching the application...')
    exit(app.exec_())

If you run it as it is, it works - in a single thread. But if you uncomment the "Move logic to another thread" block, then it fails, because there is RuntimeError: no running event loop for asyncio.create_task().

How do I fix this? I guess the desired behavior is obvious - it should behave the same as it did with a single thread, except everything inside logic should run in a separate thread.

A sub-question is, what exactly does qasync do? Is my understanding correct that it somehow allows to use the same main Qt event loop for asyncio? So qasync.QEventLoop(app) returns a... some kind of asyncio-compatible proxy object to the existing Qt event loop?..

Answer 1

I think the problem is due to that you are setting QEventLoop created from the QApplication (which is in main thread). Here is my version to show how to run asyncio from a separate thread:

import logging
import sys
import datetime
import asyncio
import qasync
import threading
from PyQt5.QtCore import QObject, pyqtSlot, QThread, pyqtSignal
from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QWidget, QVBoxLayout


class Logic(QObject):  # Logic to run in a separate thread
    enabled = False

    @pyqtSlot()
    def start(self):
        logger.debug('Start called')
        self.enabled = True
        asyncio.create_task(self.process())

    @pyqtSlot()
    def stop(self):
        logger.debug('Stop called')
        self.enabled = False

    async def process(self):
        while self.enabled:
            # demonstrate that we can emit a signal from a thread
            self.message.emit(f'Processing ({datetime.datetime.now()}), worker thread id: {threading.get_ident()}...')
            await asyncio.sleep(0.5)

    # signal declaration
    message = pyqtSignal(str)

# subclass QThread. override the run function to create an event loop and run forever
class WorkerThread(QThread):
    def run(self):
        loop = qasync.QEventLoop(self)
        asyncio.set_event_loop(loop)
        loop.run_forever()

if __name__ == '__main__':
    logging.basicConfig(format='%(levelname)s:%(name)s: %(message)s')
    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)

    app = QApplication(sys.argv)
    logger.debug(f'main thread id: {threading.get_ident()}')

    # Move logic to another thread
    logic = Logic()    
    thread = WorkerThread()
    logic.moveToThread(thread)
    thread.start()

    window = QMainWindow()
    window.setCentralWidget(QWidget())
    window.centralWidget().setLayout(QVBoxLayout())

    window.centralWidget().layout().addWidget(QPushButton(text='Start', clicked=logic.start))
    window.centralWidget().layout().addWidget(QPushButton(text='Stop', clicked=logic.stop))
    # connect logic in workerThread to lambda function in this thread
    logic.message.connect(lambda msg: logger.debug(f'current thread: {threading.get_ident()}, {msg}'))

    window.show()

    logger.debug('Launching the application...')
    exit(app.exec_())