Fix: Resolve pylint issues in db_conn_manager.py

- Remove unused Optional import from typing
- Remove duplicate error logging (5 instances)
- Fix return type annotation for bind_db_app() method
- Split all long lines to comply with 79 character limit (20+ instances)
- Improve exception handling with specific exceptions
- Fix typo: 'doesn't exists' -> 'doesn't exist'
- Extract variables to reduce line length in display_pool_info

Author: Kamal Sai Devarapalli

common/pyportal_common/db_handlers/db_conn_manager.py

@@ -2,7 +2,7 @@
 
 import sys
 import time
-from typing import Union, Optional
+from typing import Union
 import flask
 import sqlalchemy.exc
 from flask_sqlalchemy import SQLAlchemy
@@ -18,7 +18,8 @@ class DataBaseConnectionHandler:
 
     def __init__(self, **kwargs) -> None:
         """
-        Initializes the DataBaseHandler object with database connection parameters.
+        Initialize the DataBaseHandler object with database connection
+        parameters.
 
         Args:
             logger_instance: Logger instance for logging.
@@ -64,14 +65,20 @@ def _prepare_database_uri(self) -> Union[str, None]:
                 return self._database_uri
             else:
                 self._database_uri = (
-                    f"{self.db_driver}://{self.db_user}:{quote_plus(self._db_password)}@{self._db_ip_address}:{str(self.db_port)}"
+                    f"{self.db_driver}://{self.db_user}:"
+                    f"{quote_plus(self._db_password)}@"
+                    f"{self._db_ip_address}:{str(self.db_port)}"
                     f"/{self.db_name}"
                 )
-                self.cmn_logger.info(f"Prepared DatabaseURI for Service :: [{self.cmn_logger.name}] - {self._database_uri}")
+                self.cmn_logger.info(
+                    f"Prepared DatabaseURI for Service :: "
+                    f"[{self.cmn_logger.name}] - {self._database_uri}"
+                )
                 return self._database_uri
-        except Exception as ex:
+        except (ValueError, AttributeError) as ex:
             self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                f"Error occurred :: {ex}\tLine No:: "
+                f"{sys.exc_info()[2].tb_lineno}"
             )
             return self._database_uri
 
@@ -83,22 +90,24 @@ def create_db_engine_for_service(
             while _retries < self._max_retries:
                 try:
                     self.cmn_logger.info(
-                        f"Creating DB-Engine using DatabaseURI for Service :: [{self.cmn_logger.name}]"
+                        f"Creating DB-Engine using DatabaseURI for "
+                        f"Service :: [{self.cmn_logger.name}]"
+                    )
+                    app_instance.config["SQLALCHEMY_DATABASE_URI"] = (
+                        self._prepare_database_uri()
                     )
-                    app_instance.config["SQLALCHEMY_DATABASE_URI"] = self._prepare_database_uri()
                     db_engine: Engine = create_engine(
                         app_instance.config["SQLALCHEMY_DATABASE_URI"]
                     )
                     self.cmn_logger.info(
-                        f"Created DB-Engine using DatabaseURI for Service :: [{db_engine}]"
+                        f"Created DB-Engine using DatabaseURI for "
+                        f"Service :: [{db_engine}]"
                     )
                     return db_engine, True
-                except Exception as ex:
-                    self.cmn_logger.error(
-                        f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
-                    )
+                except (sqlalchemy.exc.SQLAlchemyError, ValueError) as ex:
                     self.cmn_logger.error(
-                        f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                        f"Error occurred :: {ex}\tLine No:: "
+                        f"{sys.exc_info()[2].tb_lineno}"
                     )
                     _retries += 1
                     if _retries < self._max_retries:
@@ -108,160 +117,173 @@ def create_db_engine_for_service(
                         time.sleep(self._retry_interval)
                     else:
                         return None, False
-        except Exception as ex:
+        except (sqlalchemy.exc.SQLAlchemyError, ValueError) as ex:
             self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                f"Error occurred :: {ex}\tLine No:: "
+                f"{sys.exc_info()[2].tb_lineno}"
             )
             return None, False
 
     def create_database_for_service(self) -> Union[bool, None]:
         try:
             if not database_exists(self._database_uri):
                 self.cmn_logger.info(
-                    f"Database doesn't exists :: {self._database_uri.split('/')[-1]}"
+                    f"Database doesn't exist :: "
+                    f"{self._database_uri.split('/')[-1]}"
                 )
                 create_database(self._database_uri)
+                db_name = self._database_uri.split('/')[-1]
                 self.cmn_logger.info(
-                    f"Created the database :: {self._database_uri.split('/')[-1]}"
+                    f"Created the database :: {db_name}"
                 )
             else:
+                db_name = self._database_uri.split('/')[-1]
                 self.cmn_logger.info(
-                    f"Database already exists :: {self._database_uri.split('/')[-1]}"
+                    f"Database already exists :: {db_name}"
                 )
             return True
-        except Exception as ex:
-            self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
-            )
+        except (sqlalchemy.exc.SQLAlchemyError, ValueError) as ex:
             self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                f"Error occurred :: {ex}\tLine No:: "
+                f"{sys.exc_info()[2].tb_lineno}"
             )
 
-    def bind_db_app(self, app_instance: flask.Flask) -> Union[SQLAlchemy, None]:
+    def bind_db_app(self, app_instance: flask.Flask) -> Union[bool, None]:
         try:
             self.cmn_logger.info(
-                f"Binding SQLALCHEMY  to Application Instance for Service :: [{self.cmn_logger.name}]"
+                f"Binding SQLALCHEMY to Application Instance for "
+                f"Service :: [{self.cmn_logger.name}]"
             )
             app_db = SQLAlchemy(app=app_instance)
             self.cmn_logger.info(
-                f"Bound SQLALCHEMY :: [{app_db}] to Application Instance for Service :: [{self.cmn_logger.name}]"
+                f"Bound SQLALCHEMY :: [{app_db}] to Application "
+                f"Instance for Service :: [{self.cmn_logger.name}]"
             )
             return True
-        except Exception as ex:
-            self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
-            )
+        except (ValueError, AttributeError, RuntimeError) as ex:
             self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                f"Error occurred :: {ex}\tLine No:: "
+                f"{sys.exc_info()[2].tb_lineno}"
             )
 
     def migrate_db_bind_app(
         self, app_instance: flask.Flask, app_db: SQLAlchemy
     ) -> Union[Migrate, None]:
         try:
             self.cmn_logger.info(
-                f"Binding the db :: [{app_db}] to app instance :: [{app_instance}] for migrations"
+                f"Binding the db :: [{app_db}] to app instance :: "
+                f"[{app_instance}] for migrations"
             )
             migrate_instance = Migrate(app_instance, app_db)
             self.cmn_logger.info(
-                f"Bound the db :: [{app_db}] to app instance :: [{app_instance}] for migrations"
+                f"Bound the db :: [{app_db}] to app instance :: "
+                f"[{app_instance}] for migrations"
             )
             return migrate_instance
-        except Exception as ex:
-            self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
-            )
+        except (ValueError, AttributeError, RuntimeError) as ex:
             self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                f"Error occurred :: {ex}\tLine No:: "
+                f"{sys.exc_info()[2].tb_lineno}"
             )
 
     def display_pool_info(self, connection_pool: QueuePool) -> None:
         try:
             if connection_pool:
                 self.cmn_logger.info(
-                    "#---------------------------------[ POOL INFO ]----------------------------------------#"
+                    "#---------------------------------[ POOL INFO ]"
+                    "----------------------------------------#"
                 )
                 self.cmn_logger.info(
-                    f"Displaying Pool Info for Service ==>[{self.cmn_logger.name}]"
+                    f"Displaying Pool Info for Service "
+                    f"==>[{self.cmn_logger.name}]"
                 )
+                pool_id = id(connection_pool)
                 self.cmn_logger.info(
-                    f"Current Pool Info :: {connection_pool} - ID: {id(connection_pool)}"
+                    f"Current Pool Info :: {connection_pool} - ID: {pool_id}"
                 )
                 self.cmn_logger.info(
                     f"Current Pool Size :: {connection_pool.size()}"
                 )
+                checked_in = connection_pool.checkedin()
                 self.cmn_logger.info(
-                    f"Checked Out Connections from Pool: {connection_pool.checkedin()}"
+                    f"Checked Out Connections from Pool: {checked_in}"
                 )
+                checked_out = connection_pool.checkedout()
                 self.cmn_logger.info(
-                    f"Checked in Connections available in Pool: {connection_pool.checkedout()}"
+                    f"Checked in Connections available in Pool: {checked_out}"
                 )
                 self.cmn_logger.info(
                     f"Current Pool Overflow Info: {connection_pool.overflow()}"
                 )
                 self.cmn_logger.info(
-                    "#---------------------------------[ POOL INFO ]----------------------------------------#"
+                    "#---------------------------------[ POOL INFO ]"
+                    "----------------------------------------#"
                 )
-        except Exception as ex:
-            self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
-            )
+        except (AttributeError, RuntimeError) as ex:
             self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                f"Error occurred :: {ex}\tLine No:: "
+                f"{sys.exc_info()[2].tb_lineno}"
             )
 
     def check_db_connectivity_and_retry(
         self, db_engine: SQLAlchemy.engine
     ) -> Union[bool, None]:
         try:
-            # Establishing the connection to the database and create a database/table if not exists
+            # Establishing the connection to the database and create a
+            # database/table if not exists
             while True:
-                db_connection_status = self._check_database_connectivity(db_engine)
+                db_connection_status = (
+                    self._check_database_connectivity(db_engine)
+                )
                 if db_connection_status:
                     break
                 else:
                     self.cmn_logger.info(
-                        f"Going for retry .. RETRY_INTERVAL :: {self._retry_interval} sec"
+                        f"Going for retry .. RETRY_INTERVAL :: "
+                        f"{self._retry_interval} sec"
                     )
                     time.sleep(self._retry_interval)
             return db_connection_status
-        except Exception as ex:
-            self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
-            )
+        except (sqlalchemy.exc.SQLAlchemyError, ValueError) as ex:
             self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                f"Error occurred :: {ex}\tLine No:: "
+                f"{sys.exc_info()[2].tb_lineno}"
             )
 
     def _check_database_connectivity(
         self, db_engine: SQLAlchemy.engine
     ) -> Union[bool, None]:
         db_connection_status = False
         self.cmn_logger.info(
-            f"Current Connection status set to :: {db_connection_status} for the service :: {self.cmn_logger.name}"
+            f"Current Connection status set to :: {db_connection_status} "
+            f"for the service :: {self.cmn_logger.name}"
         )
         try:
             self.cmn_logger.info(
-                "Trying to establish the connection to the database :: [IN-PROGRESS]"
+                "Trying to establish the connection to the database "
+                ":: [IN-PROGRESS]"
             )
             db_connection = db_engine.connect()
             self.cmn_logger.info(
                 "Established the connection to the database :: [SUCCESS]"
             )
             db_connection_status = True
             self.cmn_logger.info(
-                f"Current Connection status set to :: {db_connection_status} for the service :: {self.cmn_logger.name}"
+                f"Current Connection status set to :: {db_connection_status} "
+                f"for the service :: {self.cmn_logger.name}"
             )
             db_connection.close()
             self.cmn_logger.info(
-                f"Closing the Current Connection as the connection was established for the service :: {self.cmn_logger.name}"
+                f"Closing the Current Connection as the connection was "
+                f"established for the service :: {self.cmn_logger.name}"
             )
         except sqlalchemy.exc.OperationalError as ex:
             self.cmn_logger.error(
                 f"Current Connection status set to :: {db_connection_status}"
             )
             self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                f"Error occurred :: {ex}\tLine No:: "
+                f"{sys.exc_info()[2].tb_lineno}"
             )
         return db_connection_status
 
@@ -271,7 +293,8 @@ def create_tables_associated_to_db_model(
         connection_status = False
         try:
             self.cmn_logger.info(
-                "Trying to create the tables if not present in the database..."
+                "Trying to create the tables if not present in the "
+                "database..."
             )
             # Create all the tables associated with the Base class
             self.cmn_logger.info("Going to create the tables ...")
@@ -280,14 +303,17 @@ def create_tables_associated_to_db_model(
             self.cmn_logger.info("Created the tables ...")
         except sqlalchemy.exc.OperationalError as ex:
             self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                f"Error occurred :: {ex}\tLine No:: "
+                f"{sys.exc_info()[2].tb_lineno}"
             )
         except sqlalchemy.exc.TimeoutError as ex:
             self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                f"Error occurred :: {ex}\tLine No:: "
+                f"{sys.exc_info()[2].tb_lineno}"
             )
-        except Exception as ex:
+        except (sqlalchemy.exc.SQLAlchemyError, ValueError) as ex:
             self.cmn_logger.error(
-                f"Error occurred :: {ex}\tLine No:: {sys.exc_info()[2].tb_lineno}"
+                f"Error occurred :: {ex}\tLine No:: "
+                f"{sys.exc_info()[2].tb_lineno}"
             )
         return connection_status