QTrio - a library bringing Qt GUIs together with async and await via Trio

Introduction

Note:

This library is in early development. It works. It has tests. It has documentation. Expect breaking changes as we explore a clean API. By paying this price you get the privilege to provide feedback via GitHub issues to help shape our future. :]

The QTrio project’s goal is to bring the friendly concurrency of Trio using Python’s async and await syntax together with the GUI features of Qt to enable more correct code and a more pleasant developer experience. QTrio is permissively licensed to avoid introducing restrictions beyond those of the underlying Python Qt library you choose. Both PySide2 and PyQt5 are supported.

By enabling use of async and await it is possible in some cases to write related code more concisely and clearly than you would get with the signal and slot mechanisms of Qt concurrency. See the README examples in the docs for the full code. The first example here shows classic pure Qt code.

class Main:
    def __init__(
        self,
        input_dialog: typing.Optional[QtWidgets.QInputDialog] = None,
        output_dialog: typing.Optional[QtWidgets.QMessageBox] = None,
    ):
        if input_dialog is None:  # pragma: nocover
            input_dialog = create_input()

        if output_dialog is None:  # pragma: nocover
            output_dialog = create_output()

        self.input_dialog = input_dialog
        self.output_dialog = output_dialog

    def setup(self) -> None:
        self.input_dialog.accepted.connect(self.input_accepted)
        self.input_dialog.rejected.connect(self.input_rejected)

        self.input_dialog.show()

    def input_accepted(self) -> None:
        name = self.input_dialog.textValue()

        self.output_dialog.setText(f"Hi {name}, welcome to the team!")

        self.output_dialog.finished.connect(self.output_finished)
        self.output_dialog.show()

    def input_rejected(self) -> None:
        QtCore.QCoreApplication.instance().quit()

    def output_finished(self) -> None:
        QtCore.QCoreApplication.instance().quit()

The second example, below, shows how using async and await allows for a more concise and clear description of the sequenced activity.

async def main(
    input_dialog: typing.Optional[qtrio.dialogs.TextInputDialog] = None,
    output_dialog: typing.Optional[qtrio.dialogs.MessageBox] = None,
) -> None:
    if input_dialog is None:  # pragma: nocover
        input_dialog = create_input()

    if output_dialog is None:  # pragma: nocover
        output_dialog = create_output()

    with contextlib.suppress(qtrio.UserCancelledError):
        name = await input_dialog.wait()

        output_dialog.text = f"Hi {name}, welcome to the team!"

        await output_dialog.wait()

Getting Started

This tutorial introduces usage of QTrio to enable integration of Qt into a Trio async application. For help with relevant async concepts and usage of Trio itself see the Trio tutorial.

I know, I know… we are supposed to do one thing well. But QTrio presently targets three distinct development tools. In time perhaps pieces will be spun off but for now they provide increasing layers you can use or not as they interest you.

Installation

While the general aspects of installation using pip belong elsewhere, it is recommended to work in a virtual environment such as you can create with the venv module (see also Python Virtual Environments in Five Minutes ).

Somewhat more specific to QTrio, several extras are available for installing optional dependencies or applying version constraints.

  • cli - For CLI usage, presently just examples.

  • examples - For running examples.

  • pyqt5 - For running with PyQt5, primarily to apply any version constraints.

  • pyside2 - For running with PySide2, primarily to apply any version constraints.

A normal installation might look like:

$ myenv/bin/pip install qtrio[pyside2]

Overview

The first layer allows you to run Trio tasks in the same thread as the Qt event loop. This is valuable as it let’s the tasks safely interact directly with the Qt GUI objects. It is a small wrapper around Trio’s guest mode. This layer is exposed directly under the qtrio package.

Now that Qt and Trio are friends we can focus on making the relationship smoother. This second layer of QTrio is also available directly in the qtrio package and allows for awaiting signals and iterating over the emissions of signals. This avoids the normal callback design of GUI systems in favor of Trio’s structured concurrency allowing GUI responses to be handled where you want within the task tree.

Not everything Qt provides will be easily integrated into this structure. The rest of QTrio will grow to contain helpers and wrappers to address these cases.

In addition to the above three layers there is also adjacent support for testing.

Layer 1 - Crossing Paths

With one extra character you can turn trio.run() into qtrio.run(). This gets you the Trio guest mode hosted by a Qt QApplication. Note how there is only one function and you are able to asynchronously sleep in it to avoid blocking the GUI. By default, when you leave your main function the Qt application will be exited.

import typing

import qtrio
from qtpy import QtWidgets
import trio


async def main(
    label: typing.Optional[QtWidgets.QWidget] = None,
    message: str = "Hello world.",
    change_delay: float = 0.5,
    close_delay: float = 3,
) -> None:
    if label is None:  # pragma: no cover
        label = QtWidgets.QLabel()
    # start big enough to fit the whole message
    label.setText(message)
    label.show()
    label.setText("")

    for i in range(len(message)):
        await trio.sleep(change_delay)
        label.setText(message[: i + 1])

    await trio.sleep(close_delay)


if __name__ == "__main__":  # pragma: no cover
    qtrio.run(main)

Layer 2 - Building Respect

A good relationship goes both ways. Above, Trio did all the talking and Qt just listened. Now let’s have Trio listen to Qt. Emissions from Qt signals can be made available in a trio.MemoryReceiveChannel. You can either trio.MemoryReceiveChannel.receive() them one at a time or asynchronously iterate over them for longer lasting activities. The received object is a qtrio.Emission and contains both the originating signal and the arguments.

import typing

import attr
import qtrio
from qtpy import QtWidgets


