### Install qasync using pip Source: https://github.com/cabbagedevelopment/qasync/blob/master/README.md Use this command to install qasync using the pip package manager. ```bash pip install qasync ``` -------------------------------- ### Using qasync.run() for cross-version compatibility Source: https://context7.com/cabbagedevelopment/qasync/llms.txt qasync.run() is a convenience wrapper that automatically installs QEventLoop, supporting Python 3.8–3.13. It simplifies the setup for older Python versions by temporarily setting a DefaultQEventLoopPolicy. ```python import asyncio import sys import qasync from PySide6.QtWidgets import QApplication, QPushButton, QVBoxLayout, QWidget from qasync import QEventLoop class Window(QWidget): def __init__(self): super().__init__() layout = QVBoxLayout(self) btn = QPushButton("Click me", self) btn.clicked.connect(self._on_click) layout.addWidget(btn) def _on_click(self): print("Button clicked") async def main(): app = QApplication.instance() app_close = asyncio.Event() app.aboutToQuit.connect(app_close.set) win = Window() win.show() await app_close.wait() if __name__ == "__main__": app = QApplication(sys.argv) qasync.run(main()) # works on Python 3.8 – 3.13 ``` -------------------------------- ### Install qasync using uv Source: https://github.com/cabbagedevelopment/qasync/blob/master/README.md Use this command to add qasync to your project dependencies when using the uv package manager. ```bash uv add qasync ``` -------------------------------- ### Basic PyQt/PySide Application with qasync Source: https://github.com/cabbagedevelopment/qasync/blob/master/README.md Demonstrates how to integrate asyncio coroutines into a PyQt/PySide application using qasync. It shows how to use @asyncSlot for asynchronous button clicks and @asyncClose for asynchronous cleanup during application exit. Requires QApplication setup and a compatible asyncio event loop factory. ```python import asyncio import sys from PySide6.QtGui import QCloseEvent from PySide6.QtWidgets import QApplication, QPushButton, QVBoxLayout, QWidget import qasync from qasync import QEventLoop, asyncClose, asyncSlot class MainWindow(QWidget): def __init__(self): super().__init__() layout = QVBoxLayout() self.button = QPushButton("Load", self) self.button.clicked.connect(self.onButtonClicked) layout.addWidget(self.button) self.setLayout(layout) @asyncSlot() async def onButtonClicked(self): """ Use async code in a slot by decorating it with @asyncSlot. """ self.button.setText("Loading...") await asyncio.sleep(1) self.button.setText("Load") @asyncClose async def closeEvent(self, event: QCloseEvent): """ Use async code in a closeEvent by decorating it with @asyncClose. """ self.button.setText("Closing...") await asyncio.sleep(1) async def main(app): app_close_event = asyncio.Event() app.aboutToQuit.connect(app_close_event.set) main_window = MainWindow() main_window.show() await app_close_event.wait() 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)) ``` -------------------------------- ### Force Qt binding with QT_API environment variable Source: https://context7.com/cabbagedevelopment/qasync/llms.txt Set the QT_API environment variable (e.g., "pyside6", "pyqt6") to force qasync to use a specific Qt binding. This is useful when multiple bindings are installed. ```python import os import sys # Force PySide6 before importing qasync os.environ["QT_API"] = "pyside6" # or "pyqt6", "pyqt5", "pyside2" import asyncio from PySide6.QtWidgets import QApplication from qasync import QEventLoop, asyncSlot # From the shell: # QT_API=pyqt6 python my_app.py class App(QApplication): pass async def main(): app_quit = asyncio.Event() QApplication.instance().aboutToQuit.connect(app_quit.set) print(f"Using Qt binding: {os.environ['QT_API']}") await app_quit.wait() if __name__ == "__main__": app = App(sys.argv) asyncio.run(main(), loop_factory=lambda: QEventLoop(app)) ``` -------------------------------- ### Offload blocking I/O with run_in_executor Source: https://context7.com/cabbagedevelopment/qasync/llms.txt Use loop.run_in_executor with None to default to QThreadExecutor, preventing blocking I/O from freezing the GUI. Results are safely returned to the main thread after await. ```python import asyncio import sys import time from PySide6.QtWidgets import QApplication, QLabel, QVBoxLayout, QWidget from qasync import QEventLoop def blocking_io() -> str: time.sleep(1) # simulate slow I/O return "Data from thread" class Window(QWidget): def __init__(self): super().__init__() layout = QVBoxLayout(self) self.lbl = QLabel("Working…", self) layout.addWidget(self.lbl) async def main(app): win = Window() win.show() loop = asyncio.get_running_loop() # Default executor → QThreadExecutor; does not block the GUI result = await loop.run_in_executor(None, blocking_io) win.lbl.setText(result) # safe: back on the main thread after await await asyncio.sleep(2) app.quit() if __name__ == "__main__": app = QApplication(sys.argv) asyncio.run(main(app), loop_factory=lambda: QEventLoop(app)) ``` -------------------------------- ### Embed QEventLoop in existing QApplication Source: https://context7.com/cabbagedevelopment/qasync/llms.txt When a QApplication is already running, use `already_running=True` to create a QEventLoop that attaches to the existing app. The caller must manage `loop.stop()` and `loop.close()`. ```python import asyncio from PySide6.QtWidgets import QApplication from qasync import QEventLoop # Assume QApplication already exists and is running (e.g., host application) app = QApplication.instance() loop = QEventLoop(app, already_running=True) asyncio.set_event_loop(loop) async def background_task(): for i in range(5): print(f"Background tick {i}") await asyncio.sleep(1) # Schedule async work on the already-running loop asyncio.ensure_future(background_task()) # When the host application quits, clean up manually: # loop.stop() # loop.close() ``` -------------------------------- ### Use asyncWrap to Run Blocking Qt Calls from Async Source: https://context7.com/cabbagedevelopment/qasync/llms.txt The `asyncWrap` function schedules a synchronous, potentially blocking callable on the Qt event loop using a single-shot QTimer. It returns an awaitable for the result, safely handling modal dialogs within async contexts. ```python import asyncio import sys from PySide6.QtWidgets import QApplication, QMessageBox, QProgressBar from qasync import QEventLoop, asyncWrap async def run_with_confirmation(): bar = QProgressBar() bar.setRange(0, 9) bar.show() for i in range(10): bar.setValue(i) await asyncio.sleep(0.2) # Show a modal dialog safely from within an async coroutine result = await asyncWrap( lambda: QMessageBox.question( None, "Continue?", "Task finished. Quit now?", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, ) ) if result == QMessageBox.StandardButton.Yes: QApplication.instance().quit() if __name__ == "__main__": app = QApplication(sys.argv) asyncio.run(run_with_confirmation(), loop_factory=lambda: QEventLoop(app)) ``` -------------------------------- ### QT_API Environment Variable Source: https://context7.com/cabbagedevelopment/qasync/llms.txt The `QT_API` environment variable allows users to explicitly select the Qt binding (PyQt6, PyQt5, PySide6, PySide2) that qasync should use. ```APIDOC ## `QT_API` environment variable — Qt binding selection qasync auto-detects the Qt binding to use by inspecting `sys.modules` first, then attempting imports in the order `PyQt6 → PyQt5 → PySide6 → PySide2`. Set the `QT_API` environment variable (case-insensitive) to force a specific binding, which is useful in environments where multiple bindings are installed. ```python import os import sys # Force PySide6 before importing qasync os.environ["QT_API"] = "pyside6" # or "pyqt6", "pyqt5", "pyside2" import asyncio from PySide6.QtWidgets import QApplication from qasync import QEventLoop, asyncSlot # From the shell: # QT_API=pyqt6 python my_app.py class App(QApplication): pass async def main(): app_quit = asyncio.Event() QApplication.instance().aboutToQuit.connect(app_quit.set) print(f"Using Qt binding: {os.environ['QT_API']}") await app_quit.wait() if __name__ == "__main__": app = App(sys.argv) asyncio.run(main(), loop_factory=lambda: QEventLoop(app)) ``` ``` -------------------------------- ### Using QEventLoop with asyncio.run() Source: https://context7.com/cabbagedevelopment/qasync/llms.txt Integrates QEventLoop as the event loop factory for asyncio.run() on Python 3.11+. This allows asyncio tasks and Qt signals/slots to share a single thread, keeping the GUI responsive during asynchronous operations. ```python import asyncio import sys from PySide6.QtWidgets import QApplication, QLabel, QVBoxLayout, QWidget from qasync import QEventLoop class MainWindow(QWidget): def __init__(self): super().__init__() layout = QVBoxLayout(self) self.label = QLabel("Starting…", self) layout.addWidget(self.label) async def main(app: QApplication) -> None: window = MainWindow() window.show() # Simulate async work while the GUI stays responsive for i in range(5): await asyncio.sleep(1) window.label.setText(f"Tick {i + 1}") app.quit() if __name__ == "__main__": app = QApplication(sys.argv) # Python ≥ 3.11: pass QEventLoop as the loop factory asyncio.run(main(app), loop_factory=lambda: QEventLoop(app)) # Python ≤ 3.10: use qasync.run(main(app)) ``` -------------------------------- ### asyncWrap function Source: https://context7.com/cabbagedevelopment/qasync/llms.txt The `asyncWrap` function allows running blocking Qt calls from an asynchronous context. It schedules a synchronous callable on the native Qt event loop using a single-shot `QTimer` and returns its result as an awaitable. This is particularly useful for safely showing modal dialogs within an `asyncSlot` without unsafe event loop re-entry. ```APIDOC ## asyncWrap(fn, *args, **kwargs) ### Description Schedules a synchronous (potentially blocking) callable on the native Qt event loop via a single-shot `QTimer` and returns its result as an awaitable. This is essential for showing modal dialogs inside an `asyncSlot` — calling a blocking modal directly would re-enter the Qt event loop in an unsafe way; `asyncWrap` avoids that by deferring execution to the next Qt event loop iteration. ### Usage ```python from qasync import asyncWrap # Inside an async function or asyncSlot result = await asyncWrap(QMessageBox.question, None, "Title", "Message") ``` ``` -------------------------------- ### QEventLoop.run_in_executor Source: https://context7.com/cabbagedevelopment/qasync/llms.txt `run_in_executor` is a standard asyncio method augmented by qasync to default to a `QThreadExecutor` when no executor is provided. It allows offloading blocking operations to threads without blocking the GUI. ```APIDOC ## `QEventLoop.run_in_executor(executor, callback, *args)` — offload work to threads `run_in_executor` is a standard asyncio method augmented by qasync to default to a `QThreadExecutor` when no executor is provided. Pass `None` as the executor to use the default Qt thread pool, or supply a custom `QThreadExecutor` (or any `concurrent.futures.Executor`). Use `call_soon_threadsafe` to safely post callbacks back to the main thread from inside worker threads. ```python import asyncio import sys import time from PySide6.QtWidgets import QApplication, QLabel, QVBoxLayout, QWidget from qasync import QEventLoop def blocking_io() -> str: time.sleep(1) # simulate slow I/O return "Data from thread" class Window(QWidget): def __init__(self): super().__init__() layout = QVBoxLayout(self) self.lbl = QLabel("Working…", self) layout.addWidget(self.lbl) async def main(app): win = Window() win.show() loop = asyncio.get_running_loop() # Default executor → QThreadExecutor; does not block the GUI result = await loop.run_in_executor(None, blocking_io) win.lbl.setText(result) # safe: back on the main thread after await await asyncio.sleep(2) app.quit() if __name__ == "__main__": app = QApplication(sys.argv) asyncio.run(main(app), loop_factory=lambda: QEventLoop(app)) ``` ``` -------------------------------- ### Use asyncSlot to Decorate Async Qt Slots Source: https://context7.com/cabbagedevelopment/qasync/llms.txt Use `@asyncSlot()` to wrap an `async def` method, allowing it to be connected to Qt signals. This schedules the coroutine as an asyncio Task on the QEventLoop. Unhandled exceptions are forwarded to sys.excepthook. ```python import asyncio import sys from PySide6.QtWidgets import QApplication, QPushButton, QLabel, QVBoxLayout, QWidget from qasync import QEventLoop, asyncSlot class FetchWindow(QWidget): def __init__(self): super().__init__() layout = QVBoxLayout(self) self.status = QLabel("Idle", self) self.btn = QPushButton("Fetch Data", self) self.btn.clicked.connect(self.on_fetch) # connect async slot to signal layout.addWidget(self.status) layout.addWidget(self.btn) @asyncSlot() # no signal argument types → use () async def on_fetch(self): self.btn.setEnabled(False) self.status.setText("Fetching…") try: await asyncio.sleep(2) # replace with real async I/O self.status.setText("Done!") except Exception as exc: self.status.setText(f"Error: {exc}") finally: self.btn.setEnabled(True) async def main(app): app_quit = asyncio.Event() app.aboutToQuit.connect(app_quit.set) win = FetchWindow() win.show() await app_quit.wait() if __name__ == "__main__": app = QApplication(sys.argv) asyncio.run(main(app), loop_factory=lambda: QEventLoop(app)) ``` -------------------------------- ### Use asyncClose for Async Qt Close Events Source: https://context7.com/cabbagedevelopment/qasync/llms.txt The `@asyncClose` decorator allows an async `closeEvent` to be awaited before the window is destroyed. It runs the coroutine synchronously by processing Qt events until completion, ensuring graceful async teardown. ```python import asyncio import sys import aiohttp from PySide6.QtWidgets import QApplication, QPushButton, QVBoxLayout, QWidget from qasync import QEventLoop, asyncClose, asyncSlot class NetworkWindow(QWidget): def __init__(self): super().__init__() self.session = aiohttp.ClientSession() layout = QVBoxLayout(self) btn = QPushButton("Fetch", self) btn.clicked.connect(self.on_fetch) layout.addWidget(btn) @asyncClose # runs async teardown before close async def closeEvent(self, event): print("Closing HTTP session…") await self.session.close() # properly awaited before window closes print("Session closed.") @asyncSlot() async def on_fetch(self): async with self.session.get("https://jsonplaceholder.typicode.com/todos/1") as r: data = await r.json() print(data) async def main(app): quit_event = asyncio.Event() app.aboutToQuit.connect(quit_event.set) win = NetworkWindow() win.show() await quit_event.wait() if __name__ == "__main__": app = QApplication(sys.argv) asyncio.run(main(app), loop_factory=lambda: QEventLoop(app)) ``` -------------------------------- ### QThreadExecutor Source: https://context7.com/cabbagedevelopment/qasync/llms.txt `QThreadExecutor` provides a Qt thread pool for CPU-bound tasks, integrating with `loop.run_in_executor()` to keep intensive work off the asyncio thread. ```APIDOC ## `QThreadExecutor(max_workers, stack_size)` — Qt thread pool for CPU-bound tasks `QThreadExecutor` implements the `concurrent.futures.Executor` interface using `QThread` workers instead of Python's `ThreadPoolExecutor`. It integrates seamlessly with `loop.run_in_executor()`, keeping CPU-intensive work off the asyncio thread while still allowing results to flow back into async coroutines. It supports the context-manager protocol for automatic shutdown. ```python import asyncio import functools import sys import time from PySide6.QtWidgets import QApplication, QProgressBar from qasync import QEventLoop, QThreadExecutor def cpu_work(n: int) -> int: """Simulate CPU-bound computation.""" time.sleep(0.05) # blocking — must run in a thread return n * n async def main(): bar = QProgressBar() bar.setRange(0, 99) bar.show() loop = asyncio.get_running_loop() # Use QThreadExecutor with 4 worker QThreads with QThreadExecutor(4) as executor: futures = [ loop.run_in_executor(executor, functools.partial(cpu_work, i)) for i in range(100) ] for idx, fut in enumerate(asyncio.as_completed(futures)): result = await fut bar.setValue(idx) print(f"Result {idx}: {result}") QApplication.instance().quit() if __name__ == "__main__": app = QApplication(sys.argv) asyncio.run(main(), loop_factory=lambda: QEventLoop(app)) ``` ``` -------------------------------- ### asyncSlot decorator Source: https://context7.com/cabbagedevelopment/qasync/llms.txt The `asyncSlot` decorator allows asynchronous methods (coroutines) to be connected directly to Qt signals. It schedules the coroutine as an asyncio Task on the running QEventLoop and handles unhandled exceptions by forwarding them to sys.excepthook. Extra signal arguments are silently dropped if they don't match the slot signature. ```APIDOC ## asyncSlot(*types) ### Description Wraps an `async def` method so it can be connected directly to a Qt signal. It decorates the function with the appropriate `@Slot` (or `@pyqtSlot`) decorator and schedules the coroutine as an asyncio `Task` on the running `QEventLoop`. Unhandled exceptions inside the coroutine are forwarded to `sys.excepthook`. Extra trailing arguments emitted by a signal that do not match the slot signature are silently dropped. ### Usage ```python from qasync import asyncSlot class MyWidget: @asyncSlot() async def my_async_method(self): # Coroutine logic here pass ``` ``` -------------------------------- ### asyncClose decorator Source: https://context7.com/cabbagedevelopment/qasync/llms.txt The `asyncClose` decorator enables asynchronous cleanup operations before a Qt window is destroyed. It wraps a `closeEvent` or any synchronous override, allowing async code (like closing network sessions) to be awaited synchronously by processing Qt events until the task completes, thus avoiding event loop blocking. ```APIDOC ## asyncClose(fn) ### Description Wraps a `closeEvent` (or any synchronous override) so that async cleanup work can be awaited before the window is actually destroyed. It runs the coroutine synchronously by processing Qt events until the task completes, which avoids blocking the event loop while still letting async teardown (e.g., closing network sessions) finish gracefully. ### Usage ```python from qasync import asyncClose class MyWindow: @asyncClose async def closeEvent(self, event): await self.cleanup_async_resources() # ... ``` ``` -------------------------------- ### QEventLoop Embedding Source: https://context7.com/cabbagedevelopment/qasync/llms.txt Allows embedding a `QEventLoop` into an already running `QApplication`, useful for plugins or interactive environments like Jupyter. ```APIDOC ## `QEventLoop(already_running=True)` — embedding into an existing QApplication When a `QApplication` is already running (e.g., inside a plugin or Jupyter QtConsole), pass `already_running=True` to create a `QEventLoop` that attaches to the running app without taking ownership of the event loop lifecycle. In this mode the caller is responsible for calling `loop.stop()` and `loop.close()` during teardown. ```python import asyncio from PySide6.QtWidgets import QApplication from qasync import QEventLoop # Assume QApplication already exists and is running (e.g., host application) app = QApplication.instance() loop = QEventLoop(app, already_running=True) asyncio.set_event_loop(loop) async def background_task(): for i in range(5): print(f"Background tick {i}") await asyncio.sleep(1) # Schedule async work on the already-running loop asyncio.ensure_future(background_task()) # When the host application quits, clean up manually: # loop.stop() # loop.close() ``` ``` -------------------------------- ### QThreadExecutor for CPU-bound tasks Source: https://context7.com/cabbagedevelopment/qasync/llms.txt Use QThreadExecutor to run CPU-intensive work off the asyncio thread. It integrates with loop.run_in_executor() and supports the context-manager protocol for automatic shutdown. ```python import asyncio import functools import sys import time from PySide6.QtWidgets import QApplication, QProgressBar from qasync import QEventLoop, QThreadExecutor def cpu_work(n: int) -> int: """Simulate CPU-bound computation.""" time.sleep(0.05) # blocking — must run in a thread return n * n async def main(): bar = QProgressBar() bar.setRange(0, 99) bar.show() loop = asyncio.get_running_loop() # Use QThreadExecutor with 4 worker QThreads with QThreadExecutor(4) as executor: futures = [ loop.run_in_executor(executor, functools.partial(cpu_work, i)) for i in range(100) ] for idx, fut in enumerate(asyncio.as_completed(futures)): result = await fut bar.setValue(idx) print(f"Result {idx}: {result}") QApplication.instance().quit() if __name__ == "__main__": app = QApplication(sys.argv) asyncio.run(main(), loop_factory=lambda: QEventLoop(app)) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.