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 attr
import qtrio
from qts import QtWidgets
import trio
import trio_typing
@attr.s(auto_attribs=True, eq=False)
class Widget:
message: str
change_delay: float = 0.5
close_delay: float = 3
label: QtWidgets.QLabel = attr.ib(factory=QtWidgets.QLabel)
text_changed = qtrio.Signal(str)
done_event: trio.Event = attr.ib(factory=trio.Event)
def setup(self) -> None:
self.label.setText(self.message)
self.label.setText("")
async def show(self) -> None:
# TODO: maybe raise if already started?
# start big enough to fit the whole message
self.label.setText(self.message)
self.label.show()
self.label.setText("")
def set_text(self, text: str) -> None:
self.label.setText(text)
self.text_changed.emit(text)
async def serve(
self,
*,
task_status: trio_typing.TaskStatus[None] = trio.TASK_STATUS_IGNORED,
) -> None:
await self.show()
task_status.started()
partial_message = ""
for character in self.message:
partial_message += character
await trio.sleep(self.change_delay)
self.set_text(partial_message)
await trio.sleep(self.close_delay)
self.done_event.set()
async def start_widget(
message: str,
change_delay: float = 0.5,
close_delay: float = 3,
hold_event: typing.Optional[trio.Event] = None,
*,
cls: typing.Type[Widget] = Widget,
task_status: trio_typing.TaskStatus[Widget] = trio.TASK_STATUS_IGNORED,
) -> None:
self = cls(message=message, change_delay=change_delay, close_delay=close_delay)
self.setup()
task_status.started(self)
if hold_event is not None:
await hold_event.wait()
await self.serve()
if __name__ == "__main__": # pragma: no cover
qtrio.run(start_widget, "Hello world.")
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 qts import QtWidgets
import trio
import trio_typing
@attr.s(auto_attribs=True, eq=False)
class Widget:
message: str
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)
text_changed = qtrio.Signal(str)
def setup(self) -> None:
self.button.setText("More")
self.layout.addWidget(self.button)
self.layout.addWidget(self.label)
self.widget.setLayout(self.layout)
async def show(self) -> None:
# TODO: maybe raise if already started?
# start big enough to fit the whole message
self.label.setText(self.message)
self.widget.show()
self.label.setText("")
def set_text(self, text: str) -> None:
self.label.setText(text)
self.text_changed.emit(text)
async def serve(
self,
*,
task_status: trio_typing.TaskStatus[None] = trio.TASK_STATUS_IGNORED,
) -> None:
async with qtrio.enter_emissions_channel(
signals=[self.button.clicked]
) as emissions:
i = 1
await self.show()
task_status.started()
async for _ in emissions.channel: # pragma: no branch
self.set_text(self.message[:i])
i += 1
if i > len(self.message):
break
# wait for another click to finish
await emissions.channel.receive()
async def start_widget(
message: str,
hold_event: typing.Optional[trio.Event] = None,
*,
cls: typing.Type[Widget] = Widget,
task_status: trio_typing.TaskStatus[Widget] = trio.TASK_STATUS_IGNORED,
) -> None:
self = cls(message=message)
self.setup()
task_status.started(self)
if hold_event is not None:
await hold_event.wait()
await self.serve()
if __name__ == "__main__": # pragma: no cover
qtrio.run(start_widget, "Hello world.")
Layer 3 - Best Friends¶
This space intentionally left blank.
(for now… sorry)