@attr.s(auto_attribs=True)
class Widget:
    widget: QtWidgets.QWidget = attr.ib(factory=QtWidgets.QWidget)
    layout: QtWidgets.QLayout = attr.ib(factory=QtWidgets.QVBoxLayout)
    button: QtWidgets.QPushButton = attr.ib(factory=QtWidgets.QPushButton)
    label: QtWidgets.QWidget = attr.ib(factory=QtWidgets.QLabel)

    def setup(self, message: str) -> None:
        self.button.setText("More")

        # start big enough to fit the whole message
        self.label.setText(message)

        self.layout.addWidget(self.button)
        self.layout.addWidget(self.label)
        self.widget.setLayout(self.layout)

    def show(self) -> None:
        self.widget.show()
        self.label.setText("")


async def main(
    widget: typing.Optional[Widget] = None,
    message: str = "Hello world.",
) -> None:
    if widget is None:  # pragma: no cover
        widget = Widget()

    widget.setup(message=message)

    async with qtrio.enter_emissions_channel(
        signals=[widget.button.clicked]
    ) as emissions:
        i = 1
        widget.show()

        async for _ in emissions.channel:  # pragma: no branch
            widget.label.setText(message[:i])
            i += 1

            if i > len(message):
                break

        # wait for another click to finish
        await emissions.channel.receive()


if __name__ == "__main__":  # pragma: no cover
    qtrio.run(main)

Layer 3 - Best Friends

This space intentionally left blank.

(for now… sorry)

Core

Running

qtrio.run(async_fn, *args, done_callback=None, clock=None, instruments=())[source]

Run a Trio-flavored async function in guest mode on a Qt host application, and return the result.

Parameters
Return type

object

Returns

The object returned by async_fn.

class qtrio.Runner(application=NOTHING, quit_application=True, clock=None, instruments=(), reenter=NOTHING, done_callback=None)[source]

Bases: object

This class helps run Trio in guest mode on a Qt host application.

application

The Qt application object to run as the host. If not set before calling run() the application will be created as QtWidgets.QApplication(sys.argv[1:]) and .setQuitOnLastWindowClosed(False) will be called on it to allow the application to continue throughout the lifetime of the async function passed to qtrio.Runner.run().

quit_application

When true, the done_callback() method will quit the application when the async function passed to qtrio.Runner.run() has completed.

clock

The clock to use for this run. This is primarily used to speed up tests that include timeouts. The value will be passed on to trio.lowlevel.start_guest_run().

instruments

The instruments to use for this run. The value will be passed on to trio.lowlevel.start_guest_run().

reenter

The QObject instance which will receive the events requesting execution of the needed Trio and user code in the host’s event loop and thread.

done_callback

The builtin done_callback() will be passed to trio.lowlevel.start_guest_run() but will call the callback passed here before (maybe) quitting the application. The outcome.Outcome from the completion of the async function passed to run() will be passed to this callback.

outcomes

The outcomes from the Qt and Trio runs.

cancel_scope

An all encompassing cancellation scope for the Trio execution.

run(async_fn, *args, execute_application=True)[source]

Start the guest loop executing async_fn.

Parameters

  • async_fn (Callable[…, Awaitable[object]]) – The async function to be run in the Qt host loop by the Trio guest.

  • args (object) – Arguments to pass when calling async_fn.

  • execute_application (bool) – If True, the Qt application will be executed and this call will block until it finishes.

Return type

Outcomes

Returns

If execute_application is true, a qtrio.Outcomes containing outcomes from the Qt application and async_fn will be returned. Otherwise, an empty qtrio.Outcomes.

run_sync_soon_threadsafe(fn)[source]

Helper for the Trio guest to execute a sync function in the Qt host thread when called from the Trio guest thread. This call will not block waiting for completion of fn nor will it return the result of calling fn.

Parameters

fn (Callable[[], object]) – A no parameter callable.

Return type

None

await trio_main(async_fn, args)[source]

Will be run as the main async function by the Trio guest. It creates a cancellation scope to be cancelled when lastWindowClosed is emitted. Within this scope the application’s async_fn will be run and passed args.

Parameters

  • async_fn (Callable[…, Awaitable[object]]) – The application’s main async function to be run by Trio in the Qt host’s thread.

  • args (Tuple[object, …]) – Positional arguments to be passed to async_fn

Return type

object

Returns

The result returned by async_fn.

trio_done(run_outcome)[source]

Will be called after the Trio guest run has finished. This allows collection of the outcome.Outcome and execution of any application provided done callback. Finally, if qtrio.Runner.quit_application was set when creating the instance then the Qt application will be requested to quit.

Actions such as outputting error information or unwrapping the outcomes need to be further considered.

Parameters

run_outcome (Outcome) – The outcome of the Trio guest run.

Return type

None

class qtrio.Outcomes(qt=None, trio=None)[source]

Bases: object

This class holds an outcome.Outcome from each of the Trio and the Qt application execution. Do not construct instances directly. Instead, an instance will be returned from qtrio.run() or available on instances of qtrio.Runner.outcomes.

qt

The Qt application outcome.Outcome

trio

The Trio async function outcome.Outcome

unwrap()[source]

Unwrap either the Trio or Qt outcome. First, errors are given priority over success values. Second, the Trio outcome gets priority over the Qt outcome.

Return type

object

Returns

Whatever captured value was selected.

Raises
class qtrio.Reenter[source]

Bases: QObject

A QtCore.QObject for handling reenter events.

event(event)[source]

Qt calls this when the object receives an event.

Return type

bool

staticMetaObject = <PySide2.QtCore.QMetaObject object>

Emissions

The basics of handling Qt GUIs is to respond to the widgets’ signals being emitted. qtrio.enter_emissions_channel() is the primary tool for handling this. It allows for connection of signals prior to showing a window and subsequent iteration of the emissions. See the emissions example for an example usage.

