tessl/pypi-qasync

Python library for using asyncio in Qt-based applications

—

Pending

Overview

Eval results

Files

Async Decorators

Name: tessl/pypi-qasync
Author: tessl

Decorators that enable Qt slots and event handlers to run as async coroutines, providing seamless integration between asyncio's async/await syntax and Qt's signal-slot system.

Capabilities

Async Slot Decorator

Converts Qt slots to run asynchronously on the asyncio event loop, allowing signal handlers to use async/await syntax without blocking the UI thread.

def asyncSlot(*args, **kwargs):
    """
    Make a Qt slot run asynchronously on the asyncio loop.
    
    This decorator allows Qt slots to be defined as async functions that will
    be executed as coroutines when the signal is emitted.
    
    Args:
        *args: Signal parameter types (same as Qt's Slot decorator)
        **kwargs: Additional keyword arguments for Qt's Slot decorator
        
    Returns:
        Decorator function that wraps the async slot
        
    Raises:
        TypeError: If slot signature doesn't match signal parameters
    """

Usage Example

import asyncio
import sys
from PySide6.QtWidgets import QApplication, QWidget, QPushButton, QVBoxLayout, QLabel
from qasync import QEventLoop, asyncSlot

class AsyncWidget(QWidget):
    def __init__(self):
        super().__init__()
        self.setup_ui()
        
    def setup_ui(self):
        layout = QVBoxLayout()
        
        self.status_label = QLabel("Ready")
        self.button = QPushButton("Start Async Task")
        
        # Connect button to async slot
        self.button.clicked.connect(self.handle_button_click)
        
        layout.addWidget(self.status_label)
        layout.addWidget(self.button)
        self.setLayout(layout)
    
    @asyncSlot()
    async def handle_button_click(self):
        """Handle button click asynchronously."""
        self.status_label.setText("Processing...")
        self.button.setEnabled(False)
        
        try:
            # Simulate async work (network request, file I/O, etc.)
            await asyncio.sleep(2)
            result = await self.fetch_data()
            
            self.status_label.setText(f"Complete: {result}")
        except Exception as e:
            self.status_label.setText(f"Error: {e}")
        finally:
            self.button.setEnabled(True)
    
    async def fetch_data(self):
        # Simulate async data fetching
        await asyncio.sleep(1)
        return "Data loaded successfully"

if __name__ == "__main__":
    app = QApplication(sys.argv)
    widget = AsyncWidget()
    widget.show()
    
    app_close_event = asyncio.Event()
    app.aboutToQuit.connect(app_close_event.set)
    
    asyncio.run(app_close_event.wait(), loop_factory=QEventLoop)

Signal Parameter Handling

The decorator automatically handles signal parameter matching, removing excess parameters if the slot signature doesn't match the signal.

Usage Example

from PySide6.QtCore import QTimer, pyqtSignal
from qasync import asyncSlot

class TimerWidget(QWidget):
    # Custom signal with parameters
    data_received = pyqtSignal(str, int)
    
    def __init__(self):
        super().__init__()
        
        # Connect signals with different parameter counts
        self.data_received.connect(self.handle_data_full)
        self.data_received.connect(self.handle_data_partial)
        
        # Timer signal (no parameters)
        timer = QTimer()
        timer.timeout.connect(self.handle_timeout)
        timer.start(1000)
    
    @asyncSlot(str, int)  # Matches signal parameters exactly
    async def handle_data_full(self, message, value):
        print(f"Full data: {message}, {value}")
        await asyncio.sleep(0.1)
    
    @asyncSlot(str)  # Only uses first parameter
    async def handle_data_partial(self, message):
        print(f"Partial data: {message}")
        await asyncio.sleep(0.1)
    
    @asyncSlot()  # No parameters
    async def handle_timeout(self):
        print("Timer tick")
        await asyncio.sleep(0.1)

Async Close Handler

Decorator for enabling async cleanup operations before application or widget closure.

def asyncClose(fn):
    """
    Allow async code to run before application or widget closure.
    
    This decorator wraps close event handlers to allow async operations
    during the shutdown process.
    
    Args:
        fn: Async function to run during close event
        
    Returns:
        Wrapped function that handles the close event
    """

Usage Example

