Generating BlueChi clients¶
BlueChi provides introspection data for its public API. These XML files are located in the data directory of the project and can be also be used as input to generate clients.
There is a variety of code generation tools for various programming languages. There is a list
C
: gdbus-codegenGo
: dbus-codegen-goRust
: dbus-codegen-rust- and many more
Using generated code for clients has the advantage of reducing boilerplate code as well as providing abstracted functions for the different interfaces and operations on them.
Typed Python Client¶
For python
, the dynamically generated proxies (e.g. from dasbus) don’t require any generation and work well on their own. Because of that there are no larger projects to generate python code from a D-Bus specification. These proxies, however, don’t provide (type) hints.
Therefore, the BlueChi project contains its own generator to output typed python bindings based on introspection data.
Installation¶
The auto-generated, typed python bindings are published as an RPM package as well as a PyPI package. So it can be installed via:
# install from rpm package
dnf install python3-bluechi
# or from PyPI
pip install bluechi
Or build and install it directly from source:
git clone git@github.com:containers/bluechi.git
cd bluechi
pip install -r src/bindings/generator/requirements.txt
bash build-scripts/generate-bindings.sh python
bash build-scripts/build-bindings.sh python
pip install src/bindings/python/dist/bluechi-0.6.0-py3-none-any.whl
Usage¶
The bluechi
python package consists of two subpackages:
api
: auto-generated code based the BlueChi D-BUS API descriptionext
: custom written code to simplify common tasks
Functions from the auto-generated api
subpackage reduce boilerplate code. It also removes the need to explicitly specify the D-Bus name, paths and interfaces and offers abstracted and easy to use functions. For example, lets compare the python code for listing all nodes.
Using the auto-generated bindings, the code would look like this:
#!/usr/bin/env python
# SPDX-License-Identifier: MIT-0
from bluechi.api import Manager
for node in Manager().list_nodes():
# node[name, obj_path, status]
print(f"Node: {node[0]}, State: {node[2]}")
The functions from the ext
subpackage leverage the functions in api
and combine them to simplify recurring tasks. A good example here is to wait for a systemd job to finish. When starting a systemd unit via BlueChi (e.g. see start unit example), it returns the path to the corresponding job for the start operation of the systemd unit. This means that the unit hasn’t started yet, but is queued to be. BlueChi will emit a signal when the job has been completed. Starting a unit and waiting for this signal has been encapsulated by the ext.Unit
class:
#!/usr/bin/env python
# SPDX-License-Identifier: MIT-0
from bluechi.ext import Unit
result = Unit("my-node-name").start_unit("chronyd.service")
print(result)
If there is a common task for which bluechi
can provide a simplifying function, please submit an new RFE request.
For more examples, please have a look at the next section
More examples¶
The following code snippets showcase more examples on how bluechi
can be used.
Stop a unit¶
#!/usr/bin/env python
# SPDX-License-Identifier: MIT-0
from bluechi.ext import Unit
result = Unit("my-node-name").stop_unit("chronyd.service")
print(result)
Enable a unit¶
#!/usr/bin/env python
# SPDX-License-Identifier: MIT-0
from bluechi.ext import Unit
response = Unit("my-node-name").enable_unit_files(
["chronyd.service", "bluechi-agent.service"]
)
if response.carries_install_info:
print("The unit files included enablement information")
else:
print("The unit files did not include any enablement information")
for change in response.changes:
if change.change_type == "symlink":
print(
f"Created symlink {change.symlink_file}"
" -> {enabled_service_info.symlink_dest}"
)
elif change.change_type == "unlink":
print(f'Removed "{change.symlink_file}".')
Disable a unit¶
#!/usr/bin/env python
# SPDX-License-Identifier: MIT-0
from bluechi.ext import Unit
Unit("my-node-name").disable_unit_files(["chronyd.service", "bluechi-agent.service"])
List all units on all nodes¶
#!/usr/bin/env python
# SPDX-License-Identifier: MIT-0
from bluechi.api import Node
for unit in Node("my-node-name").list_units():
# unit[name, description, ...]
print(f"{unit[0]} - {unit[1]}")
List all active services on all nodes¶
#!/usr/bin/env python
# SPDX-License-Identifier: MIT-0
from bluechi.api import Manager
for unit in Manager().list_units():
# unit[node, name, description, load_state, active_state, ...]
if unit[4] == "active" and unit[1].endswith(".service"):
print(f"Node: {unit[0]}, Unit: {unit[1]}")
Get a unit property¶
#!/usr/bin/env python
# SPDX-License-Identifier: MIT-0
from bluechi.api import Node
cpu_weight = Node("my-node-name").get_unit_property(
"ldconfig.service", "org.freedesktop.systemd1.Service", "CPUWeight"
)
print(cpu_weight)
Set a unit property¶
#!/usr/bin/env python
# SPDX-License-Identifier: MIT-0
from dasbus.typing import Variant
from bluechi.api import Node
Node("my-node-name").set_unit_properties(
"ldconfig.service", False, [("CPUWeight", Variant("t", 18446744073709551615))]
)
Monitor the connections of all nodes¶
# SPDX-License-Identifier: MIT-0
from bluechi.api import Manager, Node
from dasbus.loop import EventLoop
from dasbus.typing import Variant
loop = EventLoop()
nodes = []
for node in Manager().list_nodes():
n = Node(node[0])
def changed_wrapper(node_name: str):
def on_connection_status_changed(status: Variant):
con_status = status.get_string()
print(f"Node {node_name}: {con_status}")
return on_connection_status_changed
n.on_status_changed(changed_wrapper(n.name))
nodes.append(n)
loop.run()
Monitor the connection on the managed node¶
# SPDX-License-Identifier: MIT-0
from bluechi.api import Agent
from dasbus.loop import EventLoop
from dasbus.typing import Variant
def on_connection_status_changed(status: Variant):
con_status = status.get_string()
print(con_status)
loop = EventLoop()
agent = Agent()
agent.on_status_changed(on_connection_status_changed)
loop.run()