feat: implement validate API key check endpoint and GCS verification

[yiyche]

mainly added

• api router file: pipeline/workflow/ingestion-helper/routes/ingestion.py (POST /ingestion/start)
• api key validation: tests DC_API_KEY against https://api.datacommons.org/v2/node before job launch to fail fast on 401/403.
• aditional GCS artifact verification: ensure bucket + file combo exists
• check for already running ingestion jobs (comment on PR under DCP repo)

[codacy-production]

Not up to standards ⛔

🔴 Issues 1 high · 1 minor

Alerts:
⚠ 2 issues (≤ 0 issues of at least minor severity)

Results:
2 new issues

Category	Results
ErrorProne	1 high
CodeStyle	1 minor

View in Codacy

🟢 Metrics 62 complexity

Metric	Results
Complexity	62

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.}

[gemini-code-assist]

Code Review

This pull request introduces a new ingestion router with /start and /config endpoints to validate, trigger, and inspect Cloud Run ingestion jobs. It also adds GCS bucket and file existence checks to the storage client, updates dependencies, and includes comprehensive unit tests. The review feedback focuses on improving robustness and error handling: guarding against potential AttributeErrors when auto-discovering job configurations, logging warnings for failed Cloud Run API requests, and updating error messages to account for permission issues during GCS bucket and file checks.

[gemini-code-assist]

When auto-discovering the input directory from the Cloud Run job, env_var.get('value') can be None if the environment variable is defined using valueFrom (e.g., referencing a secret or configmap). Guard against None values before calling .removeprefix() to prevent an AttributeError. Additionally, if the API call to fetch job details returns a non-200 status code, logging a warning with the status code and response text will make debugging configuration or permission issues much easier.

Suggested change

	if job_resp.status_code == 200:
	job_data = job_resp.json()
	containers = job_data.get("template", {}).get("template", {}).get("containers", [])
	if containers:
	env_vars = containers[0].get("env", [])
	for env_var in env_vars:
	if env_var.get("name") == "GCS_INPUT_FOLDER":
	input_dir = env_var.get("value")
	break
	elif env_var.get("name") == "INPUT_DIR":
	parts = env_var.get("value").removeprefix("gs://").split("/", 1)
	if len(parts) > 1:
	input_dir = parts[1]
	break
	if job_resp.status_code == 200:
	job_data = job_resp.json()
	containers = job_data.get("template", {}).get("template", {}).get("containers", [])
	if containers:
	env_vars = containers[0].get("env", [])
	for env_var in env_vars:
	if env_var.get("name") == "GCS_INPUT_FOLDER":
	input_dir = env_var.get("value")
	break
	elif env_var.get("name") == "INPUT_DIR":
	val = env_var.get("value")
	if val:
	parts = val.removeprefix("gs://").split("/", 1)
	if len(parts) > 1:
	input_dir = parts[1]
	break
	else:
	logging.warning(f"Failed to fetch Cloud Run job details for auto-discovery (HTTP {job_resp.status_code}): {job_resp.text}")

[gemini-code-assist]

If the API call to query active executions returns a non-200 status code (e.g., due to permission or configuration issues), it currently fails silently. Logging a warning with the status code and response text will help operators diagnose why the execution check failed.

Suggested change

	if exec_resp.status_code == 200:
	executions_data = exec_resp.json()
	for execution in executions_data.get("executions", []):
	# If completionTime is missing, the execution is actively running or pending
	if "completionTime" not in execution:
	raise HTTPException(
	status_code=409,
	detail=f"An ingestion job execution is already running: {execution.get('name')}"
	)
	if exec_resp.status_code == 200:
	executions_data = exec_resp.json()
	for execution in executions_data.get("executions", []):
	# If completionTime is missing, the execution is actively running or pending
	if "completionTime" not in execution:
	raise HTTPException(
	status_code=409,
	detail=f"An ingestion job execution is already running: {execution.get('name')}"
	)
	else:
	logging.warning(f"Failed to query active executions (HTTP {exec_resp.status_code}): {exec_resp.text}")

[gemini-code-assist]

If the GCS bucket exists but the service account lacks permission to access it, check_bucket_exists will catch the exception and return False. Raising an HTTPException stating that the bucket 'does not exist' is misleading in this case. Update the error message to indicate that the bucket may either not exist or not be accessible.

Suggested change

	if not storage.check_bucket_exists(bucket_name):
	raise HTTPException(
	status_code=400,
	detail=f"GCS bucket '{bucket_name}' does not exist."
	)
	if not storage.check_bucket_exists(bucket_name):
	raise HTTPException(
	status_code=400,
	detail=f"GCS bucket '{bucket_name}' does not exist or is not accessible."
	)