How can Flask rest endpoint communicate to a pyqt application running on the same python program

Question

I am trying to create a simple rest endpoint, that when hit, it changes the status of a tray icon that is installed by the program itself.

I believe that I need Flask and the QApplication to run in different threads and I tried to do this.

However I am not familiar with python and most probably I am doing something wrong below:

import sys
import asyncio
from PyQt6.QtGui import QIcon
from PyQt6.QtWidgets import (
    QSystemTrayIcon, 
    QApplication, 
    QWidget
    )
from flask import Flask 



async def gui():
    app = QApplication(sys.argv)
    w = QWidget()
    trayIcon = QSystemTrayIcon(QIcon("disconnected.png"), w)
    trayIcon.show()        
    print("running gui")
    return app.exec()


async def rest():
    app1 = Flask(__name__) 
    
    @app1.route("/status/connected") 
    def setStatusConnected(): 
        return "I will signal gui to change icon to connected"
    @app1.route("/status/disconnected") 
    def setStatusDisconnected(): 
        return "I will signal gui to change icon to disconnected"

    print("creating rest endpoint")
    return app1.run()  



async def main():
    return await asyncio.gather(gui(), rest() )

if __name__ == "__main__":
    asyncio.run(main())

When I run the code above I see only "running gui" and I can see the tray icon being installed to the status bar. I do not see "creating rest endpoint"

If I comment-out the "return app.exec()". I see the "creating rest endpoint" and I can access the endpoints, but I don't see the the tray icon.

Whay is am I missing?

extra info: this is my Pipfile (Using pipenv)

[[source]]
url = "https://pypi.org/simple"
verify_ssl = true
name = "pypi"

[packages]
pyqt6 = "*"
flask = {extras = ["async"], version = "*"}

[dev-packages]

[requires]
python_version = "3.10"

Answer 1

ayncio is useful when programs uses await inside its code - inside app.run() and inside app.exec() - but as I know Qt and Flask don't do this. (but maybe Flask has version which uses asyncio)

You may need to use threading to run programs in separated threads (or one program run in separated thread and other run in current thread).

Version 1: both programs in separated threads

def gui(): # without `async
    #...

def rest(): # without `async
    #...

if __name__ == "__main__":
    from threading import Thread
    
    t1 = Thread(target=gui)
    t2 = Thread(target=rest)
    
    t1.start()
    t2.start()
    
    # it blocks this function and wait for end of threads
    t1.join()
    t2.join()  

Version 2: one programs in separated thread and other in current thread

def gui(): # without `async
    #...

def rest(): # without `async
    #...

if __name__ == "__main__":
    from threading import Thread
    
    t1 = Thread(target=gui)
    t1.start()

    rest() # it blocks this function

    # wait for end of thread
    t1.join()

---

Full working code

import sys
import asyncio
from PyQt6.QtGui import QIcon
from PyQt6.QtWidgets import (
    QSystemTrayIcon, 
    QApplication, 
    QWidget
    )
from flask import Flask 


def gui():
    app = QApplication(sys.argv)
    
    w = QWidget()
    trayIcon = QSystemTrayIcon(QIcon("disconnected.png"), w)
    trayIcon.show()        
    
    print("running gui")
    
    app.exec()


def rest():
    app = Flask(__name__) 
    
    @app.route("/connect") 
    def connect(): 
        return "I will signal gui to change icon to connected"
        
    @app.route("/disconnect") 
    def disconnect(): 
        return "I will signal gui to change icon to disconnected"

    print("creating rest endpoint")
    
    app.run()  


def version_1():
    from threading import Thread
    
    t1 = Thread(target=gui)
    t2 = Thread(target=rest)
    
    t1.start()
    t2.start()
    
    # it blocks this function and wait for end of threads
    t1.join()
    t2.join()

def version_2()
    from threading import Thread
    
    t1 = Thread(target=gui)
    t1.start()

    rest() # it blocks this function

    # wait for end of thread
    t1.join()


if __name__ == "__main__":
    #version_1()
    version_2()

---

EDIT:

I not sure if these threads may use Qt signals to communicates. They may need to use queue to comunicate, and Qt may need QTimer to check periodically if there is new information in queue.

Other idea: Flask should have url which returns current status (information if it is connected or disconnected) and Qt could use QTime with probably QtNetwork.QNetworkRequest(QtCore.QUrl(url)) to check this status. This way you can always use other GUI framework to check status. Or you can check status in text console (ie. using curl, wget)

If you plan to use Flask only to change status in Qt then maybe you should use QtNetwork to run server directly in Qt without using Flask.

Answer 2

Following furas answer I provide a working solution of running both flask and pyqt app that installs a system tray icon with the ability to update the icon status based on received rest requests:

• used threading instead of asyncio
• used a Qtimer inside gui that checks every second a mutable variable set by rest endpoint and updates

import sys
from PyQt6.QtGui import QIcon
from PyQt6.QtCore import QTimer
from threading import Thread
from PyQt6.QtWidgets import (
    QSystemTrayIcon, 
    QApplication, 
    QWidget
    )
from flask import Flask 

class MutableValue:
    def __init__(self, v):
        self.value = v
event = MutableValue('disconnected')

def gui():
    def iconFromValue(value):
        return QIcon(value+".png");
    def checkStatus():
        #TODO: check if value changed before applying icon
        trayIcon.setIcon(iconFromValue(event.value)) 
    app = QApplication(sys.argv)
    # timer to check every second for an update from rest
    timer = QTimer()
    timer.timeout.connect(checkStatus)
    timer.start(1000)
    w = QWidget()
    trayIcon = QSystemTrayIcon(iconFromValue(event.value), w)
    trayIcon.show()
    print("running gui..")
    return app.exec()


def rest():
    app1 = Flask(__name__) 
    
    @app1.route("/status/connected") 
    def setStatusConnected(): 
        event.value='connected'
        return "Changed status to connected."
        
    @app1.route("/status/disconnected") 
    def setStatusDisconnected(): 
        event.value='disconnected'
        return "Changed status to disconnected."
    print("creating rest endpoint..")
    return app1.run()  



def main():
    g=Thread(target=gui, daemon=True)
    r=Thread(target=rest, daemon=True)
    
    g.start()
    r.start()
    g.join()
    r.join()

if __name__ == "__main__":
    sys.exit(main())