[DCP Preprocessor] Move preprocessor to workflow - part 1

[gmechali]

This piece just reworks the output of the preprocessor to store the workflow's input in GCS so the workflow can propagate it to dataflow.

[gemini-code-assist]

Code Review

This pull request replaces the direct invocation of trigger_ingestion_workflow with a new _handle_workflow_handoff method in simple/stats/runner.py to write processed import metadata to a GCS handshake JSON file. Feedback suggests validating that GCP_SPANNER_INSTANCE_ID and GCP_SPANNER_DATABASE_NAME are set before writing the handshake payload to prevent downstream workflow steps from failing silently.

[gemini-code-assist]

When WORKFLOW_EXECUTION_ID is set, the preprocessor is running as part of the ingestion workflow. In this context, GCP_SPANNER_INSTANCE_ID and GCP_SPANNER_DATABASE_NAME are required environment variables. If they are missing, writing an incomplete handshake payload with empty strings will cause downstream workflow steps to fail silently or unexpectedly. It is safer to validate these variables and log an error/skip the handshake if they are not set.

    workflow_id = os.getenv("WORKFLOW_EXECUTION_ID")
    temp_location = os.getenv("TEMP_LOCATION", "")

    if not workflow_id or not temp_location.startswith("gs://"):
      logging.warning(
          "WORKFLOW_EXECUTION_ID not set or TEMP_LOCATION (%s) is not a GCS path. Skipping GCS handshake.",
          temp_location)
      return

    spanner_instance = os.getenv("GCP_SPANNER_INSTANCE_ID")
    spanner_database = os.getenv("GCP_SPANNER_DATABASE_NAME")
    if not spanner_instance or not spanner_database:
      logging.error(
          "GCP_SPANNER_INSTANCE_ID or GCP_SPANNER_DATABASE_NAME not set. Skipping GCS handshake.")
      return

    raw_import_name = "_".join(item["importName"] for item in self.trigger_workflow_info)
    sanitized_import_name = re.sub(r'[^a-zA-Z0-9_-]', '_', raw_import_name)[:64]

    handshake_payload = {
        "spannerInstanceId": spanner_instance,
        "spannerDatabaseId": spanner_database,
        "importName": sanitized_import_name,
        "importList": json.dumps(self.trigger_workflow_info),
        "tempLocation": temp_location,
        "region": os.getenv("REGION", "")
    }

[codacy-production]

Up to standards ✅

🟢 Issues 0 issues

Results:
0 new issues

View in Codacy

🟢 Metrics 0 complexity

Metric	Results
Complexity	0

View in Codacy

_{NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.}