Refactor runner.py and utils.py for improved readability and consistency

- Standardized string formatting across command initialization in `runner.py` for better consistency.
- Enhanced the structure of the `DbReplicatorRunner` class by using multi-line arguments for improved readability.
- Updated logging messages in both `runner.py` and `utils.py` to use consistent string formatting.
- Improved import organization in `runner.py` and `utils.py` for better clarity and maintainability.
- Added helper methods in `base_replication_test.py` to streamline replication setup and target database creation in tests.

Author: jaredmdobson

mysql_ch_replicator/runner.py

@@ -1,67 +1,71 @@
 import os
-import time
 import sys
 import threading
-from uvicorn import Config, Server
-from fastapi import APIRouter, FastAPI
-
+import time
 from logging import getLogger
 
-from .config import Settings
-from .mysql_api import MySQLApi
-from .utils import ProcessRunner, GracefulKiller
+from fastapi import APIRouter, FastAPI
+from uvicorn import Config, Server
 
 from . import db_replicator
-
+from .config import Settings
+from .mysql_api import MySQLApi
+from .utils import GracefulKiller, ProcessRunner
 
 logger = getLogger(__name__)
 
 
-
 class BinlogReplicatorRunner(ProcessRunner):
     def __init__(self, config_file):
-        super().__init__(f'{sys.argv[0]} --config {config_file} binlog_replicator')
+        super().__init__(f"{sys.argv[0]} --config {config_file} binlog_replicator")
 
 
 class DbReplicatorRunner(ProcessRunner):
-    def __init__(self, db_name, config_file, worker_id=None, total_workers=None, initial_only=False):
-        cmd = f'{sys.argv[0]} --config {config_file} --db {db_name} db_replicator'
-        
+    def __init__(
+        self,
+        db_name,
+        config_file,
+        worker_id=None,
+        total_workers=None,
+        initial_only=False,
+    ):
+        cmd = f"{sys.argv[0]} --config {config_file} --db {db_name} db_replicator"
+
         if worker_id is not None:
-            cmd += f' --worker_id={worker_id}'
-        
+            cmd += f" --worker_id={worker_id}"
+
         if total_workers is not None:
-            cmd += f' --total_workers={total_workers}'
-        
+            cmd += f" --total_workers={total_workers}"
+
         if initial_only:
-            cmd += ' --initial_only=True'
-            
+            cmd += " --initial_only=True"
+
         super().__init__(cmd)
 
 
 class DbOptimizerRunner(ProcessRunner):
     def __init__(self, config_file):
-        super().__init__(f'{sys.argv[0]} --config {config_file} db_optimizer')
+        super().__init__(f"{sys.argv[0]} --config {config_file} db_optimizer")
 
 
 class RunAllRunner(ProcessRunner):
     def __init__(self, db_name, config_file):
-        super().__init__(f'{sys.argv[0]} --config {config_file} run_all --db {db_name}')
+        super().__init__(f"{sys.argv[0]} --config {config_file} run_all --db {db_name}")
 
 
 app = FastAPI()
 
 
-
 class Runner:
-
     DB_REPLICATOR_RUN_DELAY = 5
 
-    def __init__(self, config: Settings, wait_initial_replication: bool, databases: str):
+    def __init__(
+        self, config: Settings, wait_initial_replication: bool, databases: str
+    ):
         self.config = config
         self.databases = databases or config.databases
         self.wait_initial_replication = wait_initial_replication
-        self.runners: dict[str: DbReplicatorRunner] = {}
+        self.runners: dict[str, DbReplicatorRunner] = {}
         self.binlog_runner = None
         self.db_optimizer = None
         self.http_server = None