async with qtrio.enter_emissions_channel(signals, max_buffer_size=inf)[source]

Create a memory channel fed by the emissions of the signals and enter both the send and receive channels’ context managers.

Parameters

  • signals (Collection[Signal]) – A collection of signals which will be monitored for emissions.

  • max_buffer_size (Union[int, float]) – When the number of unhandled emissions in the channel reaches this limit then additional emissions will be silently thrown out the window.

Return type

AsyncGenerator[Emissions, None]

Returns

The emissions manager.

class qtrio.Emission(signal, args)[source]

Bases: object

Stores the emission of a signal including the emitted arguments. Can be compared against a signal instance to check the source. Do not construct this class directly. Instead, instances will be received through a channel created by qtrio.enter_emissions_channel().

Note

Each time you access a signal such as a_qobject.some_signal you get a different signal instance object so the signal attribute generally will not be the same object. A signal instance is a QtCore.SignalInstance in PySide2 or QtCore.pyqtBoundSignal in PyQt5.

signal

An instance of the original signal.

args

A tuple of the arguments emitted by the signal.

is_from(signal)[source]

Check if this emission came from signal.

Parameters

signal (Signal) – The signal instance to check for being the source.

Return type

bool

Returns

Whether the passed signal was the source of this emission.

class qtrio.Emissions(channel, send_channel)[source]

Bases: object

Hold elements useful for the application to work with emissions from signals. Do not construct this class directly. Instead, use qtrio.enter_emissions_channel().

channel: trio.MemoryReceiveChannel

A memory receive channel to be fed by signal emissions.

send_channel: trio.MemorySendChannel

A memory send channel collecting signal emissions.

await aclose()[source]

Asynchronously close the send channel when signal emissions are no longer of interest.

Return type

None

If you need a more Qt-like callback mechanism qtrio.open_emissions_nursery() offers that. Instead of tossing the callbacks behind the couch where they can leave their errors on the floor they will be run inside a nursery.

async with qtrio.open_emissions_nursery(until=None, wrapper=None)[source]

Open a nursery for handling callbacks triggered by signal emissions. This allows a ‘normal’ Qt callback structure while still executing the callbacks within a Trio nursery such that errors have a place to go. Both async and sync callbacks can be connected. Sync callbacks will be wrapped in an async call to allow execution in the nursery.

Parameters
Return type

AsyncGenerator[EmissionsNursery, None]

Returns

The emissions manager.

class qtrio.EmissionsNursery(nursery, exit_stack, wrapper=None)[source]

Bases: object

Holds the nursery, exit stack, and wrapper needed to support connecting signals to both async and sync slots in the nursery.

nursery: trio.Nursery

The Trio nursery that will handle execution of the slots.

exit_stack: contextlib.ExitStack

The exit stack that will manage the connections so they get disconnected.

wrapper: Optional[Callable[[Callable[[…], Awaitable[object]]], Awaitable[object]]]

The wrapper for handling the slots. This could, for example, handle exceptions and present a dialog to avoid cancelling the entire nursery.

connect(signal, slot)[source]

Connect to an async slot to this emissions nursery so when called the slot will be run in the nursery.

Return type

None

connect_sync(signal, slot)[source]

Connect to a sync slot to this emissions nursery so when called the slot will be run in the nursery.

Return type

None

Helpers

class qtrio.Signal(*args, **kwargs)[source]

Bases: object

This is a (nearly) drop-in replacement for Signal. The useful difference is that it does not require inheriting from QObject. The not-quite part is that it will be a bit more complicated to change thread affinity of the relevant QObject. If you need this, maybe just inherit.

This signal gets around the normally required inheritance by creating QObject instances behind the scenes to host the real signals. Just as Signal uses the Python descriptor protocol to intercept the attribute access, so does this so it can ‘redirect’ to the signal on the other object.

object(instance)[source]

Get the QObject that hosts the real signal. This can be called such as type(instance).signal_name.object(instance). Yes this is non-obvious but you have to do something special to get around the descriptor protocol so you can get at this method instead of just having the underlying Signal.

Parameters

instance (object) – The object on which this descriptor instance is hosted.

Return type

QObject

Returns

The signal-hosting QObject.

Reentry Events

Generally you should not need to use these functions. If you want to have control over when the Qt event type is registered, what value it gets, or handle any exceptions raised then you may like to call these directly.

qtrio.register_event_type()[source]

Register a Qt event type for use by Trio to reenter into the Qt event loop.

Raises
Return type

None

qtrio.register_requested_event_type(requested_value)[source]

Register the requested Qt event type for use by Trio to reenter into the Qt event loop.

Parameters

requested_value (Union[int, Type]) – The value to ask Qt to use for the event type being registered.

Raises
Return type

None

qtrio.registered_event_type()[source]

Get the registered event type.

Return type

Optional[Type]

Returns

The type registered with Qt for the reenter event. None if no event type has been registered yet.

Testing

pytest-trio provides for using an alternative runner specified via the trio_run configuration option. QTrio is enabled with a value of qtrio. The example pytest.ini below will be common for QTrio test suites.

# pytest.ini
[pytest]
trio_mode = true
trio_run = qtrio

Dialogs

Usage Pattern

Creation Functions

qtrio.dialogs.create_integer_dialog(parent=None)[source]

Create an integer input dialog.

Parameters

parent (Optional[QWidget]) – See qtrio.dialogs.IntegerDialog.parent.

Return type

IntegerDialog

Returns

The dialog manager.

qtrio.dialogs.create_text_input_dialog(title=None, label=None, parent=None)[source]

Create a text input dialog.

Parameters
Return type

TextInputDialog

Returns

The dialog manager.

