tessl/pypi-qasync
Python library for using asyncio in Qt-based applications
—
Pending
event-loop.mddocs/
Event Loop Integration
Qt-compatible asyncio event loop that enables coroutines and asyncio functionality to run directly inside Qt application event loops. The QEventLoop class provides a complete implementation of PEP 3156 that integrates seamlessly with Qt's event processing.
Capabilities
Core Event Loop
Creates an asyncio-compatible event loop that integrates with Qt's QApplication event loop, allowing async/await syntax and asyncio functionality to work within Qt applications.
class QEventLoop:
"""
Implementation of asyncio event loop that uses the Qt Event loop.
Args:
app: QApplication instance (defaults to QApplication.instance())
set_running_loop: Deprecated parameter for backwards compatibility
already_running: If True, assumes the Qt application is already running
"""
def __init__(self, app=None, set_running_loop=False, already_running=False): ...Usage Example
import asyncio
import sys
from PySide6.QtWidgets import QApplication
from qasync import QEventLoop
async def main():
print("Hello from async coroutine!")
await asyncio.sleep(1)
print("Async operation complete!")
app = QApplication(sys.argv)
loop = QEventLoop(app)
# Method 1: Run until complete
loop.run_until_complete(main())
# Method 2: Use modern asyncio.run with loop factory
asyncio.run(main(), loop_factory=lambda: QEventLoop(app))Event Loop Control
Methods for controlling the event loop lifecycle, including starting, stopping, and managing the loop state.
def run_forever(self):
"""
Run eventloop forever.
Raises:
RuntimeError: If event loop is already running
"""
def run_until_complete(self, future):
"""
Run until Future is complete.
Args:
future: Coroutine or Future to run to completion
Returns:
Result of the future
Raises:
RuntimeError: If event loop is already running or stops before completion
"""
def stop(self):
"""Stop event loop."""
def close(self):
"""
Release all resources used by the event loop.
The loop cannot be restarted after it has been closed.
Raises:
RuntimeError: If called on a running event loop
"""
def is_running(self):
"""
Return True if the event loop is running, False otherwise.
Returns:
bool: Current running state
"""Callback Scheduling
Schedule callbacks to run at specific times or as soon as possible within the event loop.
def call_soon(self, callback, *args, context=None):
"""
Register a callback to be run on the next iteration of the event loop.
Args:
callback: Function to call
*args: Arguments to pass to callback
context: Optional contextvars.Context
Returns:
asyncio.Handle: Handle that can be used to cancel the callback
"""
def call_later(self, delay, callback, *args, context=None):
"""
Register callback to be invoked after a certain delay.
Args:
delay: Delay in seconds
callback: Function to call
*args: Arguments to pass to callback
context: Optional contextvars.Context
Returns:
asyncio.Handle: Handle that can be used to cancel the callback
Raises:
TypeError: If callback is a coroutine function or not callable
"""
def call_at(self, when, callback, *args, context=None):
"""
Register callback to be invoked at a certain time.
Args:
when: Absolute time when to run callback (from time.monotonic())
callback: Function to call
*args: Arguments to pass to callback
context: Optional contextvars.Context
Returns:
asyncio.Handle: Handle that can be used to cancel the callback
"""
def time(self):
"""
Get time according to event loop's clock.
Returns:
float: Current time from time.monotonic()
"""Thread-Safe Operations
Enable thread-safe interaction with the event loop from other threads.
def call_soon_threadsafe(self, callback, *args, context=None):
"""
Thread-safe version of call_soon.
Args:
callback: Function to call
*args: Arguments to pass to callback
context: Optional contextvars.Context
"""
def run_in_executor(self, executor, callback, *args):
"""
Run callback in executor.
Args:
executor: Executor instance (uses default QThreadExecutor if None)
callback: Function to run in executor
*args: Arguments to pass to callback
Returns:
asyncio.Future: Future representing the execution
"""
def set_default_executor(self, executor):
"""
Set the default executor for run_in_executor calls.
Args:
executor: Executor instance implementing concurrent.futures.Executor interface
"""Error Handling
Configure how the event loop handles exceptions that occur during callback execution.
def set_exception_handler(self, handler):
"""
Set custom exception handler for the event loop.
Args:
handler: Function that takes (loop, context) parameters
"""
def default_exception_handler(self, context):
"""
Handle exceptions using the default behavior.
Args:
context: Dictionary containing exception information
"""
def call_exception_handler(self, context):
"""
Call the configured exception handler.
Args:
context: Dictionary containing exception information
"""Debug Support
Methods for debugging and monitoring event loop behavior.
def get_debug(self):
"""
Get debug mode status.
Returns:
bool: True if debug mode is enabled
"""
def set_debug(self, enabled):
"""
Set debug mode for the event loop.
Args:
enabled: Whether to enable debug mode
"""Platform-Specific Implementations
Unix/Linux (QSelectorEventLoop)
Uses a selector-based approach with QSocketNotifier for file descriptor monitoring, compatible with asyncio.SelectorEventLoop.
Windows (QIOCPEventLoop)
Uses I/O Completion Ports (IOCP) through a proactor pattern, compatible with asyncio.ProactorEventLoop.
Context Manager Support
def __enter__(self):
"""Context manager entry."""
def __exit__(self, *args):
"""Context manager exit - stops and closes the loop."""Usage Example
async def main():
await asyncio.sleep(1)
print("Task complete!")
app = QApplication(sys.argv)
with QEventLoop(app) as loop:
loop.run_until_complete(main())
# Loop is automatically stopped and closedInstall with Tessl CLI
npx tessl i tessl/pypi-qasync