0
# Utility Functions
1
2
Helper functions for wrapping blocking operations and running asyncio coroutines with Qt event loops, providing convenient utilities for common integration patterns.
3
4
## Capabilities
5
6
### Async Wrapper Function
7
8
Wraps blocking functions to run asynchronously on the native Qt event loop, preventing blocking of the QEventLoop while executing synchronous operations.
9
10
```python { .api }
11
async def asyncWrap(fn, *args, **kwargs):
12
"""
13
Wrap a blocking function as asynchronous and run it on the native Qt event loop.
14
15
The function will be scheduled using a one-shot QTimer which prevents blocking
16
the QEventLoop. This is useful for running blocking operations like modal
17
dialogs within async slots.
18
19
Args:
20
fn: Blocking function to wrap
21
*args: Positional arguments for fn
22
**kwargs: Keyword arguments for fn
23
24
Returns:
25
Result of the wrapped function
26
27
Raises:
28
Exception: Any exception raised by the wrapped function
29
"""
30
```
31
32
#### Usage Example
33
34
```python
35
import asyncio
36
from PySide6.QtWidgets import QApplication, QMessageBox, QWidget, QPushButton, QVBoxLayout
37
from qasync import QEventLoop, asyncSlot, asyncWrap
38
39
class ModalDialogWidget(QWidget):
40
def __init__(self):
41
super().__init__()
42
self.setup_ui()
43
44
def setup_ui(self):
45
layout = QVBoxLayout()
46
47
self.button = QPushButton("Show Modal Dialog")
48
self.button.clicked.connect(self.show_dialog_async)
49
50
layout.addWidget(self.button)
51
self.setLayout(layout)
52
53
@asyncSlot()
54
async def show_dialog_async(self):
55
"""Show modal dialog without blocking the event loop."""
56
print("Before dialog")
57
58
# This would normally block the event loop
59
result = await asyncWrap(
60
lambda: QMessageBox.information(
61
self,
62
"Information",
63
"This is a modal dialog running asynchronously!"
64
)
65
)
66
67
print(f"Dialog result: {result}")
68
print("After dialog - event loop was not blocked!")
69
70
# Can continue with more async operations
71
await asyncio.sleep(1)
72
print("Async operation complete")
73
74
if __name__ == "__main__":
75
app = QApplication([])
76
widget = ModalDialogWidget()
77
widget.show()
78
79
app_close_event = asyncio.Event()
80
app.aboutToQuit.connect(app_close_event.set)
81
82
asyncio.run(app_close_event.wait(), loop_factory=QEventLoop)
83
```
84
85
### Modal Dialog Integration
86
87
Perfect for integrating modal dialogs and other blocking Qt operations within async workflows:
88
89
```python
90
from PySide6.QtWidgets import QFileDialog, QColorDialog, QInputDialog
91
from qasync import asyncWrap, asyncSlot
92
93
class DialogIntegrationWidget(QWidget):
94
@asyncSlot()
95
async def handle_file_operations(self):
96
"""Handle file operations with modal dialogs."""
97
98
# File selection dialog
99
file_path, _ = await asyncWrap(
100
lambda: QFileDialog.getOpenFileName(
101
self,
102
"Select File",
103
"",
104
"Text Files (*.txt);;All Files (*)"
105
)
106
)
107
108
if file_path:
109
print(f"Selected file: {file_path}")
110
111
# Process file asynchronously
112
content = await self.read_file_async(file_path)
113
114
# Show result dialog
115
await asyncWrap(
116
lambda: QMessageBox.information(
117
self,
118
"File Loaded",
119
f"Loaded {len(content)} characters"
120
)
121
)
122
123
@asyncSlot()
124
async def handle_user_input(self):
125
"""Get user input through modal dialogs."""
126
127
# Input dialog
128
text, ok = await asyncWrap(
129
lambda: QInputDialog.getText(
130
self,
131
"Input Dialog",
132
"Enter your name:"
133
)
134
)
135
136
if ok and text:
137
# Color selection dialog
138
color = await asyncWrap(
139
lambda: QColorDialog.getColor(parent=self)
140
)
141
142
if color.isValid():
143
print(f"User: {text}, Color: {color.name()}")
144
145
async def read_file_async(self, file_path):
146
"""Read file content asynchronously."""
147
await asyncio.sleep(0.1) # Simulate async file reading
148
try:
149
with open(file_path, 'r') as f:
150
return f.read()
151
except Exception as e:
152
print(f"Error reading file: {e}")
153
return ""
154
```
155
156
### Asyncio Runner Function
157
158
Provides a convenient way to run asyncio coroutines with automatic Qt event loop integration, supporting both modern and legacy Python versions.
159
160
```python { .api }
161
def run(*args, **kwargs):
162
"""
163
Run asyncio coroutines with Qt event loop integration.
164
165
This function provides a drop-in replacement for asyncio.run that automatically
166
uses QEventLoop as the event loop implementation.
167
168
Python 3.12+: Uses asyncio.run with loop_factory parameter
169
Python <3.12: Uses DefaultQEventLoopPolicy for backward compatibility
170
171
Args:
172
*args: Arguments passed to asyncio.run
173
**kwargs: Keyword arguments passed to asyncio.run
174
175
Returns:
176
Result of the coroutine execution
177
"""
178
```
179
180
#### Usage Example
181
182
```python
183
import asyncio
184
import sys
185
from PySide6.QtWidgets import QApplication
186
from qasync import run
187
188
async def main():
189
"""Main async application logic."""
190
print("Starting async application...")
191
192
# Your async application logic here
193
await asyncio.sleep(1)
194
print("Async operation 1 complete")
195
196
await asyncio.sleep(1)
197
print("Async operation 2 complete")
198
199
print("Application complete!")
200
201
if __name__ == "__main__":
202
# Create Qt application
203
app = QApplication(sys.argv)
204
205
# Method 1: Using qasync.run (works with all Python versions)
206
run(main())
207
208
# Method 2: Modern asyncio.run with loop factory (Python 3.12+)
209
# asyncio.run(main(), loop_factory=QEventLoop)
210
```
211
212
### Application Lifecycle Integration
213
214
Integrate asyncio coroutines with Qt application lifecycle:
215
216
```python
217
import sys
218
from PySide6.QtWidgets import QApplication, QMainWindow
219
from qasync import run
220
221
class AsyncMainWindow(QMainWindow):
222
def __init__(self):
223
super().__init__()
224
self.setWindowTitle("Async Application")
225
self.show()
226
227
async def application_lifecycle():
228
"""Manage the complete application lifecycle."""
229
230
# Initialize Qt application
231
app = QApplication(sys.argv)
232
233
# Create main window
234
window = AsyncMainWindow()
235
236
# Set up application close event
237
app_close_event = asyncio.Event()
238
app.aboutToQuit.connect(app_close_event.set)
239
240
# Start background tasks
241
background_task = asyncio.create_task(background_worker())
242
243
try:
244
# Wait for application to close
245
await app_close_event.wait()
246
finally:
247
# Clean up background tasks
248
background_task.cancel()
249
try:
250
await background_task
251
except asyncio.CancelledError:
252
pass
253
254
print("Application shutdown complete")
255
256
async def background_worker():
257
"""Background task running during application lifetime."""
258
try:
259
while True:
260
print("Background task tick")
261
await asyncio.sleep(5)
262
except asyncio.CancelledError:
263
print("Background task cancelled")
264
raise
265
266
if __name__ == "__main__":
267
run(application_lifecycle())
268
```
269
270
## Advanced Usage Patterns
271
272
### Error Handling with asyncWrap
273
274
```python
275
from qasync import asyncWrap, asyncSlot
276
277
class ErrorHandlingWidget(QWidget):
278
@asyncSlot()
279
async def handle_with_error_checking(self):
280
"""Handle potential errors in wrapped functions."""
281
try:
282
result = await asyncWrap(self.potentially_failing_operation)
283
print(f"Success: {result}")
284
except Exception as e:
285
print(f"Wrapped function failed: {e}")
286
await self.handle_error(e)
287
288
def potentially_failing_operation(self):
289
"""A blocking operation that might fail."""
290
import random
291
if random.random() < 0.5:
292
raise ValueError("Random failure!")
293
return "Operation succeeded"
294
295
async def handle_error(self, error):
296
"""Handle errors asynchronously."""
297
await asyncWrap(
298
lambda: QMessageBox.critical(
299
self,
300
"Error",
301
f"Operation failed: {error}"
302
)
303
)
304
```
305
306
### Event Loop Policy Class
307
308
Advanced event loop policy class for custom Qt event loop configuration (Python <3.12 compatibility).
309
310
```python { .api }
311
class DefaultQEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
312
"""
313
Event loop policy that creates QEventLoop instances by default.
314
315
Used internally by qasync.run() for Python versions prior to 3.12
316
to provide backward compatibility with event loop policies.
317
"""
318
def new_event_loop(self):
319
"""
320
Create a new QEventLoop instance.
321
322
Returns:
323
QEventLoop: New Qt-compatible event loop
324
"""
325
```
326
327
### Custom Event Loop Policies
328
329
For advanced use cases, customize the event loop policy:
330
331
```python
332
import asyncio
333
from qasync import QEventLoop, DefaultQEventLoopPolicy
334
335
class CustomQEventLoopPolicy(DefaultQEventLoopPolicy):
336
def new_event_loop(self):
337
"""Create a customized QEventLoop."""
338
loop = QEventLoop()
339
loop.set_debug(True) # Enable debug mode
340
return loop
341
342
# Use custom policy
343
asyncio.set_event_loop_policy(CustomQEventLoopPolicy())
344
345
async def main():
346
print("Running with custom event loop policy")
347
await asyncio.sleep(1)
348
349
# Run with custom policy
350
asyncio.run(main())
351
```
352
353
### Integration with Asyncio Libraries
354
355
Combine qasync utilities with other asyncio libraries:
356
357
```python
358
import aiohttp
359
from qasync import run, asyncWrap, asyncSlot
360
361
class NetworkWidget(QWidget):
362
@asyncSlot()
363
async def fetch_data_with_progress(self):
364
"""Fetch data and show progress dialog."""
365
366
# Show progress dialog (non-blocking)
367
progress_dialog = QProgressDialog("Fetching data...", "Cancel", 0, 0, self)
368
progress_dialog.show()
369
370
try:
371
async with aiohttp.ClientSession() as session:
372
async with session.get('https://api.example.com/data') as response:
373
data = await response.json()
374
375
# Close progress dialog
376
progress_dialog.close()
377
378
# Show result dialog
379
await asyncWrap(
380
lambda: QMessageBox.information(
381
self,
382
"Success",
383
f"Fetched {len(data)} items"
384
)
385
)
386
387
except Exception as e:
388
progress_dialog.close()
389
await asyncWrap(
390
lambda: QMessageBox.critical(
391
self,
392
"Error",
393
f"Failed to fetch data: {e}"
394
)
395
)
396
397
# Run the application
398
async def main():
399
app = QApplication([])
400
widget = NetworkWidget()
401
widget.show()
402
403
app_close_event = asyncio.Event()
404
app.aboutToQuit.connect(app_close_event.set)
405
406
await app_close_event.wait()
407
408
if __name__ == "__main__":
409
run(main())
410
```