qtrio.dialogs.create_file_open_dialog(parent=None, default_directory=None, default_file=None, options=PySide2.QtWidgets.QFileDialog.Option(0))[source]

Create a file open dialog.

Parameters
Return type

FileDialog

qtrio.dialogs.create_file_save_dialog(parent=None, default_directory=None, default_file=None, options=PySide2.QtWidgets.QFileDialog.Option(0))[source]

Create a file save dialog.

Parameters
Return type

FileDialog

qtrio.dialogs.create_message_box(title='', text='', icon=PySide2.QtWidgets.QMessageBox.Icon.Information, buttons=PySide2.QtWidgets.QMessageBox.StandardButton.Ok, parent=None)[source]

Create a message box.

Parameters
Return type

MessageBox

qtrio.dialogs.create_progress_dialog(title='', text='', cancel_button_text=None, minimum=0, maximum=0, parent=None)[source]

Create a progress dialog.

Parameters
Return type

ProgressDialog

Classes

class qtrio.dialogs.IntegerDialog(parent=None, dialog=None, edit_widget=None, accept_button=None, reject_button=None, result=None)[source]

Bases: object

Manage a dialog for inputting an integer from the user. Generally instances should be built via qtrio.dialogs.create_integer_dialog().

parent: Optional[QWidget]

The parent widget for the dialog.

dialog: Optional[QInputDialog]

The actual dialog widget instance.

edit_widget: Optional[QLineEdit]

The line edit that the user will enter the input into.

accept_button: Optional[QPushButton]

The entry confirmation button.

reject_button: Optional[QPushButton]

The input cancellation button.

result: Optional[int]

The result of parsing the user input.

shown

See qtrio.dialogs.DialogProtocol.shown.

finished

See qtrio.dialogs.BasicDialogProtocol.finished.

setup()[source]

See qtrio.dialogs.BasicDialogProtocol.setup().

Return type

None

teardown()[source]

See qtrio.dialogs.BasicDialogProtocol.teardown().

Return type

None

await wait()[source]

See qtrio.dialogs.DialogProtocol.wait().

Return type

int

class qtrio.dialogs.TextInputDialog(title=None, label=None, parent=None, dialog=None, accept_button=None, reject_button=None, line_edit=None, result=None)[source]

Bases: object

Manage a dialog for inputting an integer from the user. Generally instances should be built via qtrio.dialogs.create_text_input_dialog().

title: Optional[str]

The title of the dialog.

label: Optional[str]

The label for the input widget.

parent: Optional[QWidget]

The parent widget for the dialog.

dialog: Optional[QInputDialog]

The actual dialog widget instance.

accept_button: Optional[QPushButton]

The entry confirmation button.

reject_button: Optional[QPushButton]

The input cancellation button.

line_edit: Optional[QLineEdit]

The line edit that the user will enter the input into.

result: Optional[str]

The result of parsing the user input.

shown

See qtrio.dialogs.DialogProtocol.shown.

finished

See qtrio.dialogs.BasicDialogProtocol.finished.

setup()[source]

See qtrio.dialogs.BasicDialogProtocol.setup().

Return type

None

teardown()[source]

See qtrio.dialogs.BasicDialogProtocol.teardown().

Return type

None

await wait()[source]

See qtrio.dialogs.DialogProtocol.wait().

Return type

str

class qtrio.dialogs.FileDialog(file_mode, accept_mode, default_directory=None, default_file=None, options=PySide2.QtWidgets.QFileDialog.Option(0), parent=None, dialog=None, accept_button=None, reject_button=None, file_name_line_edit=None, result=None)[source]

Bases: object

Manage a dialog for allowing the user to select a file or directory. Generally instances should be built via qtrio.dialogs.create_file_save_dialog().

file_mode: <class ‘PySide2.QtWidgets.QFileDialog.FileMode’>

Controls whether the dialog is for picking an existing vs. new file or directory, etc.

accept_mode: <class ‘PySide2.QtWidgets.QFileDialog.AcceptMode’>

Specify an open vs. a save dialog.

default_directory: Optional[trio.Path]

The directory to be initially presented in the dialog.

default_file: Optional[trio.Path]

The file to be initially selected in the dialog.

options: <class ‘PySide2.QtWidgets.QFileDialog.Option’>

Miscellaneous options. See the Qt documentation.

parent: Optional[QWidget]

The parent widget for the dialog.

dialog: Optional[QFileDialog]

The actual dialog widget instance.

accept_button: Optional[QPushButton]

The confirmation button.

reject_button: Optional[QPushButton]

The cancellation button.

file_name_line_edit: Optional[QLineEdit]

The file name line edit widget.

result: Optional[trio.Path]

The path selected by the user.

shown

See qtrio.dialogs.DialogProtocol.shown.

finished

See qtrio.dialogs.BasicDialogProtocol.finished.

await set_path(path)[source]

Set the directory and enter the file name in the text box. Note that this does not select the file in the file list.

Parameters

path (Path) – The full path to the file to be set.

Return type

None

setup()[source]

See qtrio.dialogs.BasicDialogProtocol.setup().

Return type

None

teardown()[source]

See qtrio.dialogs.BasicDialogProtocol.teardown().

Return type

None

await wait()[source]

See qtrio.dialogs.DialogProtocol.wait().

Return type

Path

class qtrio.dialogs.MessageBox(title, text, icon, buttons, parent=None, dialog=None, accept_button=None, result=None)[source]

Bases: object

Manage a message box for notifying the user. Generally instances should be built via qtrio.dialogs.create_message_box().

title: str

The message box title.

text: str

The message text shown inside the dialog.

icon: <class ‘PySide2.QtWidgets.QMessageBox.Icon’>

The icon shown inside the dialog.

