Address reviewer comments

Author: vish-cs

import-automation/executor/app/configs.py

@@ -142,17 +142,18 @@ class ExecutorConfig:
     # Invoke import tool genmcf.
     invoke_import_tool: bool = True
     # Invoke differ tool.
-    invoke_differ_tool: bool = True 
+    invoke_differ_tool: bool = True
     # Invoke validations before upload.
-    invoke_import_validation: bool = True 
+    invoke_import_validation: bool = True
     # Ignore validation status during import.
     ignore_validation_status: bool = False
     # Import validation config file path (relative to data repo).
     validation_config_file: str = 'tools/import_validation/validation_config.json'
     # Latest import version (overwrite)
     import_version_override: str = ''
     # Relative path to version folder for graph files.
-    graph_data_paths: List[str] = dataclasses.field(default_factory=lambda: ['/*/*/*.mcf'])
+    graph_data_paths: List[str] = dataclasses.field(
+        default_factory=lambda: ['/*/*/*.mcf'])
     # Maximum time venv creation can take in seconds.
     venv_create_timeout: float = 3600
     # Maximum time downloading a file can take in seconds.

import-automation/workflow/import-helper/main.py

@@ -1,3 +1,17 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 import base64
 import functions_framework
 import json

import-automation/workflow/ingestion-helper/main.py

@@ -42,6 +42,7 @@
 if not FLAGS.is_parsed():
     FLAGS(['ingestion_helper'])
 
+
 def _validate_params(request_json, required_params):
     for param in required_params:
         if param not in request_json:
@@ -135,7 +136,7 @@ def ingestion_helper(request):
             return (validation_error, 400)
         import_name = request_json['importName']
         status = request_json['status']
-        logging.info(f'Updating {import_name} {status}')
+        logging.info(f'Updating import {import_name} to status {status}')
         params = import_utils.get_import_params(request_json)
         spanner.update_import_status(params)
         return (f"Updated import {import_name} to status {params['status']}",
@@ -153,18 +154,19 @@ def ingestion_helper(request):
         import_name = request_json['importName']
         version = request_json['version']
         comment = request_json['comment']
+        short_import_name = import_name.split(':')[-1]
         caller = import_utils.get_caller_identity(request)
         logging.info(
-            f"[ImportVersionAuditLog] Import {import_name} version {version} caller: {caller} comment: {comment}"
+            f"Import {short_import_name} version {version} caller: {caller} comment: {comment}"
         )
         if version == 'staging':
             version = storage.get_staging_version(import_name)
         summary = storage.get_import_summary(import_name, version)
         params = import_utils.create_import_params(summary)
         params['status'] = 'READY'
         storage.update_version_file(import_name, version)
-        spanner.update_version_history(import_name, version, comment)
+        spanner.update_version_history(import_name, version, caller, comment)
         spanner.update_import_status(params)
-        return (f'Updated import {import_name} to version {version}', 200)
+        return (f'Updated import {short_import_name} to version {version}', 200)
     else:
         return (f'Unknown actionType: {actionType}', 400)

import-automation/workflow/ingestion-helper/spanner_client.py

@@ -27,7 +27,6 @@ class SpannerClient:
     and getting/updating import statuses.
     """
     _LOCK_ID = "global_ingestion_lock"
-    _DEFAULT_GRAPH_PATH = "/*/*/*.mcf"
 
     def __init__(self, project_id: str, instance_id: str, database_id: str):
         """Initializes a Spanner client and connects to a specific database."""
@@ -202,9 +201,9 @@ def update_import_status(self, params: dict):
         exec_time = params['exec_time']
         data_volume = params['data_volume']
         status = params['status']
-        version = params['version'].removesuffix(self._DEFAULT_GRAPH_PATH)
+        version = params['version']
         next_refresh = params['next_refresh']
-        graph_paths = params.get('graph_paths', [self._DEFAULT_GRAPH_PATH])
+        graph_paths = params['graph_paths']
         logging.info(f"Updating import status for {import_name} to {status}")
 
         def _record(transaction: Transaction):
@@ -237,7 +236,7 @@ def _record(transaction: Transaction):
             raise
 
     def update_version_history(self, import_name: str, version: str,
-                               comment: str):
+                               caller: str, comment: str):
         """Updates the version history table.
 
         Args:
@@ -248,8 +247,12 @@ def update_version_history(self, import_name: str, version: str,
         logging.info(f"Updating version history for {import_name} to {version}")
 
         def _record(transaction: Transaction):
-            columns = ["ImportName", "Version", "UpdateTimestamp", "Comment"]
-            values = [[import_name, version, spanner.COMMIT_TIMESTAMP, comment]]
+            columns = [
+                "ImportName", "Version", "UpdateTimestamp", "Caller", "Comment"
+            ]
+            values = [[
+                import_name, version, spanner.COMMIT_TIMESTAMP, caller, comment
+            ]]
             transaction.insert(table="ImportVersionHistory",
                                columns=columns,
                                values=values)

import-automation/workflow/ingestion-helper/storage_client.py

@@ -88,12 +88,12 @@ def update_version_file(self, import_name: str, version: str):
             output_dir = import_name.replace(':', '/')
             version_file = self.bucket.blob(
                 os.path.join(output_dir, 'latest_version.txt'))
-            version_file.upload_from_string(version)
             self.bucket.copy_blob(
                 self.bucket.blob(
                     os.path.join(output_dir, version,
                                  'import_metadata_mcf.mcf')), self.bucket,
                 os.path.join(output_dir, 'import_metadata_mcf.mcf'))
+            version_file.upload_from_string(version)
         except exceptions.NotFound as e:
             logging.error(f'Error updating version file for {import_name}: {e}')
             raise