qasync

repository·master·Indexed 19 days ago

https://github.com/cabbagedevelopment/qasync

A Python library providing a PEP 3156 event loop implementation that allows asyncio coroutines to be used within PyQt and PySide applications. It features the QEventLoop for integrating asyncio with the Qt event loop, decorators like @asyncSlot and @asyncClose for handling asynchronous Qt slots and cleanup, and QThreadExecutor for offloading CPU-intensive tasks to background threads.

Tokens
2.2K
Snippets
11
Records
13
Agent score
15%

What's inside qasync

  1. Execute CPU-intensive tasks with run_in_executor

    master
    QEventLoop provides a run_in_executor method that is functionally identical to asyncio's version. This allows you to run CPU-intensive tasks in parallel without blocking the Qt event loop. By default, it uses QThreadExecutor, but you can provide any class that implements the concurrent.futures.Executor interface.
  2. Use async code in closeEvent with @asyncClose

    master

    To perform asynchronous cleanup or logic when a window is closing, decorate the closeEvent method with @asyncClose.

    from qasync import asyncClose
    from PySide6.QtGui import QCloseEvent
    
    class MainWindow(QWidget):
        @asyncClose
        async def closeEvent(self, event: QCloseEvent):
            self.button.setText("Closing...")
            await asyncio.sleep(1)
  3. Use async code in Qt slots with @asyncSlot

    master

    To use an async function as a Qt slot (e.g., responding to a button click), decorate the method with @asyncSlot(). This allows you to await coroutines directly within the slot logic.

    from qasync import asyncSlot
    
    class MainWindow(QWidget):
        @asyncSlot()
        async def onButtonClicked(self):
            self.button.setText("Loading...")
            await asyncio.sleep(1)
            self.button.setText("Load")
  4. Run the asyncio event loop with QEventLoop

    master

    To integrate the asyncio event loop with the Qt event loop, use the loop_factory argument in asyncio.run() (for Python 3.11+) or use qasync.run() (for Python 3.10 and older).

    import asyncio
    import sys
    from PySide6.QtWidgets import QApplication
    import qasync
    from qasync import QEventLoop
    
    async def main(app):
        # Your application logic here
        pass
    
    if __name__ == "__main__":
        app = QApplication(sys.argv)
        
        # For Python 3.11 or newer:
        asyncio.run(main(app), loop_factory=QEventLoop)
        
        # For Python 3.10 or older:
        # qasync.run(main(app))
  5. Configure Qt API via environment variable

    master

    You can control which Qt binding qasync uses by setting the QT_API environment variable. This is useful if multiple bindings (e.g., PyQt5 and PySide6) are installed on your system.

    Supported values (case-insensitive):

    • PyQt6
    • PyQt5
    • PySide6
    • PySide2

    If QT_API is not set, qasync will attempt to use the binding that is already imported in sys.modules, or search for the first available binding in the preference order: PyQt6, PyQt5, PySide6, PySide2.

    # Example: Forcing PySide6
    export QT_API=PySide6
    python my_qt_app.py
  6. Run asyncio code with run()

    master

    The run() function is a convenience wrapper around asyncio.run(). It automatically sets the QEventLoop as the event loop policy, ensuring that the provided coroutine runs on the Qt event loop.

    • Python 3.12+: Uses loop_factory to create the QEventLoop.
    • Python < 3.12: Sets a DefaultQEventLoopPolicy to ensure asyncio.run() uses the QEventLoop.
    import asyncio
    from qasync import run
    
    async def main():
        await asyncio.sleep(1)
        print("Done")
    
    run(main())
  7. Use QEventLoop to integrate asyncio with Qt

    master

    The QEventLoop class is an implementation of the asyncio event loop that uses the Qt event loop. It allows you to run asynchronous code within a Qt application.

    If you are using an existing, already running QApplication, you must instantiate QEventLoop with already_running=True. In this case, you are responsible for manual cleanup using stop() and close().

    On Windows, it uses QIOCPEventLoop (Proactor), and on Unix-like systems, it uses QSelectorEventLoop (Selector).

    import asyncio
    from qasync import QEventLoop
    
    # Example usage with asyncio.run
    async def main():
        await asyncio.sleep(1)
        print("Hello from qasync!")
    
    # Note: In Python 3.12+, run() is simplified. 
    # For older versions, you might need to provide a loop_factory.
    asyncio.run(main(), loop_factory=lambda: QEventLoop(app))
  8. Run async code before application close with asyncClose

    master

    The asyncClose decorator allows you to run asynchronous cleanup or logic immediately before a Qt application or component is closed. It wraps the decorated function, creates an asyncio.Task, and processes Qt events until the task is complete.

    from qasync import asyncClose
    
    @asyncClose()
    async def cleanup():
        await asyncio.sleep(1)
        print("Cleanup complete")
    
    # Call this when the app is closing
    cleanup()
  9. Run blocking Qt code asynchronously with asyncWrap

    master

    The asyncWrap function wraps a blocking function (like a modal Qt dialog) so it can be awaited within an async function without blocking the main asyncio event loop. It schedules the function to run using a one-shot QTimer in the next event loop iteration.

    from qasync import asyncWrap
    from PySide6.QtWidgets import QMessageBox
    
    async def show_dialog():
        # This prevents the modal dialog from freezing the asyncio loop
        result = await asyncWrap(
            lambda: QMessageBox.information(None, "Title", "Message")
        )
        print(f"Dialog result: {result}")
  10. Use QThreadExecutor for thread pooling in Qt

    master

    The QThreadExecutor provides a thread pool implementation using QThread objects, following the same API as concurrent.futures.Executor. This is useful for offloading blocking tasks to background threads while remaining compatible with the Qt environment.

    Use it as a context manager to ensure proper shutdown of worker threads.

    from qasync import QThreadExecutor
    
    with QThreadExecutor(max_workers=5) as executor:
        future = executor.submit(lambda x: x * x, 10)
        result = future.result()
        assert result == 100