API Reference

qtinter provides the following functions and classes in its public API:

Context managers for asyncio-Qt interop:

using_asyncio_from_qt() enables Qt-driven code to use asyncio-based components.
using_qt_from_asyncio() enables asyncio-driven code to use Qt-based components.

Helper functions to make interp code fit naturally into the current coding pattern:

asyncsignal() makes a Qt signal awaitable; useful for asyncio-driven code.
asyncsignalstream() exposes a Qt signal as an asynchronous iterator; useful for asyncio-driven code.
asyncslot() connects a coroutine function to a Qt signal; useful for Qt-driven code.
modal() allows the asyncio event loop to continue running in a nested Qt event loop.
multisignal() collects multiple Qt signals and re-emits them with a tag.
run_task() creates an asyncio.Task and eagerly executes its first step.

Loop factory to create event loop objects directly:

new_event_loop() creates an asyncio-compatible logical event loop object that runs on top of a physical Qt event loop.

Low-level classes that do the actual work of bridging Qt and asyncio:

Event loop interface
Event loop objects
Event loop policy objects

Private API that supports the internal implementation of qtinter.

Context managers

qtinter.using_asyncio_from_qt()

Context manager that enables enclosed Qt-driven code to use asyncio-based libraries.

Your code is Qt-driven if it calls app.exec() or equivalent as its entry point.

Example:

app = QtWidgets.QApplication([])
with qtinter.using_asyncio_from_qt():
    app.exec()

qtinter.using_qt_from_asyncio()

Context manager that enables enclosed asyncio-driven code to use Qt components.

Your code is asyncio-driven if it calls asyncio.run() or equivalent as its entry point.

Note

This context manager modifies the global (per-interpreter) asyncio event loop policy. Do not use this context manager if your code uses event loops from multiple threads. Instead, call new_event_loop() to create an event loop object and call its methods directly. Since Python 3.11, use asyncio.Runner and pass new_event_loop as its loop_factory parameter.

Helper functions

async qtinter.asyncsignal(signal: BoundSignal[Unpack[Ts]]) → Tuple[Unpack[Ts]]

Wait for signal to emit and return the emitted arguments in a tuple.

signal must be a bound Qt signal object, i.e. a bound PyQt5.QtCore.pyqtSignal, PyQt6.QtCore.pyqtSignal, PySide2.QtCore.Signal or PySide6.QtCore.Signal, or an object with a connect method of equivalent semantics, such as an instance of multisignal.

signal is connected to using an AutoConnection when the returned coroutine object is awaited. It is disconnected from after the signal is emitted once.

Note

Signals that require immediate response from the slot cannot be used with this function. An example is proxyAuthenticationRequired.

Note

This function will wait indefinitely if the signal is never emitted, e.g. if the sender object is deleted before emitting a signal. To handle the latter situation, keep a strong reference to the sender object, or listen to its destroyed signal.

qtinter.asyncsignalstream(signal: BoundSignal[Unpack[Ts]]) → AsyncIterator[Tuple[Unpack[Ts]]]

Return an asynchronous iterator that produces the emitted arguments from signal as a tuple.

signal is connected to via an AutoConnection before the function returns. It is disconnected from when the returned iterator object is deleted. Emitted arguments in the interim are stored in an internal buffer that grows without bound. It is advised to consume the iterator timely to avoid exhausting memory.

Example:

timer = QtCore.QTimer()
timer.setInterval(1000)
timer.start()

what = 'tick'
async for _ in qtinter.asyncsignalstream(timer.timeout):
    print(what)
    what = 'tock' if what == 'tick' else 'tick'

qtinter.asyncslot(fn: Callable[[Unpack[Ts]], Coroutine[T]], *, task_runner: Callable[[Coroutine[T]], asyncio.Task[T]] = qtinter.run_task) → Callable[[Unpack[Ts]], asyncio.Task[T]]

Return a callable object wrapping coroutine function fn so that it can be connected to a Qt signal.

