sdks/python: address gemini feedback

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: mohamedawnallah

sdks/python/apache_beam/transforms/enrichment_handlers/cloudsql.py

@@ -63,9 +63,8 @@ class TableFieldsQueryConfig:
   def __post_init__(self):
     if not self.table_id or not self.where_clause_template:
       raise ValueError(
-          "TableFieldsQueryConfig and " +
-          "TableFunctionQueryConfig must provide table_id " +
-          "and where_clause_template")
+          "TableFieldsQueryConfig must provide table_id and " +
+          "where_clause_template")
 
     if not self.where_clause_fields:
       raise ValueError(
@@ -83,9 +82,8 @@ class TableFunctionQueryConfig:
   def __post_init__(self):
     if not self.table_id or not self.where_clause_template:
       raise ValueError(
-          "TableFieldsQueryConfig and " +
-          "TableFunctionQueryConfig must provide table_id " +
-          "and where_clause_template")
+          "TableFunctionQueryConfig must provide table_id and " +
+          "where_clause_template")
 
     if not self.where_clause_value_fn:
       raise ValueError(
@@ -264,7 +262,7 @@ def __init__(
       connection_config = CloudSQLConnectionConfig(
         db_adapter=DatabaseTypeAdapter.POSTGRESQL,
         instance_connection_uri="apache-beam-testing:us-central1:itests",
-        user=postgres,
+        user='postgres',
         password= os.getenv("CLOUDSQL_PG_PASSWORD"))
       query_config=TableFieldsQueryConfig('my_table',"id = '{}'",['id']),
       cloudsql_handler = CloudSQLEnrichmentHandler(
@@ -319,6 +317,7 @@ def __enter__(self):
         url=self._connection_config.get_db_url(), creator=connector)
 
   def _execute_query(self, query: str, is_batch: bool, **params):
+    connection = None
     try:
       connection = self._engine.connect()
       transaction = connection.begin()
@@ -328,7 +327,8 @@ def _execute_query(self, query: str, is_batch: bool, **params):
         if is_batch:
           data = [row._asdict() for row in result]
         else:
-          data = result.first()._asdict()
+          result_row = result.first()
+          data = result_row._asdict() if result_row else {}
         # Explicitly commit the transaction.
         transaction.commit()
         return data
@@ -337,8 +337,8 @@ def _execute_query(self, query: str, is_batch: bool, **params):
         raise RuntimeError(f"Database operation failed: {e}")
     except Exception as e:
       raise Exception(
-          f'Could not execute the query: {query}. Please check if '
-          f'the query is properly formatted and the table exists. {e}')
+          f'Could not execute the query. Please check if the query is properly '
+          f'formatted and the table exists. {e}')
     finally:
       if connection:
         connection.close()
@@ -454,8 +454,8 @@ def get_cache_key(self, request: Union[beam.Row, list[beam.Row]]):
                 req_dict[field]
                 for field in self._query_config.where_clause_fields
             ]
-          key = ";".join(["%s"] * len(current_values))
-          cache_keys.extend([key % tuple(current_values)])
+          key = ';'.join(map(repr, current_values))
+          cache_keys.append(key)
         except KeyError as e:
           raise KeyError(
               "Make sure the values passed in `where_clause_fields` are the "

sdks/python/apache_beam/transforms/enrichment_handlers/cloudsql_it_test.py

@@ -103,7 +103,6 @@ def start_sql_db_container(
           sql_db_container.start()
           host = sql_db_container.get_container_host_ip()
           port = int(sql_db_container.get_exposed_port(5432))
-
         elif database_type == DatabaseTypeAdapter.MYSQL:
           user, password, db_id = "test", "test", "test"
           sql_db_container = MySqlContainer(
@@ -115,7 +114,6 @@ def start_sql_db_container(
           sql_db_container.start()
           host = sql_db_container.get_container_host_ip()
           port = int(sql_db_container.get_exposed_port(3306))
-
         elif database_type == DatabaseTypeAdapter.SQLSERVER:
           user, password, db_id = "SA", "A_Str0ng_Required_Password", "tempdb"
           sql_db_container = SqlServerContainer(
@@ -457,7 +455,10 @@ def test_sql_enrichment_on_non_existent_table(self):
       with TestPipeline() as p:
         _ = (p | beam.Create(requests) | Enrichment(handler))
 
-    expect_err_msg_contains = "Could not execute the query"
+    expect_err_msg_contains = (
+      "Could not execute the query. Please check if the query is properly "
+      "formatted and the table exists."
+    ) 
     self.assertIn(expect_err_msg_contains, str(context.exception))
 
   @pytest.mark.usefixtures("cache_container")
@@ -498,10 +499,12 @@ def test_sql_enrichment_with_redis(self):
       if not response:
         raise ValueError("No cache entry found for %s" % key)
 
-    # Mock the CloudSQL enrichment handler to avoid actual database calls.
-    # This simulates a cache hit scenario by returning predefined data.
+    # Mocks the CloudSQL enrichment handler to prevent actual database calls.
+    # This ensures that a cache hit scenario does not trigger any database
+    # interaction, raising an exception if an unexpected call occurs.
     actual = CloudSQLEnrichmentHandler.__call__
-    CloudSQLEnrichmentHandler.__call__ = MagicMock(return_value=(beam.Row()))
+    CloudSQLEnrichmentHandler.__call__ = MagicMock(
+      side_effect=Exception("Database should not be called on a cache hit."))
 
     # Run a second pipeline to verify cache is being used.
     with TestPipeline(is_integration_test=True) as test_pipeline:
@@ -542,7 +545,6 @@ def setUpClass(cls):
   @classmethod
   def tearDownClass(cls):
     super().tearDownClass()
-    cls._db = None
 
 
 @unittest.skipUnless(