ceph-volume: add type annotations to devices.lvm.listing

This commit adds the Python type annotations to `devices.lvm.listing`.

Signed-off-by: Guillaume Abrioux <[email protected]>

Author: guits

src/ceph-volume/ceph_volume/api/lvm.py

@@ -352,6 +352,7 @@ class PVolume:
     def __init__(self, **kw: Any) -> None:
         self.pv_name: str = ''
         self.pv_uuid: str = ''
+        self.lv_uuid: str = ''
         for k, v in kw.items():
             setattr(self, k, v)
         self.pv_api = kw

src/ceph-volume/ceph_volume/devices/lvm/listing.py

@@ -2,10 +2,12 @@
 import argparse
 import json
 import logging
+import typing
 from textwrap import dedent
 from ceph_volume import decorators
 from ceph_volume.api import lvm as api
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -22,12 +24,12 @@
       {tag_name: <25} {value}"""
 
 
-def readable_tag(tag):
+def readable_tag(tag: str) -> str:
     actual_name = tag.split('.')[-1]
     return actual_name.replace('_', ' ')
 
 
-def pretty_report(report):
+def pretty_report(report: typing.Dict[str, typing.Any]) -> None:
     output = []
     for osd_id, devices in sorted(report.items()):
         output.append(
@@ -60,7 +62,7 @@ def pretty_report(report):
     print(''.join(output))
 
 
-def direct_report():
+def direct_report() -> typing.Dict[str, typing.Any]:
     """
     Other non-cli consumers of listing information will want to consume the
     report without the need to parse arguments or other flags. This helper
@@ -71,15 +73,15 @@ def direct_report():
 
 
 # TODO: Perhaps, get rid of this class and simplify this module further?
-class List(object):
+class List:
 
     help = 'list logical volumes and devices associated with Ceph'
 
-    def __init__(self, argv):
+    def __init__(self, argv: typing.List[str]) -> None:
         self.argv = argv
 
     @decorators.needs_root
-    def list(self, args):
+    def list(self, args: argparse.Namespace) -> None:
         report = self.single_report(args.device) if args.device else \
                  self.full_report()
         if args.format == 'json':
@@ -94,12 +96,12 @@ def list(self, args):
                 raise SystemExit('No valid Ceph lvm devices found')
             pretty_report(report)
 
-    def create_report(self, lvs):
+    def create_report(self, lvs: typing.List[api.Volume]) -> typing.Dict[str, typing.Any]:
         """
         Create a report for LVM dev(s) passed. Returns '{}' to denote failure.
         """
 
-        report = {}
+        report: typing.Dict[str, typing.Any] = {}
 
         pvs = api.get_pvs()
 
@@ -120,7 +122,7 @@ def create_report(self, lvs):
 
         return report
 
-    def create_report_non_lv_device(self, lv):
+    def create_report_non_lv_device(self, lv: api.Volume) -> typing.Dict[str, typing.Any]:
         report = {}
         if lv.tags.get('ceph.type', '') in ['data', 'block']:
             for dev_type in ['journal', 'wal', 'db']:
@@ -134,13 +136,13 @@ def create_report_non_lv_device(self, lv):
                               'path': dev}
         return report
 
-    def full_report(self):
+    def full_report(self) -> typing.Dict[str, typing.Any]:
         """
         Create a report of all Ceph LVs. Returns '{}' to denote failure.
         """
         return self.create_report(api.get_lvs())
 
-    def single_report(self, osd):
+    def single_report(self, osd: str) -> typing.Dict[str, typing.Any]:
         """
         Generate a report for a single device. This can be either a logical
         volume in the form of vg/lv, a device with an absolute path like
@@ -154,8 +156,12 @@ def single_report(self, osd):
             lv = api.get_lvs_from_path(osd)
         else:
             vg_name, lv_name = osd.split('/')
-            lv = [api.get_single_lv(filters={'lv_name': lv_name,
-                                             'vg_name': vg_name})]
+            _lv = api.get_single_lv(filters={'lv_name': lv_name,
+                                             'vg_name': vg_name})
+            if _lv is not None:
+                lv = [_lv]
+            else:
+                raise RuntimeError(f'Unexpected error while reporting {osd}')
 
         report = self.create_report(lv)
 
@@ -166,16 +172,16 @@ def single_report(self, osd):
                     'ceph.{}_device'.format(dev_type): osd})
                 if lvs:
                     # just taking the first lv here should work
-                    lv = lvs[0]
-                    phys_dev = self.create_report_non_lv_device(lv)
-                    osd_id = lv.tags.get('ceph.osd_id')
+                    _lv = lvs[0]
+                    phys_dev = self.create_report_non_lv_device(_lv)
+                    osd_id = _lv.tags.get('ceph.osd_id')
                     if osd_id:
                         report[osd_id] = [phys_dev]
 
 
         return report
 
-    def main(self):
+    def main(self) -> None:
         sub_command_help = dedent("""
         List devices or logical volumes associated with Ceph. An association is
         determined if a device has information relating to an OSD. This is