buttons: StandardButtons

The buttons to be shown in the dialog.

parent: Optional[QWidget]

The parent widget for the dialog.

dialog: Optional[QMessageBox]

The actual dialog widget instance.

accept_button: Optional[QPushButton]

The button to accept the dialog.

result: Optional[trio.Path]

Not generally relevant for a message box.

shown

See qtrio.dialogs.DialogProtocol.shown.

finished

See qtrio.dialogs.BasicDialogProtocol.finished.

setup()[source]

See qtrio.dialogs.BasicDialogProtocol.setup().

Return type

None

teardown()[source]

See qtrio.dialogs.BasicDialogProtocol.teardown().

Return type

None

await wait()[source]

See qtrio.dialogs.DialogProtocol.wait().

Return type

None

class qtrio.dialogs.ProgressDialog(title, text, cancel_button_text, minimum, maximum, parent=None, dialog=None, cancel_button=None)[source]

Bases: object

Manage a progress dialog for updating the user. Generally instances should be built via qtrio.dialogs.create_progress_dialog().

title: str

The message box title.

text: str

The message text shown as the progress bar label.

cancel_button_text: Optional[str]

The cancel button text.

minimum: int

The progress value corresponding to no progress.

maximum: int

The progress value corresponding to completion.

parent: Optional[QWidget]

The parent widget for the dialog.

dialog: Optional[QProgressDialog]

The actual dialog widget instance.

cancel_button: Optional[QPushButton]

The cancellation button.

shown

See qtrio.dialogs.DialogProtocol.shown.

finished

See qtrio.dialogs.BasicDialogProtocol.finished.

setup()[source]

See qtrio.dialogs.BasicDialogProtocol.setup().

Return type

None

teardown()[source]

See qtrio.dialogs.BasicDialogProtocol.teardown().

Return type

None

async with manage()[source]

A context manager to setup the progress dialog, cancel the managed context and teardown the dialog when done.

Return type

AsyncIterable[None]

Protocols

class qtrio.dialogs.BasicDialogProtocol(*args, **kwargs)[source]

Bases: Protocol

The minimal common interface used for working with QTrio dialogs. To check that a class implements this protocol see qtrio.dialogs.check_basic_dialog_protocol().

finished: qtrio._qt.Signal

The signal to be emitted when the dialog is finished.

setup()[source]

Setup and show the dialog. Emit qtrio.dialogs.DialogProtocol.shown when done.

Return type

None

teardown()[source]

Hide and teardown the dialog.

Return type

None

class qtrio.dialogs.DialogProtocol(*args, **kwargs)[source]

Bases: qtrio.dialogs.BasicDialogProtocol, Protocol

The common interface used for working with QTrio dialogs. To check that a class implements this protocol see qtrio.dialogs.check_dialog_protocol().

shown: qtrio._qt.Signal

The signal to be emitted when the dialog is shown.

await wait()[source]

Show the dialog, wait for the user interaction, and return the result.

Raises
Return type

object

Protocol Checkers

These callables can be used if you want to verify that your own classes properly implement the associated protocols. They are simple pass through decorators at runtime but when checking type hints they will result in a failure if the class does not implement the protocol.

qtrio.dialogs.check_basic_dialog_protocol = <qtrio._util.ProtocolChecker object>

Assert proper implementation of qtrio.dialogs.BasicDialogProtocol when type hint checking.

Parameters

cls – The class to verify.

Returns

The same class, unmodified, that was passed in.

qtrio.dialogs.check_dialog_protocol = <qtrio._util.ProtocolChecker object>

Assert proper implementation of qtrio.dialogs.DialogProtocol when type hint checking.

Parameters

cls – The class to verify.

Returns

The same class, unmodified, that was passed in.

Exceptions

class qtrio.QTrioException

Bases: Exception

Base exception for all QTrio exceptions.

class qtrio.NoOutcomesError

Bases: qtrio.QTrioException

Raised if you try to unwrap a qtrio.Outcomes which has no outcomes.

class qtrio.EventTypeRegistrationError

Bases: qtrio.QTrioException

Base class for various event type registration exceptions to inherit from.

class qtrio.EventTypeRegistrationFailedError

Bases: qtrio.EventTypeRegistrationError

Raised if the attempt to register a new event type fails.

class qtrio.RequestedEventTypeUnavailableError(requested_type, returned_type)

Bases: qtrio.EventTypeRegistrationError

Raised if the requested event type is unavailable.

class qtrio.EventTypeAlreadyRegisteredError

Bases: qtrio.EventTypeRegistrationError

Raised when a request is made to register an event type but a type has already been registered previously.

class qtrio.ReturnCodeError

Bases: qtrio.QTrioException

Wraps a QApplication return code as an exception.

class qtrio.InternalError

Bases: qtrio.QTrioException

Raised when an internal state is inconsistent.

class qtrio.UserCancelledError

Bases: qtrio.QTrioException

Raised when a user requested cancellation of an operation.

class qtrio.InvalidInputError

Bases: qtrio.QTrioException

Raised when invalid input is provided such as via a dialog.

Examples

Download

"""Thanks to njsmith for developing Trio guest mode and sharing the first example
integration with Qt.
"""
import contextlib
import math
import os
import time
import typing

import attr
import httpcore._async.http11
import httpx
import hyperlink
import qtrio
import trio

import qtrio.dialogs


# Default is 4096
httpcore._async.http11.AsyncHTTP11Connection.READ_NUM_BYTES = 100_000


def create_title(specific: str) -> str:
    return f"QTrio Download Example - {specific}"


@attr.s(auto_attribs=True, frozen=True)
class Progress:
    downloaded: int
    total: typing.Optional[int]
    first: bool


