replace f-strings with parametrized queries to avoid codacy error

Author: crong12

ptbcontrib/postgres_persistence/postgrespersistence.py

@@ -144,28 +144,29 @@ def __init_database(self) -> None:
             # If column exists, check if there's a valid row with id = 1
             data_valid = False
             if column_valid:
-                check_data_qry = f"""
-                    SELECT 1 FROM persistence WHERE id = {self.PERSISTENCE_ID};"""
-                data_valid = self._session.execute(text(check_data_qry)).first() is not None
+                check_data_qry = """
+                    SELECT 1 FROM persistence WHERE id = :id;"""
+                result = self._session.execute(text(check_data_qry), {"id": self.PERSISTENCE_ID})
+                data_valid = result.first() is not None
 
             needs_migration = not (column_valid and data_valid)
 
             if needs_migration:
                 self.logger.info("Old database schema detected. Running migration...")
                 migration_commands = [
                     "ALTER TABLE persistence ADD COLUMN id INT;",
-                    f"""UPDATE persistence SET id = {self.PERSISTENCE_ID} WHERE ctid = (
-                        SELECT ctid FROM persistence LIMIT 1);""",
+                    """UPDATE persistence SET id = :id WHERE ctid = ("
+                        "SELECT ctid FROM persistence LIMIT 1);""",
                     "DELETE FROM persistence WHERE id IS NULL;",
                     "ALTER TABLE persistence ALTER COLUMN id SET NOT NULL;",
                     "ALTER TABLE persistence ADD PRIMARY KEY (id);",
-                    (
-                        f"ALTER TABLE persistence ADD CONSTRAINT single_row "
-                        f"CHECK (id = {self.PERSISTENCE_ID});"
-                    ),
+                    "ALTER TABLE persistence ADD CONSTRAINT single_row CHECK (id = :id);",
                 ]
                 for command in migration_commands:
-                    self._session.execute(text(command))
+                    if ":id" in command:
+                        self._session.execute(text(command), {"id": self.PERSISTENCE_ID})
+                    else:
+                        self._session.execute(text(command))
                 self.logger.info("Database migration successful!")
 
             self._session.commit()

tests/test_postgres_persistence.py

@@ -252,7 +252,7 @@ def mock_execute(query, params=None):
                 return FakeExecResultValidPK()
 
             # Check for data validation query (id=1 exists)
-            if "WHERE id = 1" in query.text and "information_schema" not in query.text:
+            if "WHERE id = :id" in query.text and "information_schema" not in query.text:
                 return FakeExecResultValidData()
 
             return FakeExecResult()
@@ -267,7 +267,7 @@ def mock_execute(query, params=None):
         # Verify no migration commands were run
         migration_commands = [
             "ALTER TABLE persistence ADD COLUMN id INT",
-            "UPDATE persistence SET id = 1",
+            "UPDATE persistence SET id = :id",
             "DELETE FROM persistence WHERE id IS NULL",
         ]
         for migration_cmd in migration_commands:
@@ -347,11 +347,11 @@ def mock_execute(query, params=None):  # pylint: disable=unused-argument
         # Verify migration commands were run in correct order
         expected_migration_steps = [
             "ALTER TABLE persistence ADD COLUMN id INT",
-            "UPDATE persistence SET id = 1",
+            "UPDATE persistence SET id = :id",
             "DELETE FROM persistence WHERE id IS NULL",
             "ALTER TABLE persistence ALTER COLUMN id SET NOT NULL",
             "ALTER TABLE persistence ADD PRIMARY KEY (id)",
-            "ALTER TABLE persistence ADD CONSTRAINT single_row CHECK (id = 1)",
+            "ALTER TABLE persistence ADD CONSTRAINT single_row CHECK (id = :id)",
         ]
 
         for expected_step in expected_migration_steps:
@@ -500,7 +500,7 @@ def mock_execute(query, params=None):  # pylint: disable=unused-argument
 
         # Verify that migration includes step to update first row to id=1
         update_first_row = any(
-            "UPDATE persistence SET id = 1" in query and "LIMIT 1" in query
+            "UPDATE persistence SET id = :id" in query and "LIMIT 1" in query
             for query in executed_queries
         )
         assert update_first_row