@@ -71,13 +75,15 @@ def __init__(self, config: Settings, wait_initial_replication: bool, databases:
 
     def run_server(self):
         if not self.config.http_host or not self.config.http_port:
-            logger.info('http server disabled')
+            logger.info("http server disabled")
             return
-        logger.info('starting http server')
+        logger.info("starting http server")
 
         config = Config(app=app, host=self.config.http_host, port=self.config.http_port)
         self.router = APIRouter()
-        self.router.add_api_route("/restart_replication", self.restart_replication, methods=["GET"])
+        self.router.add_api_route(
+            "/restart_replication", self.restart_replication, methods=["GET"]
+        )
         app.include_router(self.router)
 
         self.http_server = Server(config)
@@ -87,15 +93,15 @@ def restart_replication(self):
         self.replication_restarted = False
         self.need_restart_replication = True
         while not self.replication_restarted:
-            logger.info('waiting replication restarted..')
+            logger.info("waiting replication restarted..")
             time.sleep(1)
         return {"restarted": True}
 
     def is_initial_replication_finished(self, db_name):
         state_path = os.path.join(
             self.config.binlog_replicator.data_dir,
             db_name,
-            'state.pckl',
+            "state.pckl",
         )
         state = db_replicator.State(state_path)
         return state.status == db_replicator.Status.RUNNING_REALTIME_REPLICATION
@@ -111,43 +117,48 @@ def restart_dead_processes(self):
     def restart_replication_if_required(self):
         if not self.need_restart_replication:
             return
-        logger.info('restarting replication')
+        logger.info("restarting replication")
         for db_name, runner in self.runners.items():
-            logger.info(f'stopping runner {db_name}')
+            logger.info(f"stopping runner {db_name}")
             runner.stop()
-            path = os.path.join(self.config.binlog_replicator.data_dir, db_name, 'state.pckl')
+            path = os.path.join(
+                self.config.binlog_replicator.data_dir, db_name, "state.pckl"
+            )
             if os.path.exists(path):
-                logger.debug(f'removing {path}')
+                logger.debug(f"removing {path}")
                 os.remove(path)
 
-        logger.info('starting replication')
+        logger.info("starting replication")
         self.restart_dead_processes()
         self.need_restart_replication = False
         self.replication_restarted = True
 
     def check_databases_updated(self, mysql_api: MySQLApi):
-        logger.debug('check if databases were created / removed in mysql')
+        logger.debug("check if databases were created / removed in mysql")
         databases = mysql_api.get_databases()
-        logger.info(f'mysql databases: {databases}')
+        logger.info(f"mysql databases: {databases}")
         databases = [db for db in databases if self.config.is_database_matches(db)]
-        logger.info(f'mysql databases filtered: {databases}')
+        logger.info(f"mysql databases filtered: {databases}")
         for db in databases:
             if db in self.runners:
                 continue
-            logger.info(f'running replication for {db} (database created in mysql)')
-            runner = self.runners[db] = DbReplicatorRunner(db_name=db, config_file=self.config.settings_file)
+            logger.info(f"running replication for {db} (database created in mysql)")
+            runner = self.runners[db] = DbReplicatorRunner(
+                db_name=db, config_file=self.config.settings_file
+            )
             runner.run()
 
         for db in self.runners.keys():
             if db in databases:
                 continue
-            logger.info(f'stop replication for {db} (database removed from mysql)')
+            logger.info(f"stop replication for {db} (database removed from mysql)")
             self.runners[db].stop()
             self.runners.pop(db)
 
     def run(self):
         mysql_api = MySQLApi(
-            database=None, mysql_settings=self.config.mysql,
+            database=None,
+            mysql_settings=self.config.mysql,
         )
         databases = mysql_api.get_databases()
         databases = [db for db in databases if self.config.is_database_matches(db)]
@@ -173,8 +184,10 @@ def run(self):
                 break
             if not self.is_initial_replication_finished(db_name=db):
                 continue
-            logger.info(f'running replication for {db} (initial replication finished)')
-            runner = self.runners[db] = DbReplicatorRunner(db_name=db, config_file=self.config.settings_file)
+            logger.info(f"running replication for {db} (initial replication finished)")
+            runner = self.runners[db] = DbReplicatorRunner(
+                db_name=db, config_file=self.config.settings_file
+            )
             runner.run()
 
         # Second - run replication for other DBs one by one and wait until initial replication finished
@@ -184,44 +197,54 @@ def run(self):
             if killer.kill_now:
                 break
 
-            logger.info(f'running replication for {db} (initial replication not finished - waiting)')
-            runner = self.runners[db] = DbReplicatorRunner(db_name=db, config_file=self.config.settings_file)
+            logger.info(
+                f"running replication for {db} (initial replication not finished - waiting)"
+            )
+            runner = self.runners[db] = DbReplicatorRunner(
+                db_name=db, config_file=self.config.settings_file
+            )
             runner.run()
             if not self.wait_initial_replication:
                 continue
 
-            while not self.is_initial_replication_finished(db_name=db) and not killer.kill_now:
+            while (
+                not self.is_initial_replication_finished(db_name=db)
+                and not killer.kill_now
+            ):
                 time.sleep(1)
                 self.restart_dead_processes()
 
-        logger.info('all replicators launched')
+        logger.info("all replicators launched")
 
         last_check_db_updated = time.time()
         while not killer.kill_now:
             time.sleep(1)
             self.restart_replication_if_required()
             self.restart_dead_processes()
-            if time.time() - last_check_db_updated > self.config.check_db_updated_interval:
+            if (
+                time.time() - last_check_db_updated
+                > self.config.check_db_updated_interval
+            ):
                 self.check_databases_updated(mysql_api=mysql_api)
                 last_check_db_updated = time.time()
 
-        logger.info('stopping runner')
+        logger.info("stopping runner")
 
         if self.binlog_runner is not None:
-            logger.info('stopping binlog replication')
+            logger.info("stopping binlog replication")
             self.binlog_runner.stop()
 
         if self.db_optimizer is not None:
-            logger.info('stopping db_optimizer')
+            logger.info("stopping db_optimizer")
             self.db_optimizer.stop()
 
         for db_name, db_replication_runner in self.runners.items():
-            logger.info(f'stopping replication for {db_name}')
+            logger.info(f"stopping replication for {db_name}")
             db_replication_runner.stop()
 
         if self.http_server:
             self.http_server.should_exit = True
 
         server_thread.join()
 
-        logger.info('stopped')
+        logger.info("stopped")