Utility Functions

async def asyncWrap(fn, *args, **kwargs):
    """
    Wrap a blocking function as asynchronous and run it on the native Qt event loop.
    
    The function will be scheduled using a one-shot QTimer which prevents blocking
    the QEventLoop. This is useful for running blocking operations like modal
    dialogs within async slots.
    
    Args:
        fn: Blocking function to wrap
        *args: Positional arguments for fn
        **kwargs: Keyword arguments for fn
        
    Returns:
        Result of the wrapped function
        
    Raises:
        Exception: Any exception raised by the wrapped function
    """

import asyncio
from PySide6.QtWidgets import QApplication, QMessageBox, QWidget, QPushButton, QVBoxLayout
from qasync import QEventLoop, asyncSlot, asyncWrap

class ModalDialogWidget(QWidget):
    def __init__(self):
        super().__init__()
        self.setup_ui()
    
    def setup_ui(self):
        layout = QVBoxLayout()
        
        self.button = QPushButton("Show Modal Dialog")
        self.button.clicked.connect(self.show_dialog_async)
        
        layout.addWidget(self.button)
        self.setLayout(layout)
    
    @asyncSlot()
    async def show_dialog_async(self):
        """Show modal dialog without blocking the event loop."""
        print("Before dialog")
        
        # This would normally block the event loop
        result = await asyncWrap(
            lambda: QMessageBox.information(
                self, 
                "Information", 
                "This is a modal dialog running asynchronously!"
            )
        )
        
        print(f"Dialog result: {result}")
        print("After dialog - event loop was not blocked!")
        
        # Can continue with more async operations
        await asyncio.sleep(1)
        print("Async operation complete")

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

from PySide6.QtWidgets import QFileDialog, QColorDialog, QInputDialog
from qasync import asyncWrap, asyncSlot

class DialogIntegrationWidget(QWidget):
    @asyncSlot()
    async def handle_file_operations(self):
        """Handle file operations with modal dialogs."""
        
        # File selection dialog
        file_path, _ = await asyncWrap(
            lambda: QFileDialog.getOpenFileName(
                self, 
                "Select File", 
                "", 
                "Text Files (*.txt);;All Files (*)"
            )
        )
        
        if file_path:
            print(f"Selected file: {file_path}")
            
            # Process file asynchronously
            content = await self.read_file_async(file_path)
            
            # Show result dialog
            await asyncWrap(
                lambda: QMessageBox.information(
                    self, 
                    "File Loaded", 
                    f"Loaded {len(content)} characters"
                )
            )
    
    @asyncSlot()
    async def handle_user_input(self):
        """Get user input through modal dialogs."""
        
        # Input dialog
        text, ok = await asyncWrap(
            lambda: QInputDialog.getText(
                self, 
                "Input Dialog", 
                "Enter your name:"
            )
        )
        
        if ok and text:
            # Color selection dialog
            color = await asyncWrap(
                lambda: QColorDialog.getColor(parent=self)
            )
            
            if color.isValid():
                print(f"User: {text}, Color: {color.name()}")
    
    async def read_file_async(self, file_path):
        """Read file content asynchronously."""
        await asyncio.sleep(0.1)  # Simulate async file reading
        try:
            with open(file_path, 'r') as f:
                return f.read()
        except Exception as e:
            print(f"Error reading file: {e}")
            return ""

def run(*args, **kwargs):
    """
    Run asyncio coroutines with Qt event loop integration.
    
    This function provides a drop-in replacement for asyncio.run that automatically
    uses QEventLoop as the event loop implementation.
    
    Python 3.12+: Uses asyncio.run with loop_factory parameter
    Python <3.12: Uses DefaultQEventLoopPolicy for backward compatibility
    
    Args:
        *args: Arguments passed to asyncio.run
        **kwargs: Keyword arguments passed to asyncio.run
        
    Returns:
        Result of the coroutine execution
    """

import asyncio
import sys
from PySide6.QtWidgets import QApplication
from qasync import run

async def main():
    """Main async application logic."""
    print("Starting async application...")
    
    # Your async application logic here
    await asyncio.sleep(1)
    print("Async operation 1 complete")
    
    await asyncio.sleep(1)
    print("Async operation 2 complete")
    
    print("Application complete!")