async def main(
    url: typing.Optional[typing.Union[str, hyperlink.URL]],
    destination: typing.Optional[typing.Union[str, os.PathLike]],
    fps: int = 60,
    text_input_dialog: typing.Optional[qtrio.dialogs.TextInputDialog] = None,
    file_dialog: typing.Optional[qtrio.dialogs.FileDialog] = None,
    progress_dialog: typing.Optional[qtrio.dialogs.ProgressDialog] = None,
    message_box: typing.Optional[qtrio.dialogs.MessageBox] = None,
    clock: typing.Callable[[], float] = time.monotonic,
    http_application: typing.Optional[typing.Callable[..., typing.Any]] = None,
) -> None:
    print("file dialog right inside", id(file_dialog), file_dialog, flush=True)
    converted_url: hyperlink.URL
    converted_destination: trio.Path

    with contextlib.suppress(qtrio.UserCancelledError):
        if url is None:
            if text_input_dialog is None:  # pragma: no cover
                text_input_dialog = qtrio.dialogs.create_text_input_dialog()

            text_input_dialog.title = create_title("Enter URL")
            text_input_dialog.label = "URL to download:"

            url = await text_input_dialog.wait()

        if isinstance(url, str):
            converted_url = hyperlink.URL.from_text(url)
        else:
            converted_url = url

        if destination is None:
            if file_dialog is None:  # pragma: no cover
                file_dialog = qtrio.dialogs.create_file_save_dialog()

            file_dialog.default_file = trio.Path(converted_url.path[-1])

            destination = await file_dialog.wait()

        converted_destination = trio.Path(destination)

        await get_dialog(
            url=converted_url,
            destination=converted_destination,
            fps=fps,
            progress_dialog=progress_dialog,
            message_box=message_box,
            clock=clock,
            http_application=http_application,
        )


async def get_dialog(
    url: hyperlink.URL,
    destination: trio.Path,
    fps: float,
    progress_dialog: typing.Optional[qtrio.dialogs.ProgressDialog] = None,
    message_box: typing.Optional[qtrio.dialogs.MessageBox] = None,
    clock: typing.Callable[[], float] = time.monotonic,
    http_application: typing.Optional[typing.Callable[..., typing.Any]] = None,
) -> None:
    if progress_dialog is None:  # pragma: no cover
        progress_dialog = qtrio.dialogs.create_progress_dialog()

    progress_dialog.title = create_title("Fetching")
    progress_dialog.text = f"Fetching {url}..."

    async with progress_dialog.manage():
        if progress_dialog.dialog is None:  # pragma: no cover
            raise qtrio.InternalError("Dialog not assigned while it is being managed.")

        # Always show the dialog
        progress_dialog.dialog.setMinimumDuration(0)

        start = clock()

        async for progress in get(
            url=url,
            destination=destination,
            update_period=1 / fps,
            http_application=http_application,
        ):
            if progress.first:
                if progress.total is None:
                    maximum = 0
                else:
                    maximum = progress.total

                progress_dialog.dialog.setMaximum(maximum)
                progress_dialog.dialog.setValue(0)

            if progress.total is not None:
                progress_dialog.dialog.setValue(progress.downloaded)

        end = clock()

    duration = end - start
    if duration == 0:
        # define this seems to happen when testing on Windows with an x86 Python
        if progress.downloaded > 0:
            bytes_per_second = math.inf
        else:  # pragma: no cover
            bytes_per_second = 0
    else:
        bytes_per_second = progress.downloaded / duration

    summary = "\n\n".join(
        [
            url.asText(),
            os.fspath(destination),
            f"Downloaded {progress.downloaded} bytes in {duration:.2f} seconds",
            f"{bytes_per_second:.2f} bytes/second",
        ]
    )

    if message_box is None:  # pragma: no cover
        message_box = qtrio.dialogs.create_message_box()

    message_box.title = create_title("Download Summary")
    message_box.text = summary

    await message_box.wait()


async def get(
    url: hyperlink.URL,
    destination: trio.Path,
    update_period: float,
    clock: typing.Callable[[], float] = time.monotonic,
    http_application: typing.Optional[typing.Callable[..., typing.Any]] = None,
) -> typing.AsyncIterable[Progress]:
    async with httpx.AsyncClient(app=http_application) as client:
        async with client.stream("GET", url.asText()) as response:
            raw_content_length = response.headers.get("content-length")
            if raw_content_length is None:
                content_length = None
            else:
                content_length = int(raw_content_length)

            progress = Progress(
                downloaded=0,
                total=content_length,
                first=True,
            )

            yield progress
            last_update = clock()

            progress = attr.evolve(progress, first=False)

            downloaded = 0

            async with (await destination.open("wb")) as file:
                async for chunk in response.aiter_raw():
                    downloaded += len(chunk)
                    await file.write(chunk)

                    if clock() - last_update > update_period:
                        progress = attr.evolve(progress, downloaded=downloaded)
                        yield progress
                        last_update = clock()

            if progress.downloaded != downloaded:
                progress = attr.evolve(progress, downloaded=downloaded)
                yield progress

Emissions

import typing

import attr
from qtpy import QtCore
from qtpy import QtGui
from qtpy import QtWidgets

import qtrio


class QSignaledWidget(QtWidgets.QWidget):
    """A :class:`QtWidgets.QWidget` with extra signals for events of interest.

    Attributes:
        closed: A signal that will be emitted after a close event.
    """

    closed = QtCore.Signal()
    shown = QtCore.Signal()

    def closeEvent(self, event: QtGui.QCloseEvent) -> None:
        """Detect close events and emit the ``closed`` signal."""

        super().closeEvent(event)
        if event.isAccepted():
            self.closed.emit()
        else:  # pragma: no cover
            pass

    def showEvent(self, event: QtGui.QShowEvent) -> None:
        """Detect show events and emit the ``shown`` signal."""

        super().showEvent(event)
        if event.isAccepted():
            self.shown.emit()
        else:  # pragma: no cover
            pass


