Skip to content

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

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 description
  • ext: 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()