if __name__ == "__main__":
    # Create Qt application
    app = QApplication(sys.argv)
    
    # Method 1: Using qasync.run (works with all Python versions)
    run(main())
    
    # Method 2: Modern asyncio.run with loop factory (Python 3.12+)
    # asyncio.run(main(), loop_factory=QEventLoop)

import sys
from PySide6.QtWidgets import QApplication, QMainWindow
from qasync import run

class AsyncMainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Async Application")
        self.show()

async def application_lifecycle():
    """Manage the complete application lifecycle."""
    
    # Initialize Qt application
    app = QApplication(sys.argv)
    
    # Create main window
    window = AsyncMainWindow()
    
    # Set up application close event
    app_close_event = asyncio.Event()
    app.aboutToQuit.connect(app_close_event.set)
    
    # Start background tasks
    background_task = asyncio.create_task(background_worker())
    
    try:
        # Wait for application to close
        await app_close_event.wait()
    finally:
        # Clean up background tasks
        background_task.cancel()
        try:
            await background_task
        except asyncio.CancelledError:
            pass
        
        print("Application shutdown complete")

async def background_worker():
    """Background task running during application lifetime."""
    try:
        while True:
            print("Background task tick")
            await asyncio.sleep(5)
    except asyncio.CancelledError:
        print("Background task cancelled")
        raise

if __name__ == "__main__":
    run(application_lifecycle())

from qasync import asyncWrap, asyncSlot

class ErrorHandlingWidget(QWidget):
    @asyncSlot()
    async def handle_with_error_checking(self):
        """Handle potential errors in wrapped functions."""
        try:
            result = await asyncWrap(self.potentially_failing_operation)
            print(f"Success: {result}")
        except Exception as e:
            print(f"Wrapped function failed: {e}")
            await self.handle_error(e)
    
    def potentially_failing_operation(self):
        """A blocking operation that might fail."""
        import random
        if random.random() < 0.5:
            raise ValueError("Random failure!")
        return "Operation succeeded"
    
    async def handle_error(self, error):
        """Handle errors asynchronously."""
        await asyncWrap(
            lambda: QMessageBox.critical(
                self, 
                "Error", 
                f"Operation failed: {error}"
            )
        )

class DefaultQEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
    """
    Event loop policy that creates QEventLoop instances by default.
    
    Used internally by qasync.run() for Python versions prior to 3.12
    to provide backward compatibility with event loop policies.
    """
    def new_event_loop(self):
        """
        Create a new QEventLoop instance.
        
        Returns:
            QEventLoop: New Qt-compatible event loop
        """

import asyncio
from qasync import QEventLoop, DefaultQEventLoopPolicy

class CustomQEventLoopPolicy(DefaultQEventLoopPolicy):
    def new_event_loop(self):
        """Create a customized QEventLoop."""
        loop = QEventLoop()
        loop.set_debug(True)  # Enable debug mode
        return loop

# Use custom policy
asyncio.set_event_loop_policy(CustomQEventLoopPolicy())

async def main():
    print("Running with custom event loop policy")
    await asyncio.sleep(1)

# Run with custom policy
asyncio.run(main())

import aiohttp
from qasync import run, asyncWrap, asyncSlot

class NetworkWidget(QWidget):
    @asyncSlot()
    async def fetch_data_with_progress(self):
        """Fetch data and show progress dialog."""
        
        # Show progress dialog (non-blocking)
        progress_dialog = QProgressDialog("Fetching data...", "Cancel", 0, 0, self)
        progress_dialog.show()
        
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get('https://api.example.com/data') as response:
                    data = await response.json()
            
            # Close progress dialog
            progress_dialog.close()
            
            # Show result dialog
            await asyncWrap(
                lambda: QMessageBox.information(
                    self, 
                    "Success", 
                    f"Fetched {len(data)} items"
                )
            )
            
        except Exception as e:
            progress_dialog.close()
            await asyncWrap(
                lambda: QMessageBox.critical(
                    self, 
                    "Error", 
                    f"Failed to fetch data: {e}"
                )
            )

# Run the application
async def main():
    app = QApplication([])
    widget = NetworkWidget()
    widget.show()
    
    app_close_event = asyncio.Event()
    app.aboutToQuit.connect(app_close_event.set)
    
    await app_close_event.wait()

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