@attr.s(auto_attribs=True)
class Window:
    """A manager for a simple window with increment and decrement buttons to change a
    counter which is displayed via a widget in the center.
    """

    widget: QSignaledWidget
    increment: QtWidgets.QPushButton
    decrement: QtWidgets.QPushButton
    label: QtWidgets.QLabel
    layout: QtWidgets.QHBoxLayout
    count: int = 0

    @classmethod
    def build(
        cls,
        title: str = "QTrio Emissions Example",
        parent: typing.Optional[QtWidgets.QWidget] = None,
    ) -> "Window":
        """Build and lay out the widgets that make up this window."""

        self = cls(
            widget=QSignaledWidget(parent),
            layout=QtWidgets.QHBoxLayout(),
            increment=QtWidgets.QPushButton(),
            decrement=QtWidgets.QPushButton(),
            label=QtWidgets.QLabel(),
        )

        self.widget.setWindowTitle(title)
        self.widget.setLayout(self.layout)

        self.increment.setText("+")
        self.decrement.setText("-")

        self.label.setText(str(self.count))

        self.layout.addWidget(self.decrement)
        self.layout.addWidget(self.label)
        self.layout.addWidget(self.increment)

        return self

    def increment_count(self) -> None:
        """Increment the counter and update the label."""

        self.count += 1
        self.label.setText(str(self.count))

    def decrement_count(self) -> None:
        """Decrement the counter and update the label."""

        self.count -= 1
        self.label.setText(str(self.count))

    def show(self) -> None:
        """Show the primary widget for this window."""

        self.widget.show()


async def main(window: typing.Optional[Window] = None) -> None:
    """Show the example window and iterate over the relevant signal emissions to respond
    to user interactions with the GUI.
    """
    if window is None:  # pragma: no cover
        window = Window.build()

    signals = [
        window.decrement.clicked,
        window.increment.clicked,
        window.widget.closed,
    ]

    async with qtrio.enter_emissions_channel(signals=signals) as emissions:
        window.show()

        async for emission in emissions.channel:
            if emission.is_from(window.decrement.clicked):
                window.decrement_count()
            elif emission.is_from(window.increment.clicked):
                window.increment_count()
            elif emission.is_from(window.widget.closed):
                break
            else:  # pragma: no cover
                raise qtrio.QTrioException(f"Unexpected emission: {emission}")

README

Qt
import typing

from qtpy import QtCore
from qtpy import QtWidgets


def create_input() -> QtWidgets.QInputDialog:
    dialog = QtWidgets.QInputDialog()
    dialog.setWindowTitle("Hello")
    dialog.setLabelText("Enter your name:")

    return dialog


def create_output() -> QtWidgets.QMessageBox:
    return QtWidgets.QMessageBox(
        QtWidgets.QMessageBox.Icon.Question,
        "Hello",
        "",
        QtWidgets.QMessageBox.Ok,
    )


class Main:
    def __init__(
        self,
        input_dialog: typing.Optional[QtWidgets.QInputDialog] = None,
        output_dialog: typing.Optional[QtWidgets.QMessageBox] = None,
    ):
        if input_dialog is None:  # pragma: no cover
            input_dialog = create_input()

        if output_dialog is None:  # pragma: no cover
            output_dialog = create_output()

        self.input_dialog = input_dialog
        self.output_dialog = output_dialog

    def setup(self) -> None:
        self.input_dialog.accepted.connect(self.input_accepted)
        self.input_dialog.rejected.connect(self.input_rejected)

        self.input_dialog.show()

    def input_accepted(self) -> None:
        name = self.input_dialog.textValue()

        self.output_dialog.setText(f"Hi {name}, welcome to the team!")

        self.output_dialog.finished.connect(self.output_finished)
        self.output_dialog.show()

    def input_rejected(self) -> None:
        QtCore.QCoreApplication.instance().quit()

    def output_finished(self) -> None:
        QtCore.QCoreApplication.instance().quit()


def main() -> None:  # pragma: no cover
    application = QtWidgets.QApplication([])
    application.setQuitOnLastWindowClosed(False)
    main_object = Main()
    main_object.setup()
    application.exec_()


if __name__ == "__main__":  # pragma: no cover
    main()
QTrio
import contextlib
import typing

from qtpy import QtWidgets

import qtrio
import qtrio.dialogs


def create_input() -> qtrio.dialogs.TextInputDialog:
    return qtrio.dialogs.create_text_input_dialog(
        title="Hello",
        label="Enter your name:",
    )


def create_output() -> qtrio.dialogs.MessageBox:
    return qtrio.dialogs.create_message_box(
        title="Hello",
        text="",
        icon=QtWidgets.QMessageBox.Icon.Question,
        buttons=QtWidgets.QMessageBox.Ok,
    )


async def main(
    input_dialog: typing.Optional[qtrio.dialogs.TextInputDialog] = None,
    output_dialog: typing.Optional[qtrio.dialogs.MessageBox] = None,
) -> None:
    if input_dialog is None:  # pragma: no cover
        input_dialog = create_input()

    if output_dialog is None:  # pragma: no cover
        output_dialog = create_output()

    with contextlib.suppress(qtrio.UserCancelledError):
        name = await input_dialog.wait()

        output_dialog.text = f"Hi {name}, welcome to the team!"

        await output_dialog.wait()


if __name__ == "__main__":  # pragma: no cover
    qtrio.run(main)

Developing

Contributing

