Example D-Bus usage
Examples using dnf5daemon server via D-Bus from Python.
Print upgrades
Print all available upgrades, the repository they come from and severity of associated advisory if present.
1import dbus
2import os
3
4DNFDAEMON_BUS_NAME = 'org.rpm.dnf.v0'
5DNFDAEMON_OBJECT_PATH = '/' + DNFDAEMON_BUS_NAME.replace('.', '/')
6
7IFACE_SESSION_MANAGER = '{}.SessionManager'.format(DNFDAEMON_BUS_NAME)
8IFACE_REPO = '{}.rpm.Repo'.format(DNFDAEMON_BUS_NAME)
9IFACE_REPOCONF = '{}.rpm.RepoConf'.format(DNFDAEMON_BUS_NAME)
10IFACE_RPM = '{}.rpm.Rpm'.format(DNFDAEMON_BUS_NAME)
11IFACE_GOAL = '{}.Goal'.format(DNFDAEMON_BUS_NAME)
12IFACE_ADVISORY = '{}.Advisory'.format(DNFDAEMON_BUS_NAME)
13
14
15bus = dbus.SystemBus()
16iface_session = dbus.Interface(
17 bus.get_object(DNFDAEMON_BUS_NAME, DNFDAEMON_OBJECT_PATH),
18 dbus_interface=IFACE_SESSION_MANAGER)
19
20session = iface_session.open_session(
21 dbus.Dictionary({}, signature=dbus.Signature('sv')))
22
23iface_rpm = dbus.Interface(
24 bus.get_object(DNFDAEMON_BUS_NAME, session),
25 dbus_interface=IFACE_RPM)
26iface_advisory = dbus.Interface(
27 bus.get_object(DNFDAEMON_BUS_NAME, session),
28 dbus_interface=IFACE_ADVISORY)
29
30# Information about the severity of the upgrade is stored in the updateinfo
31# repository metadata and is not directly accessible from the list of possible
32# upgrades. This means we first need to retrieve all advisories and then search
33# them for severity for all potential upgrade packages.
34
35# First get all available advisories, we are interested only in "severity" and
36# "collections" (list of packages and modules) fields.
37options = {
38 "advisory_attrs": [
39 # "advisoryid",
40 # "name",
41 # "title",
42 # "type",
43 "severity",
44 # "status",
45 # "vendor",
46 # "description",
47 # "buildtime",
48 # "message",
49 # "rights",
50 "collections",
51 # "references",
52 ],
53 "availability": "available",
54}
55advisory_list = iface_advisory.list(options)
56
57# auxiliary dictionary to map package NEVRA to severity of advisory it belongs to
58nevra_to_severity = dict()
59for adv in advisory_list:
60 severity = str(adv["severity"])
61 for col in adv["collections"]:
62 if "packages" in col:
63 for pkg in col["packages"]:
64 nevra = pkg["nevra"]
65 nevra_to_severity[nevra] = severity
66
67
68# retrieve potential upgrades and print the packages along with their
69# respective severities
70options = {
71 "package_attrs": [
72 "nevra",
73 "repo_id",
74 ],
75 "scope": "upgrades",
76 "latest-limit": 1,
77}
78upgrades = iface_rpm.list(options)
79for pkg in upgrades:
80 nevra = pkg["nevra"]
81 severity = nevra_to_severity.get(nevra, "<unknown>")
82 print("{} (@{}): {}".format(nevra, pkg["repo_id"], severity))
System upgrade
Perform a system-upgrade.
1import dbus
2
3DNFDAEMON_BUS_NAME = 'org.rpm.dnf.v0'
4DNFDAEMON_OBJECT_PATH = '/' + DNFDAEMON_BUS_NAME.replace('.', '/')
5
6IFACE_SESSION_MANAGER = '{}.SessionManager'.format(DNFDAEMON_BUS_NAME)
7IFACE_RPM = '{}.rpm.Rpm'.format(DNFDAEMON_BUS_NAME)
8IFACE_GOAL = '{}.Goal'.format(DNFDAEMON_BUS_NAME)
9
10
11bus = dbus.SystemBus()
12iface_session = dbus.Interface(
13 bus.get_object(DNFDAEMON_BUS_NAME, DNFDAEMON_OBJECT_PATH),
14 dbus_interface=IFACE_SESSION_MANAGER)
15
16# set the releasever to the new distribution release
17session = iface_session.open_session(
18 dbus.Dictionary({"releasever": "40"}, signature=dbus.Signature('sv')))
19
20iface_rpm = dbus.Interface(
21 bus.get_object(DNFDAEMON_BUS_NAME, session),
22 dbus_interface=IFACE_RPM)
23iface_goal = dbus.Interface(
24 bus.get_object(DNFDAEMON_BUS_NAME, session),
25 dbus_interface=IFACE_GOAL)
26
27# Add system upgrade to the transaction
28options = {
29 "mode": "distrosync",
30}
31iface_rpm.system_upgrade(options)
32
33# resolve the transaction
34resolved, result = iface_goal.resolve({})
35
36# now you can print the transaction table and ask the user for confirmation
37print("Resolved.")
38
39if result == 0:
40 # execute the transaction offline (durint the next reboot)
41 iface_goal.do_transaction({"offline": True}, timeout=2000)
42 print("Reboot to continue with system upgrade...")
43else:
44 errors = iface_goal.get_transaction_problems_string()
45 print("Errors while resolving the transaction:")
46 for error in errors:
47 print(error)
list_fd()
D-Bus API bindings for dnfdaemon org.rpm.dnf.v0.rpm.Rpm.list_fd() exmaple.
1import dbus
2
3DNFDAEMON_BUS_NAME = 'org.rpm.dnf.v0'
4DNFDAEMON_OBJECT_PATH = '/' + DNFDAEMON_BUS_NAME.replace('.', '/')
5
6IFACE_SESSION_MANAGER = '{}.SessionManager'.format(DNFDAEMON_BUS_NAME)
7IFACE_RPM = '{}.rpm.Rpm'.format(DNFDAEMON_BUS_NAME)
8IFACE_GOAL = '{}.Goal'.format(DNFDAEMON_BUS_NAME)
9
10
11bus = dbus.SystemBus()
12iface_session = dbus.Interface(
13 bus.get_object(DNFDAEMON_BUS_NAME, DNFDAEMON_OBJECT_PATH),
14 dbus_interface=IFACE_SESSION_MANAGER)
15
16# set the releasever to the new distribution release
17session = iface_session.open_session(
18 dbus.Dictionary({"releasever": "40"}, signature=dbus.Signature('sv')))
19
20iface_rpm = dbus.Interface(
21 bus.get_object(DNFDAEMON_BUS_NAME, session),
22 dbus_interface=IFACE_RPM)
23iface_goal = dbus.Interface(
24 bus.get_object(DNFDAEMON_BUS_NAME, session),
25 dbus_interface=IFACE_GOAL)
26
27# Add system upgrade to the transaction
28options = {
29 "mode": "distrosync",
30}
31iface_rpm.system_upgrade(options)
32
33# resolve the transaction
34resolved, result = iface_goal.resolve({})
35
36# now you can print the transaction table and ask the user for confirmation
37print("Resolved.")
38
39if result == 0:
40 # execute the transaction offline (durint the next reboot)
41 iface_goal.do_transaction({"offline": True}, timeout=2000)
42 print("Reboot to continue with system upgrade...")
43else:
44 errors = iface_goal.get_transaction_problems_string()
45 print("Errors while resolving the transaction:")
46 for error in errors:
47 print(error)