add quick validation for the api key and add log if api key isn't valid

packages/datacommons-admin/datacommons_admin/admin_cli.py

@@ -12,23 +12,23 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from pathlib import Path
 import re
 import sys
 import urllib.request
+from pathlib import Path
 from typing import Any, Tuple
 
 import click
 from google.api_core import exceptions
 from google.cloud import storage
 
-from . import __version__
 from datacommons_admin.infra_templates import (
     BACKEND_TF_TEMPLATE,
     README_TEMPLATE,
     REMOTE_STATE_TEMPLATE,
 )
 
+from . import __version__
 
 DEFAULT_BUCKET_LOCATION = "US"
 GITHUB_RAW_BASE_URL = "https://raw.githubusercontent.com/datacommonsorg/datacommons"
@@ -95,7 +95,7 @@ def _create_and_configure_bucket(
     new_bucket.iam_configuration.uniform_bucket_level_access_enabled = True
     new_bucket.versioning_enabled = True
     new_bucket.patch()
-    click.secho(f"  Enabling versioning...", fg="bright_black")
+    click.secho("  Enabling versioning...", fg="bright_black")
     click.secho("  Configuring bucket IAM policy...", fg="bright_black")
     policy = new_bucket.get_iam_policy(requested_policy_version=3)
     policy["roles/storage.objectAdmin"].add(f"projectEditor:{project_id}")
@@ -145,8 +145,7 @@ def _ensure_bucket_ready(
             storage_client, bucket_name, project_id, location
         ):
             return True
-        else:
-            _abort_bucket_setup(is_default)
+        _abort_bucket_setup(is_default)
 
 
 def _configure_remote_state(
@@ -247,7 +246,7 @@ def _validate_namespace(ns: str) -> Tuple[bool, str]:
 
 def _resolve_project_config(
     project_id: str, namespace: str, force: bool
-) -> Tuple[str, str, Path]:
+) -> tuple[str, str, Path]:
     """Resolves project ID and namespace, and determines target directory."""
     if project_id:
         _log_resolved_value("Project ID", project_id, is_default=False)
@@ -503,19 +502,19 @@ def init(
     )
 
 
-def _setup_ingestion_client() -> Tuple[Any, str, str]:
+def _setup_ingestion_client() -> tuple[Any, str, str]:
     click.secho(
         "Fetching ingestion service URL, workflow service account, and Spanner details from Terraform outputs...",
         fg="bright_black",
     )
 
+    from datacommons_admin.ingestion_helper_client import IngestionHelperClient
     from datacommons_admin.tf_utils import (
         get_ingestion_service_url,
         get_ingestion_workflow_service_account_email,
-        get_spanner_instance_id,
         get_spanner_database_id,
+        get_spanner_instance_id,
     )
-    from datacommons_admin.ingestion_helper_client import IngestionHelperClient
 
     url = get_ingestion_service_url()
     sa_email = get_ingestion_workflow_service_account_email()

packages/datacommons-admin/datacommons_admin/ingest_cli.py

@@ -12,16 +12,18 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import click
 import re
 
-from datacommons_admin.ingestion_job_client import IngestionJobClient
+import click
+
+from datacommons_admin.ingestion_helper_client import IngestionHelperClient
 from datacommons_admin.tf_utils import (
     get_ingestion_prep_job_name,
+    get_ingestion_service_url,
+    get_ingestion_workflow_name,
     get_ingestion_workflow_service_account_email,
     get_project_id,
     get_region,
-    get_ingestion_workflow_name,
 )
 
 
@@ -40,36 +42,41 @@ def ingest() -> None:
 def start(imports: str | None = None) -> None:
     """Start a data ingestion job execution."""
     click.secho("Datacommons Admin Ingest Start", fg="cyan", bold=True)
+
     click.secho(
-        "Fetching data job name and workflow service account from Terraform outputs...",
+        "Fetching Ingestion parameters from Terraform outputs...",
         fg="bright_black",
     )
 
+    helper_url = get_ingestion_service_url()
     job_name = get_ingestion_prep_job_name()
     sa_email = get_ingestion_workflow_service_account_email()
     project_id = get_project_id()
     region = get_region()
     workflow_name = get_ingestion_workflow_name()
 
-    click.secho(f"Found data job: {job_name}", fg="green")
-    click.secho(f"Found workflow service account: {sa_email}", fg="green")
-    click.secho(f"Found GCP project ID: {project_id}", fg="green")
-    click.secho(f"Found GCP region: {region}", fg="green")
+    click.secho(f"Found Ingestion Helper URL: {helper_url}", fg="green")
+    click.secho(f"Found Data Job Name: {job_name}", fg="green")
+    click.secho(f"Found Workflow Service Account: {sa_email}", fg="green")
+    click.secho(f"Found GCP Project ID: {project_id}", fg="green")
+    click.secho(f"Found GCP Region: {region}", fg="green")
     click.secho(
-        f"Starting Cloud Run job '{job_name}' via Admin API (this may take a few moments)...",
+        "Triggering data ingestion via Ingestion Helper API...",
         fg="bright_black",
     )
 
-    client = IngestionJobClient(
-        job_name,
+    client = IngestionHelperClient(
+        helper_url,
         service_account_email=sa_email,
-        project_id=project_id,
-        location=region,
     )
-    result = client.start_job(imports=imports)
+    result = client.start_ingestion(imports=imports)
 
     click.secho("Successfully started ingestion job!", fg="green", bold=True)
-    res_name = result.get("name") or result.get("metadata", {}).get("name")
+    res_name = (
+        result.get("operationName")
+        or result.get("name")
+        or result.get("metadata", {}).get("name")
+    )
 
     if res_name:
         op_pattern = r"projects/([^/]+)/locations/([^/]+)/operations/([^/]+)"
@@ -110,27 +117,28 @@ def show_config() -> None:
         fg="bright_black",
     )
 
+    helper_url = get_ingestion_service_url()
     job_name = get_ingestion_prep_job_name()
     sa_email = get_ingestion_workflow_service_account_email()
     project_id = get_project_id()
     region = get_region()
 
-    click.secho(f"Found data job: {job_name}", fg="green")
-    click.secho(f"Found workflow service account: {sa_email}", fg="green")
-    click.secho(f"Found GCP project ID: {project_id}", fg="green")
-    click.secho(f"Found GCP region: {region}", fg="green")
+    click.secho(f"Found Ingestion Helper URL: {helper_url}", fg="green")
+    click.secho(f"Found Data Job Name: {job_name}", fg="green")
+    click.secho(f"Found Workflow Service Account: {sa_email}", fg="green")
+    click.secho(f"Found GCP Project ID: {project_id}", fg="green")
+    click.secho(f"Found GCP Region: {region}", fg="green")
     click.secho(
-        f"Fetching configuration for Cloud Run job '{job_name}'...",
+        f"Fetching configuration for Cloud Run job '{job_name}' via Ingestion Helper...",
         fg="bright_black",
     )
 
-    client = IngestionJobClient(
-        job_name,
+    client = IngestionHelperClient(
+        helper_url,
         service_account_email=sa_email,
-        project_id=project_id,
-        location=region,
     )
-    env_vars = client.get_config()
+    result = client.get_job_config()
+    env_vars = result.get("config", [])
 
     click.secho("\nCurrent ingestion job configuration:", fg="cyan", bold=True)
     if not env_vars:

packages/datacommons-admin/datacommons_admin/ingestion_helper_client.py

@@ -120,3 +120,21 @@ def initialize_database(self) -> dict:
     def seed_database(self) -> dict:
         """Calls the seed_database endpoint on the ingestion helper service."""
         return self._call_endpoint("database/seed")
+
+    def start_ingestion(
+        self, job_name: str | None = None, imports: str | None = None
+    ) -> dict:
+        """Calls the ingestion/start endpoint on the ingestion helper service."""
+        payload = {}
+        if job_name:
+            payload["jobName"] = job_name
+        if imports:
+            payload["imports"] = imports
+        return self._call_endpoint("ingestion/start", payload=payload)
+
+    def get_job_config(self, job_name: str | None = None) -> dict:
+        """Calls the ingestion/config endpoint on the ingestion helper service to get environment variables."""
+        payload = {}
+        if job_name:
+            payload["jobName"] = job_name
+        return self._call_endpoint("ingestion/config", payload=payload)