import asyncio
from PySide6.QtWidgets import QMainWindow, QApplication
from qasync import QEventLoop, asyncClose

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Async Close Example")
        self.active_connections = []
        
    async def cleanup_connections(self):
        """Clean up active network connections."""
        print("Closing connections...")
        for connection in self.active_connections:
            await connection.close()
        print("All connections closed")
    
    async def save_user_data(self):
        """Save user data asynchronously."""
        print("Saving user data...")
        await asyncio.sleep(1)  # Simulate async save operation
        print("User data saved")
    
    @asyncClose
    async def closeEvent(self, event):
        """Handle window close event asynchronously."""
        print("Application closing...")
        
        # Perform async cleanup operations
        await self.cleanup_connections()
        await self.save_user_data()
        
        print("Cleanup complete")
        # Event is automatically accepted after async operations complete

if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    
    app_close_event = asyncio.Event()
    app.aboutToQuit.connect(app_close_event.set)
    
    asyncio.run(app_close_event.wait(), loop_factory=QEventLoop)

Error Handling

Both decorators provide proper error handling for async operations:

Async Slot Error Handling

Exceptions in async slots are handled through Python's standard exception handling mechanism:

import sys
from qasync import asyncSlot

class ErrorHandlingWidget(QWidget):
    @asyncSlot()
    async def risky_operation(self):
        try:
            await self.might_fail()
        except Exception as e:
            print(f"Slot error: {e}")
            # Handle error appropriately
            self.show_error_message(str(e))
    
    async def might_fail(self):
        # This might raise an exception
        raise ValueError("Something went wrong!")
    
    def show_error_message(self, message):
        # Show error to user
        from PySide6.QtWidgets import QMessageBox
        QMessageBox.critical(self, "Error", message)

Exception Propagation

Unhandled exceptions in async slots are propagated through the standard Python exception handling system (sys.excepthook) and can be logged or handled globally:

import sys
import logging

# Set up global exception handler
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)

def handle_exception(exc_type, exc_value, exc_traceback):
    if issubclass(exc_type, KeyboardInterrupt):
        sys.__excepthook__(exc_type, exc_value, exc_traceback)
        return
    
    logger.error("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))

sys.excepthook = handle_exception

Advanced Usage Patterns

Connecting Multiple Signals

class MultiSignalWidget(QWidget):
    def __init__(self):
        super().__init__()
        
        button1 = QPushButton("Button 1")
        button2 = QPushButton("Button 2")
        
        # Same async slot handles multiple signals
        button1.clicked.connect(self.handle_click)
        button2.clicked.connect(self.handle_click)
    
    @asyncSlot()
    async def handle_click(self):
        sender = self.sender()
        print(f"Async click from: {sender.text()}")
        await asyncio.sleep(0.5)

Chaining Async Operations

class ChainedOperationsWidget(QWidget):
    @asyncSlot()
    async def start_workflow(self):
        """Chain multiple async operations."""
        try:
            step1_result = await self.step_one()
            step2_result = await self.step_two(step1_result)
            final_result = await self.step_three(step2_result)
            
            self.display_result(final_result)
        except Exception as e:
            self.handle_workflow_error(e)
    
    async def step_one(self):
        await asyncio.sleep(1)
        return "Step 1 complete"
    
    async def step_two(self, input_data):
        await asyncio.sleep(1)
        return f"{input_data} -> Step 2 complete"
    
    async def step_three(self, input_data):
        await asyncio.sleep(1)
        return f"{input_data} -> Step 3 complete"

Task Cancellation

class CancellableTaskWidget(QWidget):
    def __init__(self):
        super().__init__()
        self.current_task = None
        
        self.start_button = QPushButton("Start")
        self.cancel_button = QPushButton("Cancel")
        
        self.start_button.clicked.connect(self.start_long_task)
        self.cancel_button.clicked.connect(self.cancel_task)
    
    @asyncSlot()
    async def start_long_task(self):
        if self.current_task and not self.current_task.done():
            return  # Task already running
        
        self.current_task = asyncio.create_task(self.long_running_operation())
        
        try:
            result = await self.current_task
            print(f"Task completed: {result}")
        except asyncio.CancelledError:
            print("Task was cancelled")
    
    @asyncSlot()
    async def cancel_task(self):
        if self.current_task and not self.current_task.done():
            self.current_task.cancel()
    
    async def long_running_operation(self):
        for i in range(10):
            await asyncio.sleep(1)
            print(f"Progress: {i + 1}/10")
        return "Long task complete"

Install with Tessl CLI