Examples
This page shows a few examples that demonstrate the usage of qtinter.
For each example, the source code is listed, and the lines that demonstrate
the usage of qtinter’s API are highlighted.
Color Chooser
This example implements a command-line utility that displays the RGB value of a color chosen by the user from a color dialog.
It demonstrates the use of using_qt_from_asyncio() to add Qt support
to an asyncio-based program.
Sample output:
$ python color.py
#ff8655
Source code (examples/color.py):
1"""Display the RGB code of a color chosen by the user"""
2
3import asyncio
4import qtinter # <-- import module
5from PySide6 import QtWidgets
6
7async def choose_color():
8 dialog = QtWidgets.QColorDialog()
9 dialog.show()
10 future = asyncio.Future()
11 dialog.finished.connect(future.set_result)
12 result = await future
13 if result == QtWidgets.QDialog.DialogCode.Accepted:
14 return dialog.selectedColor().name()
15 else:
16 return None
17
18if __name__ == "__main__":
19 app = QtWidgets.QApplication([])
20 with qtinter.using_qt_from_asyncio(): # <-- enable qt in asyncio code
21 color = asyncio.run(choose_color())
22 if color is not None:
23 print(color)
Digital Clock
This example displays an LCD-style digital clock.
It demonstrates the use of using_asyncio_from_qt() to add
asyncio support to a Qt application.
Sample screenshot:
Source code (examples/clock.py):
1"""Display LCD-style digital clock"""
2
3import asyncio
4import datetime
5import qtinter # <-- import module
6from PySide6 import QtWidgets
7
8class Clock(QtWidgets.QLCDNumber):
9 def __init__(self, parent=None):
10 super().__init__(parent)
11 self.setDigitCount(8)
12
13 def showEvent(self, event):
14 self._task = asyncio.create_task(self._tick())
15
16 def hideEvent(self, event):
17 self._task.cancel()
18
19 async def _tick(self):
20 while True:
21 t = datetime.datetime.now()
22 self.display(t.strftime("%H:%M:%S"))
23 await asyncio.sleep(1.0 - t.microsecond / 1000000 + 0.05)
24
25if __name__ == "__main__":
26 app = QtWidgets.QApplication([])
27
28 widget = Clock()
29 widget.setWindowTitle("qtinter - Digital Clock example")
30 widget.resize(300, 50)
31
32 with qtinter.using_asyncio_from_qt(): # <-- enable asyncio in qt code
33 widget.show()
34 app.exec()
Http Client
This example shows how to download a web page asynchronously using the
httpx module and optionally cancel the download.
Source code:
1"""Demo asyncio http download and cancellation from Qt app."""
2
3import asyncio
4import qtinter
5import sys
6import time
7from PySide6 import QtCore, QtWidgets
8from typing import Optional
9import requests
10import http.server
11import threading
12import httpx
13
14
15def create_http_server():
16 """Create a minimal local http server.
17
18 This server sleeps for 3 seconds before sending back a response.
19 This makes it easier to visualize the difference between synchronous
20 and asynchronous download.
21
22 The server implementation is unrelated to qtinter.
23 """
24 class MyRequestHandler(http.server.BaseHTTPRequestHandler):
25 def do_GET(self):
26 time.sleep(3) # simulate slow response
27 self.send_response(200)
28 self.send_header("Content-type", "text/html")
29 self.end_headers()
30 content = f"You requested {self.path}".encode("utf-8")
31 self.wfile.write(content)
32
33 server = http.server.ThreadingHTTPServer(("127.0.0.1", 0),
34 MyRequestHandler)
35 thread = threading.Thread(target=server.serve_forever)
36 thread.start()
37 return server
38
39
40class MyWidget(QtWidgets.QWidget):
41 def __init__(self):
42 super().__init__()
43
44 self.setWindowTitle("qtinter - Http Client Example")
45
46 # Show an indefinite progress bar to visualize whether the Qt event
47 # loop is blocked -- the progress bar freezes if the Qt event loop
48 # is blocked.
49 self._progress = QtWidgets.QProgressBar()
50 self._progress.setRange(0, 0)
51
52 # URL input.
53 self._url = QtWidgets.QLineEdit(self)
54
55 # The 'Sync GET' button downloads the web page synchronously,
56 # which blocks the Qt event loop. The progress bar freezes
57 # when this button is clicked until the download completes.
58 self._sync_button = QtWidgets.QPushButton("Sync GET")
59 self._sync_button.clicked.connect(self.sync_download)
60
61 # The 'Async GET' button downloads the web page asynchronously.
62 # The progress bar keeps running while the download is in progress.
63 self._async_button = QtWidgets.QPushButton("Async GET")
64
65 # [DEMO] To connect an async function to the clicked signal, wrap
66 # the async function with qtinter.asyncslot.
67 self._async_button.clicked.connect(
68 qtinter.asyncslot(self.async_download))
69
70 # [DEMO] When an async download is in progress, _async_task is set
71 # to the task executing the download, so that it may be cancelled
72 # by clicking the 'Cancel' button.
73 self._async_task: Optional[asyncio.Task] = None
74
75 # The 'Cancel' button is enabled when async download is in progress.
76 self._cancel_button = QtWidgets.QPushButton("Cancel")
77 self._cancel_button.setEnabled(False)
78 self._cancel_button.clicked.connect(self.cancel_async_download)
79
80 # Response from the http server is shown in the below box.
81 self._output = QtWidgets.QTextEdit(self)
82 self._output.setReadOnly(True)
83
84 # Set up layout.
85 self._buttons = QtWidgets.QHBoxLayout()
86 self._buttons.setContentsMargins(0, 0, 0, 0)
87 self._buttons.setSpacing(5)
88 self._buttons.addWidget(self._sync_button)
89 self._buttons.addWidget(self._async_button)
90 self._buttons.addWidget(self._cancel_button)
91 self._layout = QtWidgets.QVBoxLayout(self)
92 self._layout.setContentsMargins(10, 10, 10, 10)
93 self._layout.setSpacing(5)
94 self._layout.addWidget(self._progress)
95 self._layout.addWidget(self._url)
96 self._layout.addLayout(self._buttons)
97 self._layout.addWidget(self._output)
98
99 # Start a local HTTP server to simulate slow response. The server
100 # runs in a separate thread.
101 self._server = create_http_server()
102
103 # Set default URL to the locally-run http server.
104 self._url.setText("http://{}:{}/dummy"
105 .format(*self._server.server_address))
106
107 def closeEvent(self, event):
108 # Cancel the running download task if one exists.
109 if self._async_task is not None:
110 self._async_task.cancel()
111
112 # Shut down the local HTTP server. The shutdown() call blocks;
113 # you may observe a short freeze of the progress bar.
114 self._server.shutdown()
115 event.accept()
116
117 def sync_download(self):
118 # When the 'Sync GET' button is clicked, download the web page
119 # using the (blocking) requests library. This has two drawbacks:
120 # 1. The GUI freezes during the download. For example, the progress
121 # bar stops updating.
122 # 2. Buttons remain clickable even if disabled before download starts
123 # and re-enabled after download completes. A workaround seems to
124 # be waiting for 10 milliseconds before re-enabling the button;
125 # hard-coding a timeout is certainly not ideal.
126 url = self._url.text()
127 self._async_button.setEnabled(False)
128 try:
129 response = requests.get(url)
130 self._output.setText(response.text)
131 finally:
132 QtCore.QTimer.singleShot(
133 10, lambda: self._async_button.setEnabled(True))
134
135 # [DEMO] asynchronous slot for the 'Async GET' button.
136 async def async_download(self):
137 # Store the asyncio task wrapping the running coroutine so that
138 # it may be cancelled by calling its cancel() method.
139 self._async_task = asyncio.current_task()
140 assert self._async_task is not None
141
142 # Update GUI elements -- this is a common pattern in event handling.
143 # Because qtinter.asyncslot() executes the coroutine immediately
144 # until the first yield, the changes take effect immediately,
145 # eliminating potential race conditions. The actual GUI repainting
146 # happens after the first yield when control is returned to the Qt
147 # event loop.
148 self._sync_button.setEnabled(False)
149 self._async_button.setEnabled(False)
150 self._cancel_button.setEnabled(True)
151 self._output.clear()
152
153 try:
154 # Download web page using httpx library. The httpx library
155 # works with both asyncio and trio, and uses anyio to detect
156 # the type of the running event loop. That httpx works with
157 # qtinter shows that qtinter's event loop behaves like
158 # an asyncio event loop.
159 async with httpx.AsyncClient() as client:
160 url = self._url.text()
161 response = await client.get(url)
162 # TODO: test asyncgen close
163 body = await response.aread()
164 self._output.setText(body.decode("utf-8"))
165
166 except asyncio.CancelledError:
167 # Catching a CancelledError indicates the task is cancelled.
168 # This can happen either because the user clicked the 'Cancel'
169 # button, or because the window is closed.
170 if self.isVisible():
171 # Cancelled by the CANCEL button -- display a message box.
172 # Use qtinter.modal() to avoid blocking the asyncio event
173 # loop because QMessageBox.information() creates a nested
174 # Qt event loop.
175 await qtinter.modal(QtWidgets.QMessageBox.information)(
176 self, "Note", "Download cancelled by user!")
177 else:
178 # Cancelled by window close -- do nothing. By now the Qt
179 # event loop may have terminated already, and the underlying
180 # QiBaseEventLoop may be operating in NATIVE mode.
181 pass
182
183 finally:
184 # Restore GUI element states.
185 self._cancel_button.setEnabled(False)
186 self._async_button.setEnabled(True)
187 self._sync_button.setEnabled(True)
188 self._async_task = None
189
190 # [DEMO] When the 'Cancel' button is clicked, cancel the async download.
191 def cancel_async_download(self):
192 # The 'Cancel' button is enabled only if the async download is
193 # in progress, so the task object must be set.
194 assert self._async_task is not None
195
196 # Initiate cancellation request. This throws asyncio.CancelledError
197 # into the (suspended) coroutine, which must catch the exception and
198 # perform actual cancellation. (Note that it is possible for the
199 # task to be done before CancelledError is thrown, as a cancellation
200 # request is scheduled by call_soon() and it might so happen that
201 # a task completion callback is scheduled before it.) Cancelling
202 # a done task has no effect.
203 self._async_task.cancel()
204
205
206def main():
207 # Create a QApplication instance, as usual.
208 app = QtWidgets.QApplication([])
209
210 # Create widgets, as usual.
211 widget = MyWidget()
212 widget.resize(400, 200)
213 widget.show()
214
215 # [DEMO] To enable asyncio-based components from Qt-driven code,
216 # enclose app.exec() inside the qtinter.using_asyncio_from_qt()
217 # context manager. This context manager takes care of starting
218 # up and shutting down an asyncio-compatible logical event loop.
219 with qtinter.using_asyncio_from_qt():
220 sys.exit(app.exec())
221
222
223if __name__ == "__main__":
224 main()
Read Out
This example implements a command line utility that reads out the text from standard input. It is a cross-platform version of the macOS say command.
The example demonstrates the use of using_qt_from_asyncio() to
use a Qt component (QtTextToSpeech) in asyncio-driven code, and the
use of asyncsignal() to wait for a Qt signal.
Note
On Unix, press Ctrl+D to terminate input. On Windows, press Ctrl+Z.
Sample output (on macOS 12):
$ python read_out.py -h
usage: read_out.py [options]
Read out text from stdin.
Options:
-e Echo each line before reading it out
-h Show this screen and exit
-l locale One of en_US, fr_FR (default: en_US)
-p pitch Number between -1.0 and +1.0 (default: 0.0)
-r rate Number between -1.0 and +1.0 (default: 0.0)
-v voice One of Alex, Fiona, Fred, Samantha, Victoria (default: Alex)
Source code:
1"""Read out input using QTextToSpeech"""
2
3import asyncio
4import getopt
5import os
6import qtinter
7import sys
8from PyQt6 import QtCore, QtTextToSpeech
9
10
11async def read_out(engine: QtTextToSpeech.QTextToSpeech, echo=False):
12 while True:
13 try:
14 line = await asyncio.get_running_loop().run_in_executor(None, input)
15 except EOFError:
16 break
17 if echo:
18 print(line)
19 engine.say(line)
20 if engine.state() == QtTextToSpeech.QTextToSpeech.State.Speaking:
21 # If the line contains no speakable content, state remains Ready.
22 state, = await qtinter.asyncsignal(engine.stateChanged)
23 assert state == QtTextToSpeech.QTextToSpeech.State.Ready
24
25
26def main():
27 if sys.platform == 'darwin':
28 # QtTextToSpeech requires QEventDispatcherCoreFoundation on macOS.
29 # Set QT_EVENT_DISPATCHER_CORE_FOUNDATION or use QtGui.QGuiApplication.
30 os.environ['QT_EVENT_DISPATCHER_CORE_FOUNDATION'] = '1'
31 app = QtCore.QCoreApplication([])
32
33 engine = QtTextToSpeech.QTextToSpeech()
34 locales = dict((l.name(), l) for l in engine.availableLocales())
35 voices = dict((v.name(), v) for v in engine.availableVoices())
36 usage = (f"usage: {sys.argv[0]} [options]\n"
37 f"Read out text from stdin.\n"
38 f"Options:\n"
39 f" -e Echo each line before reading it out\n"
40 f" -h Show this screen and exit\n"
41 f" -l locale One of {', '.join(sorted(locales))} "
42 f"(default: {engine.locale().name()})\n"
43 f" -p pitch Number between -1.0 and +1.0 (default: 0.0)\n"
44 f" -r rate Number between -1.0 and +1.0 (default: 0.0)\n"
45 f" -v voice One of {', '.join(sorted(voices))} "
46 f"(default: {engine.voice().name()})\n")
47
48 try:
49 args, rest = getopt.getopt(sys.argv[1:], "ehl:p:r:v:")
50 except getopt.error:
51 print(usage, file=sys.stderr)
52 return 1
53
54 if rest:
55 print(usage, file=sys.stderr)
56 return 1
57
58 echo = False
59 for opt, val in args:
60 if opt == "-e":
61 echo = True
62 elif opt == "-h":
63 print(usage)
64 return 0
65 elif opt == "-l":
66 engine.setLocale(locales[val])
67 elif opt == "-p":
68 engine.setPitch(float(val))
69 elif opt == "-r":
70 engine.setRate(float(val))
71 elif opt == "-v":
72 engine.setVoice(voices[val])
73 else:
74 print(usage, file=sys.stderr)
75 return 1
76
77 with qtinter.using_qt_from_asyncio():
78 asyncio.run(read_out(engine, echo))
79
80
81if __name__ == "__main__":
82 sys.exit(main())
Where am I
This example implements a command line utility that prints the current geolocation.
It demonstrates the use of using_qt_from_asyncio() to use
a Qt component (QtPositioning) in asyncio-driven code.
It also demonstrates the use of asyncsignal() and
multisignal() to wait for the first of multiple Qt signals.
Sample output:
$ python where_am_i.py
12° 34' 56.7" N, 98° 76' 54.3" E, 123.456m
Source code:
1"""Report current geolocation"""
2
3import asyncio
4import os
5import qtinter
6import sys
7from PyQt6 import QtCore, QtPositioning
8
9
10async def get_location() -> str:
11 """Return the current location as a string."""
12
13 # A QGeoPositionInfoSource object needs a parent to control its lifetime.
14 app = QtCore.QCoreApplication.instance()
15 source = QtPositioning.QGeoPositionInfoSource.createDefaultSource(app)
16 if source is None:
17 raise RuntimeError("No QGeoPositionInfoSource is available")
18
19 # This is a pattern to call a Qt method *after* installing signal handlers.
20 asyncio.get_running_loop().call_soon(source.requestUpdate, 0)
21
22 # Wait for position update or error message.
23 which, [result] = await qtinter.asyncsignal(qtinter.multisignal({
24 source.positionUpdated: "ok",
25 source.errorOccurred: "error",
26 }))
27
28 if which == "ok":
29 position: QtPositioning.QGeoPositionInfo = result
30 return position.coordinate().toString()
31 else:
32 error: QtPositioning.QGeoPositionInfoSource.Error = result
33 raise RuntimeError(f"Cannot obtain geolocation: {error}")
34
35
36def main():
37 if sys.platform == 'darwin':
38 # QtPositioning requires QEventDispatcherCoreFoundation on macOS.
39 # Set QT_EVENT_DISPATCHER_CORE_FOUNDATION or use QtGui.QGuiApplication.
40 os.environ['QT_EVENT_DISPATCHER_CORE_FOUNDATION'] = '1'
41 app = QtCore.QCoreApplication([])
42 with qtinter.using_qt_from_asyncio():
43 print(asyncio.run(get_location()))
44
45
46if __name__ == "__main__":
47 main()