When the returned wrapper is called, fn is called with the same arguments to produce a coroutine object. The coroutine object is then passed to task_runner to create an asyncio.Task object that handles its execution. The task object is returned by the wrapper.

The default task_runner, run_task, eagerly executes the task until the first yield, return or raise (whichever comes first) before returning the task object. The remainder of the coroutine is scheduled for later execution.

Note

asyncslot() keeps a strong reference to any task object it creates until the task completes.

Note

If fn is a (bound) method object, the returned wrapper will also be a method object whose lifetime is equal to that of fn, except that a strong reference to the returned wrapper keeps fn alive.

qtinter.modal(fn: Callable[[Unpack[Ts]], T]) → Callable[[Unpack[Ts]], Coroutine[T]]

Return a coroutine function that wraps a regular function fn. The coroutine function takes the same arguments as fn.

When the returned coroutine function is called and awaited, fn is scheduled to be called as interleaved code immediately after the caller is suspended. The result (exception) of fn is returned (raised) by the coroutine.

Note

This function is similar to asyncio.loop.run_in_executor() except that fn is executed in the same thread as interleaved code.

This function is designed to be called from a coroutine to schedule an fn that creates a nested Qt event loop. In this case, the logical asyncio event loop is allowed to continue running without nesting. For example:

await qtinter.modal(QtWidgets.QMessageBox.warning)(self, "Title", "Message")

class qtinter.multisignal(signal_map: Mapping[BoundSignal, Any])

Collect multiple bound signals and re-emit their arguments along with their mapped value.

A multisignal object defines the following instance method:

connect(slot: Callable[[Any, Tuple], Any]) → None

Connect slot to each signal in (the keys of) signal_map, such that if signal s is mapped to v in signal_map and is emitted with arguments *args, slot is called with v and args as arguments from the thread that called connect(). Its return value is ignored.

The sender objects of signals in signal_map must be alive when connect() is called, or the process will crash with SIGSEGV.

multisignal objects do not have a disconnect method. The connections are automatically disconnected when the sender (of a signal) or the receiver (of slot) is deleted.

multisignal may be used with asyncsignal() to listen to multiple signals.

Example:

fast_timer = QtCore.QTimer()
slow_timer = QtCore.QTimer()
# ...
ms = qtinter.multisignal({
    fast_timer.timeout: 'fast',
    slow_timer.timeout: 'slow',
})
ms.connect(print)

# Output:
# fast ()
# slow ()

qtinter.run_task(coro: Coroutine[T], *, allow_task_nesting: bool = True, name: Optional[str] = None, context: Optional[contextvars.Context] = None) → asyncio.Task[T]

Create an asyncio.Task wrapping the coroutine coro and execute it immediately until the first yield, return or raise, whichever comes first. The remainder of the coroutine is scheduled for later execution. Return the asyncio.Task object.

If allow_task_nesting is True (the default), this method is allowed to be called from a running task — the calling task is ‘suspended’ before executing the first step of coro and ‘resumed’ after that step completes. If allow_task_nesting is False, this method can only be called from a callback.

An asyncio event loop must be running when this function is called.

Since Python 3.8: Added the name parameter.

Since Python 3.11: Added the context parameter.

Loop factory

qtinter.new_event_loop() → asyncio.AbstractEventLoop

Return a new instance of an asyncio-compatible event loop object that runs on top of a Qt event loop.

Use this function instead of using_qt_from_asyncio() if your code uses different types of event loops from multiple threads. For example, starting from Python 3.11, if your code uses asyncio.Runner as its entry point, pass this function as the loop_factory parameter when constructing asyncio.Runner.

Low-level classes

You normally do not need to use these low-level API directly.

Event loop interface

All event loop objects below are derived from the abstract base class QiBaseEventLoop.

class qtinter.QiBaseEventLoop

Counterpart to the (undocumented) asyncio.BaseEventLoop class, implemented on top of a Qt event loop.

In addition to asyncio’s Event Loop Methods, this class defines the following methods for Qt interop:

exec_modal(fn: Callable[[], Any]) → None

