Format code with black

Author: github-actions[bot]

control/__main__.py

@@ -12,10 +12,12 @@
 from .config import GatewayConfig
 from .utils import GatewayLogger
 
-if __name__ == '__main__':
-    parser = argparse.ArgumentParser(prog="python3 -m control",
-                                     description="Manage NVMe gateways",
-                                     formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        prog="python3 -m control",
+        description="Manage NVMe gateways",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
     parser.add_argument(
         "-c",
         "--config",

control/cephutils.py

@@ -15,28 +15,31 @@
 import json
 from .utils import GatewayLogger
 
+
 class CephUtils:
-    """Miscellaneous functions which connect to Ceph
-    """
+    """Miscellaneous functions which connect to Ceph"""
 
     def __init__(self, config):
         self.logger = GatewayLogger(config).logger
-        self.ceph_conf = config.get_with_default("ceph", "config_file", "/etc/ceph/ceph.conf")
+        self.ceph_conf = config.get_with_default(
+            "ceph", "config_file", "/etc/ceph/ceph.conf"
+        )
         self.rados_id = config.get_with_default("ceph", "id", "")
         self.anagroup_list = []
         self.last_sent = time.time()
 
     def execute_ceph_monitor_command(self, cmd):
-         self.logger.debug(f"Execute monitor command: {cmd}")
-         with rados.Rados(conffile=self.ceph_conf, rados_id=self.rados_id) as cluster:
-            rply = cluster.mon_command(cmd, b'')
+        self.logger.debug(f"Execute monitor command: {cmd}")
+        with rados.Rados(conffile=self.ceph_conf, rados_id=self.rados_id) as cluster:
+            rply = cluster.mon_command(cmd, b"")
             self.logger.debug(f"Monitor reply: {rply}")
             return rply
+
     def get_gw_id_owner_ana_group(self, pool, group, anagrp):
-        str = '{' + f'"prefix":"nvme-gw show", "pool":"{pool}", "group":"{group}"' + '}'
+        str = "{" + f'"prefix":"nvme-gw show", "pool":"{pool}", "group":"{group}"' + "}"
         self.logger.debug(f"nvme-show string: {str}")
         rply = self.execute_ceph_monitor_command(str)
-        self.logger.debug(f"reply \"{rply}\"")
+        self.logger.debug(f'reply "{rply}"')
         conv_str = rply[1].decode()
         data = json.loads(conv_str)
 
@@ -45,31 +48,35 @@ def get_gw_id_owner_ana_group(self, pool, group, anagrp):
         comp_str = f"{anagrp}: ACTIVE"
         for gateway in data["Created Gateways:"]:
             if comp_str in gateway["ana states"]:
-              gw_id = gateway["gw-id"]
-              self.logger.debug(f"found gw owner of anagrp {anagrp}: gw {gw_id}")
-              break
+                gw_id = gateway["gw-id"]
+                self.logger.debug(f"found gw owner of anagrp {anagrp}: gw {gw_id}")
+                break
         return gw_id
 
     def get_number_created_gateways(self, pool, group):
         now = time.time()
-        if (now - self.last_sent) < 10 and self.anagroup_list :
-             self.logger.info(f"Caching response of the monitor: {self.anagroup_list}")
-             return self.anagroup_list
-        else :
+        if (now - self.last_sent) < 10 and self.anagroup_list:
+            self.logger.info(f"Caching response of the monitor: {self.anagroup_list}")
+            return self.anagroup_list
+        else:
             try:
                 self.anagroup_list = []
                 self.last_sent = now
-                str = '{' + f'"prefix":"nvme-gw show", "pool":"{pool}", "group":"{group}"' + '}'
+                str = (
+                    "{"
+                    + f'"prefix":"nvme-gw show", "pool":"{pool}", "group":"{group}"'
+                    + "}"
+                )
                 self.logger.debug(f"nvme-show string: {str}")
                 rply = self.execute_ceph_monitor_command(str)
-                self.logger.debug(f"reply \"{rply}\"")
+                self.logger.debug(f'reply "{rply}"')
                 conv_str = rply[1].decode()
                 pos = conv_str.find("[")
                 if pos != -1:
                     new_str = conv_str[pos + len("[") :]
-                    pos     = new_str.find("]")
-                    new_str = new_str[: pos].strip()
-                    int_str_list = new_str.split(' ')
+                    pos = new_str.find("]")
+                    new_str = new_str[:pos].strip()
+                    int_str_list = new_str.split(" ")
                     self.logger.debug(f"new_str : {new_str}")
                     for x in int_str_list:
                         self.anagroup_list.append(int(x))
@@ -86,17 +93,27 @@ def get_number_created_gateways(self, pool, group):
     def fetch_and_display_ceph_version(self):
         try:
             rply = self.execute_ceph_monitor_command('{"prefix":"mon versions"}')
-            ceph_ver = rply[1].decode().removeprefix("{").strip().split(":")[0].removeprefix('"').removesuffix('"')
+            ceph_ver = (
+                rply[1]
+                .decode()
+                .removeprefix("{")
+                .strip()
+                .split(":")[0]
+                .removeprefix('"')
+                .removesuffix('"')
+            )
             ceph_ver = ceph_ver.removeprefix("ceph version ")
-            self.logger.info(f"Connected to Ceph with version \"{ceph_ver}\"")
+            self.logger.info(f'Connected to Ceph with version "{ceph_ver}"')
         except Exception:
             self.logger.exception(f"Failure fetching Ceph version:")
             pass
 
     def fetch_ceph_fsid(self) -> str:
         fsid = None
         try:
-            with rados.Rados(conffile=self.ceph_conf, rados_id=self.rados_id) as cluster:
+            with rados.Rados(
+                conffile=self.ceph_conf, rados_id=self.rados_id
+            ) as cluster:
                 fsid = cluster.get_fsid()
         except Exception:
             self.logger.exception(f"Failure fetching Ceph fsid:")
@@ -105,7 +122,9 @@ def fetch_ceph_fsid(self) -> str:
 
     def pool_exists(self, pool) -> bool:
         try:
-            with rados.Rados(conffile=self.ceph_conf, rados_id=self.rados_id) as cluster:
+            with rados.Rados(
+                conffile=self.ceph_conf, rados_id=self.rados_id
+            ) as cluster:
                 if cluster.pool_exists(pool):
                     return True
         except Exception:
@@ -116,8 +135,8 @@ def pool_exists(self, pool) -> bool:
 
     def service_daemon_register(self, cluster, metadata):
         try:
-            if cluster: # rados client 
-                daemon_name = metadata['id']
+            if cluster:  # rados client
+                daemon_name = metadata["id"]
                 cluster.service_daemon_register("nvmeof", daemon_name, metadata)
                 self.logger.info(f"Registered {daemon_name} to service_map!")
         except Exception:
@@ -128,46 +147,62 @@ def service_daemon_update(self, cluster, status_buffer):
             if cluster and status_buffer:
                 cluster.service_daemon_update(status_buffer)
         except Exception:
-            self.logger.exception(f"Can't update daemon status to service_map!") 
+            self.logger.exception(f"Can't update daemon status to service_map!")
 
     def create_image(self, pool_name, image_name, size) -> bool:
         # Check for pool existence in advance as we don't create it if it's not there
         if not self.pool_exists(pool_name):
-            raise rbd.ImageNotFound(f"Pool {pool_name} doesn't exist", errno = errno.ENODEV)
+            raise rbd.ImageNotFound(
+                f"Pool {pool_name} doesn't exist", errno=errno.ENODEV
+            )
 
         image_exists = False
         try:
             image_size = self.get_image_size(pool_name, image_name)
             image_exists = True
         except rbd.ImageNotFound:
-            self.logger.debug(f"Image {pool_name}/{image_name} doesn't exist, will create it using size {size}")
+            self.logger.debug(
+                f"Image {pool_name}/{image_name} doesn't exist, will create it using size {size}"
+            )
             pass
 
         if image_exists:
             if image_size != size:
-                raise rbd.ImageExists(f"Image {pool_name}/{image_name} already exists with a size of {image_size} bytes which differs from the requested size of {size} bytes",
-                                      errno = errno.EEXIST)
-            return False    # Image exists with an idetical size, there is nothing to do here
+                raise rbd.ImageExists(
+                    f"Image {pool_name}/{image_name} already exists with a size of {image_size} bytes which differs from the requested size of {size} bytes",
+                    errno=errno.EEXIST,
+                )
+            return (
+                False  # Image exists with an idetical size, there is nothing to do here
+            )
 
         with rados.Rados(conffile=self.ceph_conf, rados_id=self.rados_id) as cluster:
             with cluster.open_ioctx(pool_name) as ioctx:
                 rbd_inst = rbd.RBD()
                 try:
                     rbd_inst.create(ioctx, image_name, size)
                 except rbd.ImageExists as ex:
-                    self.logger.exception(f"Image {pool_name}/{image_name} was created just now")
-                    raise rbd.ImageExists(f"Image {pool_name}/{image_name} was just created by someone else, please retry",
-                                          errno = errno.EAGAIN)
+                    self.logger.exception(
+                        f"Image {pool_name}/{image_name} was created just now"
+                    )
+                    raise rbd.ImageExists(
+                        f"Image {pool_name}/{image_name} was just created by someone else, please retry",
+                        errno=errno.EAGAIN,
+                    )
                 except Exception as ex:
-                    self.logger.exception(f"Can't create image {pool_name}/{image_name}")
+                    self.logger.exception(
+                        f"Can't create image {pool_name}/{image_name}"
+                    )
                     raise ex
 
         return True
 
     def get_image_size(self, pool_name, image_name) -> int:
         image_size = 0
         if not self.pool_exists(pool_name):
-            raise rbd.ImageNotFound(f"Pool {pool_name} doesn't exist", errno = errno.ENODEV)
+            raise rbd.ImageNotFound(
+                f"Pool {pool_name} doesn't exist", errno=errno.ENODEV
+            )
 
         with rados.Rados(conffile=self.ceph_conf, rados_id=self.rados_id) as cluster:
             with cluster.open_ioctx(pool_name) as ioctx:
@@ -176,9 +211,14 @@ def get_image_size(self, pool_name, image_name) -> int:
                     with rbd.Image(ioctx, image_name) as img:
                         image_size = img.size()
                 except rbd.ImageNotFound:
-                    raise rbd.ImageNotFound(f"Image {pool_name}/{image_name} doesn't exist", errno = errno.ENODEV)
+                    raise rbd.ImageNotFound(
+                        f"Image {pool_name}/{image_name} doesn't exist",
+                        errno=errno.ENODEV,
+                    )
                 except Exception as ex:
-                    self.logger.exception(f"Error while trying to get the size of image {pool_name}/{image_name}")
+                    self.logger.exception(
+                        f"Error while trying to get the size of image {pool_name}/{image_name}"
+                    )
                     raise ex
 
         return image_size