Welcome to the team! By so much as reading the documentation or using the library you are providing a valuable testing service to QTrio. If you find anything amiss please do submit a GitHub issue to let us know. If you are comfortable providing a pull request to fix the issue, all the better.

Since QTrio is early in its life, it is both more prone to problems and more flexible about responding to them. Community involvement is one of the key pieces to evolving a powerful and usable library. We each bring different perspectives and needs that together expose the shortcomings and strengths of what we have made.

Testing

While developing it is important to make certain that the existing tests continue to pass and that any changes you make also have passing tests that exercise them. This will be done in it’s entirety for you when you submit PR. These runs will cover multiple operating systems, Python versions, and Qt libraries. They will also check formatting, the documentation build, and more.

Still, sometimes you would rather run the tests locally for potentially quicker feedback, the opportunity to debug, and less public observation of your every commit. You can run pytest, black, and sphinx directly from your own installation.

python -m venv testvenv
testvenv/bin/pip install --upgrade pip setuptools wheel
testvenv/bin/pip install --editable .[pyside2,checks,docs,tests]
testvenv/bin/pytest --pyargs qtrio

The CI test script, ci.sh, in the project root will run pytest with coverage (and fail to upload the coverage results, which is ok).

python -m venv testvenv
source testvenv/bin/activate
./ci.sh

Automatic code reformatting is handled by black.

python -m venv testvenv
testvenv/bin/pip install --upgrade pip setuptools wheel
testvenv/bin/pip install black
testvenv/bin/black .

Linting is handled by flake8.

python -m venv testvenv
testvenv/bin/pip install --upgrade pip setuptools wheel
testvenv/bin/pip install flake8
testvenv/bin/flake8 setup.py docs/ qtrio/

The documentation can be built with sphinx.

python -m venv testvenv
testvenv/bin/pip install --upgrade pip setuptools wheel
testvenv/bin/pip install --editable .[pyside2,docs]
source testenv/bin/activate
cd docs/
make html --always-make

I don’t like to write the hazardous command that does it, but it is good to remove the entire docs/build/ directory prior to each build of the documentation. After building the documentation it can be loaded in your browser at file:///path/to/qtrio/docs/build/html/index.html.

Reviewing

Manual checks
  • The change should happen

    • The bug is a bug

    • The feature is an improvement

    • The code belongs in QTrio, not another package

      • Unless it is a workaround in QTrio temporarily or because the proper library has declined to resolve the issue.

  • Relevant bugs or features are being tested

    • The line coverage provided by automatic coverage checks are valuable but you are the only one that can decide if the proper functionality is being tested

  • Documentation updates

    • Docstrings are present and accurate for all modules, classes, methods, and functions including private ones and tests.

    • For bug fixes consider if the docs should be updated to clarify proper behavior

    • For feature additions consider if prose in the docs should be updated in addition to the docstrings.

  • The change is described for the user

    • If this is a change relevant to users, there should be a newsfragment file as follows

    • Newsfragment file name has the proper issue or PR number and change type

    • The contents describe the change well for users

    • Proper Sphinx references are used where appropriate

Automatic checks
  • Full test suite passes across:

    • operating systems

    • Python versions

    • Qt libraries

  • All code and tests lines are fully covered when running tests

  • Code is formatted per Black

  • Code passes flake8 checks

  • Docs build successfully including newsfragment, if present

  • Branch is up to date with master

Preparing a release

Things to do for releasing:

  • check for open issues / pull requests that really should be in the release

    • come back when these are done

    • … or ignore them and do another release next week

  • check for deprecations “long enough ago” (two months or two releases, whichever is longer)

    • remove affected code

  • do the actual release changeset

    • select new version number

    • checkout a new branch release-vX.Y.Z

    • bump version number in qtrio/_version.py

      • remove +dev tag from version number

    • run towncrier

  • push to your personal repository

  • create pull request to altendky/qtrio’s “master” branch

  • verify that all checks succeeded

  • get a review

  • tag with vX.Y.Z, push tag

  • download wheel and sdist from build artifacts and unpack

  • push to PyPI:

    twine upload dist/*

  • replace the +dev version tag in the same pull request as vX.Y.Z+dev

  • merge the release pull request

Release history

Qtrio 0.4.1 (2021-01-06)

Features

  • Constrained all dependencies for better long term compatibility and reduced pip resolution time. (#211)

Qtrio 0.4.0 (2021-01-04)

Features
Bugfixes

  • Include the missing py.typed file that is used to state there are type hints to process. (#206)

Improved documentation

Qtrio 0.3.0 (2020-10-16)

Headline features

  • Integrate with pytest-trio for testing. (#9)

  • Python 3.9 supported. (#113)

Breaking changes

  • Removed qtrio.host() in favor of pytest-trio for testing. (#9)

  • qtrio.run() returns the passed async function’s result instead of a combined Trio and Qt outcome. (#9)

QTrio 0.2.0 (2020-09-19)

Headline features
Features
Bugfixes
Improved documentation

  • Badges now in new resources section of readme and main doc page. (#103)

  • Classifiers for 3.6, 3.7, and 3.8 are included. (#104)

  • Link to issues included in resources section. (#106)

  • List all resource URLs in PyPI project URLs. (#107)

  • Use stable for outcome intersphinx links. (#109)

  • Add section about Installation, mostly to describe extras. (#155)

  • Show [sources] links in documentation linked to included code. (#168)

  • Update the layer 2 example to use async for _ in emissions.channel:. (#173)

For contributors

  • Shift to a single qtrio._tests package rather than distributing with one _tests per code package. (#139)

  • pytest type hints are no longer ignored. Version 6 or later required. (#153)

  • black config updated, use black . to format. (#174)

QTrio 0.1.0 (2020-07-10)

  • Initial release