Storage mangment API: disks infos

debian/control

@@ -15,6 +15,7 @@ Depends: ${python3:Depends}, ${misc:Depends}
  , python3-miniupnpc, python3-dbus, python3-jinja2
  , python3-toml, python3-packaging, python3-publicsuffix2
  , python3-ldap, python3-zeroconf (>= 0.36), python3-lexicon,
+ , python3-pyudev
  , python-is-python3
  , nginx, nginx-extras (>=1.18)
  , apt, apt-transport-https, apt-utils, aptitude, dirmngr

share/actionsmap.yml

@@ -2082,3 +2082,15 @@ diagnosis:
                     help: Remove a filter (it should be an existing filter as listed with "ignore --list")
                     nargs: "*"
                     metavar: CRITERIA
+
+
+#############################
+#           Storage         #
+#############################
+storage:
+  category_help: Manage hard-disks, filesystem, pools
+  actions:
+    # storage_disks_list
+    disks_infos:
+      action_help: Gets infos about storage disks currently attached to this device with their caracteristics
+      api: GET /storage/disks/infos

src/disks.py

@@ -0,0 +1,66 @@
+from collections import OrderedDict
+
+import pyudev
+import psutil
+
+
+IGNORE_DISKS = "sr", "md", "dm-", "loop", "zd", "pmem"
+
+
+def disks_infos():
+    context = pyudev.Context()
+    partitions = {it.device: it for it in psutil.disk_partitions()}
+
+    def _filter_device(dev):
+        return dev.parent is not None and not any(
+            map(dev.sys_name.startswith, IGNORE_DISKS)
+        )
+
+    def _disk_map(dev):
+        attrs = {
+            "devname": dev.device_node,
+            "model": dev.get("ID_MODEL", "<unknown model>"),
+            "serial": dev.get("ID_SERIAL_SHORT", "<unknown serial>"),
+        }
+
+        try:
+            attrs["size"] = dev.attributes.asint("size")
+        except (AttributeError, UnicodeError, ValueError):
+            attrs["size"] = "<unknown size>"
+
+        attrs["links"] = list(sorted(it for it in dev.device_links))
+
+        for child_dev in sorted(
+            filter(_filter_device, dev.children), key=lambda it: it.device_node
+        ):
+            attrs.setdefault("partitions", {})
+            if child_dev.device_node in partitions:
+                attrs["partitions"][child_dev.sys_name] = {
+                    "devname": child_dev.device_node,
+                    "filesystem": partitions[child_dev.device_node].fstype,
+                    "mountpoint": partitions[child_dev.device_node].mountpoint,
+                }
+
+        if "partitions" not in attrs:
+            attrs["partitions"] = "<no partition>"
+
+        return dev.sys_name, attrs
+
+    return OrderedDict(
+        sorted(
+            (
+                _disk_map(it)
+                for it in filter(
+                    _filter_device,
+                    context.list_devices(subsystem="block", DEVTYPE="disk"),
+                )
+            ),
+            key=lambda d: d[0],
+        )
+    )
+
+
+if __name__ == "__main__":
+    import json
+
+    print(json.dumps(disks_infos(), indent=4))

src/storage.py

@@ -0,0 +1,4 @@
+def storage_disks_infos():
+    from yunohost.disks import disks_infos
+
+    return disks_infos()

src/tests/test_storage.py

@@ -0,0 +1,8 @@
+def setup_function(function):
+    ...
+
+def teardown_function(function):
+    ...
+
+def test_storage_disks_infos():
+    ...