black run

yanet-announcer.py

@@ -34,7 +34,10 @@ def skip_function(return_value: typing.Any = None):
         def decorator(func: typing.Callable):
             @functools.wraps(func)
             def wrapper(*args, **kwargs):
-                if func.__name__ in OPTIONS.skip or SKIP_CHECKS_ALL_KEYWORD in OPTIONS.skip:
+                if (
+                    func.__name__ in OPTIONS.skip
+                    or SKIP_CHECKS_ALL_KEYWORD in OPTIONS.skip
+                ):
                     LOGGER.debug("skip func execution: %s", func.__name__)
                     return return_value
                 return func(*args, **kwargs)
@@ -71,7 +74,9 @@ def get(command: str) -> typing.List[typing.Dict[str, str]]:
         if len(out) <= 1:
             return parsed_output
 
-        column_lengths: typing.List[int] = [len(column) for column in out[1].split("  ")]
+        column_lengths: typing.List[int] = [
+            len(column) for column in out[1].split("  ")
+        ]
 
         offset = 0
         headers = []
@@ -84,7 +89,9 @@ def get(command: str) -> typing.List[typing.Dict[str, str]]:
             columns: typing.Dict[str, str] = {}
 
             for i in range(0, len(column_lengths)):
-                columns[headers[i]] = out[row_i][offset : offset + column_lengths[i]].strip()
+                columns[headers[i]] = out[row_i][
+                    offset : offset + column_lengths[i]
+                ].strip()
                 offset += column_lengths[i] + 2
             parsed_output.append(columns)
 
@@ -156,6 +163,7 @@ def bgp_update(prefix_list):
         except Exception as error:
             LOGGER.error("Can not update bgp prefix: %s with error: %s", prefix, error)
 
+
 def bgp_remove(prefix_list):
     for prefix in prefix_list:
         try:
@@ -166,6 +174,7 @@ def bgp_remove(prefix_list):
         except Exception as error:
             LOGGER.error("Can not remove bgp prefix: %s with error: %s", prefix, error)
 
+
 def get_announces(types):
     for type in types:
         table_decap = Executer.get(f"yanet-cli {type}")
@@ -185,7 +194,12 @@ def get_announces(types):
 
                 announces.extend(table_decap_announce_row["announces"].split(","))
 
-            yield {"module": module, "type": type, "announces": announces, "next_module": next_module}
+            yield {
+                "module": module,
+                "type": type,
+                "announces": announces,
+                "next_module": next_module,
+            }
 
 
 @Decorator.logger_function
@@ -408,7 +422,9 @@ def init_logger():
     LOGGER = logging.getLogger(__name__)
     LOGGER.setLevel(logging.INFO)
 
-    formatter = logging.Formatter("%(filename)s:%(lineno)s - %(levelname)s - %(message)s")
+    formatter = logging.Formatter(
+        "%(filename)s:%(lineno)s - %(levelname)s - %(message)s"
+    )
     handler = logging.StreamHandler()
     handler.setFormatter(formatter)
 
@@ -418,13 +434,25 @@ def init_logger():
 def parse_args():
     global OPTIONS
 
-    parser = argparse.ArgumentParser(description="YANET announcer", formatter_class=argparse.RawTextHelpFormatter)
+    parser = argparse.ArgumentParser(
+        description="YANET announcer", formatter_class=argparse.RawTextHelpFormatter
+    )
     run_mode_group = parser.add_mutually_exclusive_group(required=True)
     run_mode_group.add_argument(
-        "-r", "--run", action="store_true", default=False, dest="daemon", help="run as a daemon"
+        "-r",
+        "--run",
+        action="store_true",
+        default=False,
+        dest="daemon",
+        help="run as a daemon",
     )
     run_mode_group.add_argument(
-        "-t", "--test", action="store_true", default=False, dest="dry_run", help="dry-run one time execution"
+        "-t",
+        "--test",
+        action="store_true",
+        default=False,
+        dest="dry_run",
+        help="dry-run one time execution",
     )
     parser.add_argument(
         "-s",
@@ -451,7 +479,9 @@ def update_config():
     # Use skip checks for skip flag rewrite opts.
     # "pop" using for back compatibility with previous format,
     #   where ANNOUNCER_CONFIG contains only prefixes
-    config_skip_checks: typing.Iterable[str] = ANNOUNCER_CONFIG.pop(SKIP_CHECKS_CONFIG_PARAM, [])
+    config_skip_checks: typing.Iterable[str] = ANNOUNCER_CONFIG.pop(
+        SKIP_CHECKS_CONFIG_PARAM, []
+    )
     if config_skip_checks and isinstance(config_skip_checks, abc.Iterable):
         OPTIONS.skip = config_skip_checks
 
@@ -491,7 +521,16 @@ def main():
             continue
 
         try:
-            for module in get_announces(["decap", "nat64stateless", "dregress", "balancer", "tun64", "nat64stateful"]):
+            for module in get_announces(
+                [
+                    "decap",
+                    "nat64stateless",
+                    "dregress",
+                    "balancer",
+                    "tun64",
+                    "nat64stateful",
+                ]
+            ):
                 if OPTIONS.dry_run:
                     LOGGER.info(module)
 
@@ -504,7 +543,8 @@ def main():
                 report_getannounces_counter = 1
             if len(current_prefixes) > 0:
                 LOGGER.warning(
-                    "Problem with get_announce(dp/cp in down state?), remove current announces: %s", current_prefixes
+                    "Problem with get_announce(dp/cp in down state?), remove current announces: %s",
+                    current_prefixes,
                 )
                 bgp_remove(current_prefixes)