Schedule fn to be called as interleaved code (i.e. not as a callback) immediately after the current callback completes. The return value of fn is ignored.

This method must be called from a coroutine or callback. There can be at most one pending fn at any time.

If the current callback raises KeyboardInterrupt or SystemExit, fn will be called the next time the loop is run.

set_mode(mode: QiLoopMode) → None:

Set loop operating mode to mode.

This method can only be called when the loop is not closed and not running, and no stop is pending.

A newly created loop object is in QiLoopMode.OWNER mode.

start() → None:

Start the loop (i.e. put it into running state) and return without waiting for it to stop.

This method can only be called in guest mode and when the loop is not already running.

class qtinter.QiLoopMode

An enum.Enum that defines the possible operating modes of a QiBaseEventLoop. Its members are:

OWNER: Appropriate for use with asyncio-driven code.

GUEST: Appropriate for use with Qt-driven code.

NATIVE: Appropriate for running clean-up code.

For details on the semantics of these modes, see Loop modes.

Event loop objects

class qtinter.QiDefaultEventLoop

In Python 3.7: alias to QiSelectorEventLoop.

Since Python 3.8: alias to QiSelectorEventLoop on Unix and QiProactorEventLoop on Windows.

class qtinter.QiProactorEventLoop(proactor=None)

Counterpart to asyncio.ProactorEventLoop, implemented on top of a Qt event loop.

Availability: Windows.

class qtinter.QiSelectorEventLoop(selector=None): Counterpart to asyncio.SelectorEventLoop, implemented on top of a Qt event loop.

Event loop policy objects

class qtinter.QiDefaultEventLoopPolicy

In Python 3.7: alias to QiSelectorEventLoopPolicy.

Since Python 3.8: alias to QiSelectorEventLoopPolicy on Unix and QiProactorEventLoopPolicy on Windows.

class qtinter.QiProactorEventLoopPolicy

Event loop policy that creates QiProactorEventLoop.

Availability: Windows.

class qtinter.QiSelectorEventLoopPolicy: Event loop policy that creates QiSelectorEventLoop.

Private API

The following classes and functions are used internally to support qtinter’s implementation. They are documented here solely for developing qtinter, and are subject to change at any time.

class qtinter.SemiWeakRef(o, ref=weakref.ref)

Return an object that is deleted when o is deleted, except that a strong reference to the returned object in user code keeps o alive.

ref should be a weak reference class suitable for o: If o is a method object, ref should be set to weakref.WeakMethod; otherwise, ref should be set to weakref.ref.

referent(): Return o if it is still live, or None.

qtinter.copy_signal_arguments(args: Tuple[Unpack[Ts]]) → Tuple[Unpack[Ts]]

Return a copy of signal arguments args where necessary.

In PyQt5/6, signal arguments passed to a slot may be temporary objects whose lifetime is only valid during the slot’s execution. In order to use the signal arguments after the slot returns, one must call this function to make a copy of them, or the program may crash with SIGSEGV when the signal arguments are accessed later.

PySide2/6 already passes a copy of the signal arguments to slots, whose lifetime is controlled by the usual Python mechanisms. This function returns args as is in this case.

qtinter.get_positional_parameter_count(fn: Callable) → int

Return the number of positional parameters of fn, or -1 if fn takes variadic positional parameters (*args).

Raises TypeError if fn takes any keyword-only parameter without a default.

qtinter.transform_slot(slot: Callable[[Unpack[Ts]], T], transform: Callable[[Callable[[Unpack[Ts]], T], Tuple[Unpack[Rs]], Unpack[Es]], R], *extras: Unpack[Es]) → Callable[[Unpack[Rs]], R]

Return a callable wrapper that takes variadic arguments *args, such that wrapper(*arg) returns transform(slot, args, *extra).

If slot is a bound method object, wrapper will also be a bound method object with the same lifetime as slot, except that a strong reference to wrapper keeps slot alive.

If slot is not a bound method object, wrapper will be a function object that holds a strong reference to slot.