feat(gcp): add BigQuery resource ingestion (#2433)

### Type of change
- [x] New feature (non-breaking change that adds functionality)

### Summary

Add BigQuery support to the GCP intel module, covering four resource
types:

- **Datasets** (`GCPBigQueryDataset`) — with `Database` ontology label
- **Tables** (`GCPBigQueryTable`) — TABLE, VIEW, MATERIALIZED_VIEW,
EXTERNAL (enriched via `tables.get` for numBytes, numRows, description,
etc.)
- **Routines** (`GCPBigQueryRoutine`) — stored procedures, UDFs,
table-valued functions
- **Connections** (`GCPBigQueryConnection`) — external data source
connections (Cloud SQL, AWS, Azure, GCP Service Account, Spark, etc.)

**Graph structure:**
```
(GCPProject)-[:RESOURCE]->(GCPBigQueryDataset)-[:HAS_TABLE]->(GCPBigQueryTable)
(GCPProject)-[:RESOURCE]->(GCPBigQueryDataset)-[:HAS_ROUTINE]->(GCPBigQueryRoutine)
(GCPProject)-[:RESOURCE]->(GCPBigQueryConnection)
(GCPBigQueryTable)-[:USES_CONNECTION]->(GCPBigQueryConnection)
(GCPBigQueryRoutine)-[:USES_CONNECTION]->(GCPBigQueryConnection)
(GCPBigQueryConnection)-[:CONNECTS_TO]->(GCPCloudSQLInstance)
(GCPBigQueryConnection)-[:CONNECTS_WITH]->(AWSRole)
(GCPBigQueryConnection)-[:CONNECTS_WITH]->(EntraServicePrincipal)
(GCPBigQueryConnection)-[:CONNECTS_WITH]->(GCPServiceAccount)
```

**Cross-cloud relationships:** BigQuery connections that reference
external identity providers are linked to existing Cartography nodes:
- `aws.accessRole.iamRoleId` → `AWSRole.id` (ARN)
- `azure.federatedApplicationClientId` → `EntraServicePrincipal.id`
- `cloudResource.serviceAccountId` → `GCPServiceAccount.email`

**Implementation details:**
- Uses BigQuery v2 discovery API (`build_client("bigquery", "v2")`) —
consistent with all other GCP modules
- Connections use a separate API service
(`bigqueryconnection.googleapis.com` v1)
- The BigQuery Connection API does not support a wildcard location —
locations are discovered from datasets + default multi-region locations
(us, eu), then queried individually
- Connection IDs are normalized from the short dot-separated format
(`project_number.location.name`) to the full resource name
(`projects/.../locations/.../connections/...`) for consistent
relationship matching
- Tables are enriched via per-table `tables.get` calls for fields not
available in `tables.list` (numBytes, numRows, description,
friendlyName, externalDataConfiguration)
- Returns `None` on API disabled / permission denied (403/404) —
preserves existing data (no cleanup on failure)
- Child resources (tables, routines) are aggregated across all datasets
into a single `load()` call for performance
- Uses direct indexing for required API fields (fail-fast on missing
data)
- Follows the flat-file pattern (like bigtable) with one file per
resource type
- Ontology mapping uses `dataset_id` (required field) for the Database
name instead of `friendly_name` (optional)

**Setup requirements** (documented in
`docs/root/modules/gcp/config.md`):
- APIs: `bigquery.googleapis.com`, `bigqueryconnection.googleapis.com`
- IAM roles: `roles/bigquery.dataViewer`,
`roles/bigquery.connectionUser`

### Related issues or links

- Fixes #2128
- Supersedes #2229

### How was this tested?

- Integration tests covering all 4 resource types with mock API
responses
- Verifies all node properties, relationships (RESOURCE, HAS_TABLE,
HAS_ROUTINE, USES_CONNECTION, CONNECTS_TO, CONNECTS_WITH), and Database
ontology label
- Tested against live GCP project with BigQuery Omni (AWS cross-cloud)
connections
- Linter passes (`make test_lint`)

### Checklist

#### General
- [x] I have read the [contributing
guidelines](https://cartography-cncf.github.io/cartography/dev/developer-guide.html).
- [x] The linter passes locally (`make lint`).
- [x] I have added/updated tests that prove my fix is effective or my
feature works.

#### Proof of functionality
- [x] New or updated unit/integration tests.

#### If you are adding or modifying a synced entity
- [x] Included Cartography sync logs from a real environment
demonstrating successful synchronization of the new/modified entity.

#### If you are changing a node or relationship
- [x] Updated the [schema
documentation](https://github.com/cartography-cncf/cartography/tree/master/docs/root/modules).

#### If you are implementing a new intel module
- [x] Used the NodeSchema [data
model](https://cartography-cncf.github.io/cartography/dev/writing-intel-modules.html#defining-a-node).

### Notes for reviewers

This builds on the work started in #2229 but addresses all review
feedback:
- Uses discovery API instead of `google.cloud.bigquery` client
- One model per file in `cartography/models/gcp/bigquery/`
- Standard error handling with `is_api_disabled_error()` + 403/404
handling
- No catch-all exceptions — let errors propagate
- No datetime conversion — passes native API values
- Aggregates child resources across parents for single `load()` call
- Integration tests (not unit tests) following the standard pattern
- Passes credentials through to `build_client()` instead of using ADC
independently
- Required fields use direct indexing (`data["field"]`) for fail-fast
semantics

---------

Signed-off-by: Jeremy Chapeau <jeremy@subimage.io>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: jychp

cartography/intel/gcp/__init__.py

@@ -15,6 +15,10 @@
 from cartography.config import Config
 from cartography.graph.job import GraphJob
 from cartography.intel.gcp import artifact_registry
+from cartography.intel.gcp import bigquery_connection
+from cartography.intel.gcp import bigquery_dataset
+from cartography.intel.gcp import bigquery_routine
+from cartography.intel.gcp import bigquery_table
 from cartography.intel.gcp import bigtable_app_profile
 from cartography.intel.gcp import bigtable_backup
 from cartography.intel.gcp import bigtable_cluster
@@ -65,7 +69,7 @@
 # and https://cloud.google.com/service-usage/docs/reference/rest/v1/services#ServiceConfig
 Services = namedtuple(
     "Services",
-    "compute storage gke dns iam kms bigtable cai aiplatform cloud_sql gcf secretsmanager artifact_registry cloud_run",
+    "compute storage gke dns iam kms bigtable cai aiplatform cloud_sql gcf secretsmanager artifact_registry cloud_run bigquery bigquery_connection",
 )
 service_names = Services(
     compute="compute.googleapis.com",
@@ -82,6 +86,8 @@
     secretsmanager="secretmanager.googleapis.com",
     artifact_registry="artifactregistry.googleapis.com",
     cloud_run="run.googleapis.com",
+    bigquery="bigquery.googleapis.com",
+    bigquery_connection="bigqueryconnection.googleapis.com",
 )
 
 
@@ -549,6 +555,62 @@ def _sync_project_resources(
                 common_job_parameters,
             )
 
+        # Build the BigQuery v2 client once — used for datasets/tables/routines
+        # and also for location discovery when syncing connections.
+        bigquery_client = None
+        if service_names.bigquery in enabled_services:
+            bigquery_client = build_client(
+                "bigquery",
+                "v2",
+                credentials=credentials,
+            )
+
+        if service_names.bigquery_connection in enabled_services:
+            logger.info("Syncing GCP project %s for BigQuery connections.", project_id)
+            bigquery_conn_client = build_client(
+                "bigqueryconnection",
+                "v1",
+                credentials=credentials,
+            )
+            bigquery_connection.sync_bigquery_connections(
+                neo4j_session,
+                bigquery_conn_client,
+                project_id,
+                gcp_update_tag,
+                common_job_parameters,
+                bigquery_client=bigquery_client,
+            )
+
+        datasets_raw = None
+        if bigquery_client is not None:
+            logger.info("Syncing GCP project %s for BigQuery.", project_id)
+            datasets_raw = bigquery_dataset.sync_bigquery_datasets(
+                neo4j_session,
+                bigquery_client,
+                project_id,
+                gcp_update_tag,
+                common_job_parameters,
+            )
+
+        if bigquery_client is not None and datasets_raw is not None:
+            bigquery_table.sync_bigquery_tables(
+                neo4j_session,
+                bigquery_client,
+                datasets_raw,
+                project_id,
+                gcp_update_tag,
+                common_job_parameters,
+            )
+
+            bigquery_routine.sync_bigquery_routines(
+                neo4j_session,
+                bigquery_client,
+                datasets_raw,
+                project_id,
+                gcp_update_tag,
+                common_job_parameters,
+            )
+
         # Clean up project-level IAM resources (service accounts and project roles)
         # Only run cleanup if IAM sync succeeded to avoid deleting valid data
         # when sync was skipped due to permission issues.

cartography/intel/gcp/bigquery_connection.py

@@ -0,0 +1,216 @@
+import logging
+
+import neo4j
+from googleapiclient.discovery import Resource
+from googleapiclient.errors import HttpError
+
+from cartography.client.core.tx import load
+from cartography.graph.job import GraphJob
+from cartography.intel.gcp.util import gcp_api_execute_with_retry
+from cartography.intel.gcp.util import is_api_disabled_error
+from cartography.models.gcp.bigquery.connection import GCPBigQueryConnectionSchema
+from cartography.util import timeit
+
+logger = logging.getLogger(__name__)
+
+
+def _get_locations(bigquery_client: Resource, project_id: str) -> list[str]:
+    """
+    List available BigQuery locations for a project using the BigQuery v2 API.
+
+    The BigQuery Connection API does not expose a locations.list endpoint, so we
+    use the BigQuery v2 API (datasets.list with a dry-run or projects API) instead.
+    BigQuery v2 does not have a dedicated locations endpoint either, so we query
+    the Cloud Resource Manager locations via the datasets API — specifically, we
+    list datasets to discover which locations the project uses, and supplement with
+    standard multi-region locations to ensure we don't miss connections in locations
+    without datasets.
+
+    Returns a deduplicated list of location IDs (e.g., ["us", "eu", "us-central1"]).
+    """
+    # Standard BigQuery multi-region and common regional locations.
+    # Connections can exist in any of these even without datasets.
+    # See https://cloud.google.com/bigquery/docs/locations
+    default_locations = {"us", "eu"}
+
+    # Discover additional locations from existing datasets
+    locations: set[str] = set(default_locations)
+    try:
+        request = bigquery_client.datasets().list(projectId=project_id, all=True)
+        while request is not None:
+            response = gcp_api_execute_with_retry(request)
+            for ds in response.get("datasets", []):
+                loc = ds.get("location")
+                if loc:
+                    locations.add(loc.lower())
+            request = bigquery_client.datasets().list_next(
+                previous_request=request,
+                previous_response=response,
+            )
+    except HttpError as e:
+        logger.debug(
+            "Could not list datasets to discover locations for project %s - %s. "
+            "Using default locations only.",
+            project_id,
+            e,
+        )
+
+    return list(locations)
+
+
+@timeit
+def get_bigquery_connections(
+    conn_client: Resource,
+    project_id: str,
+    bigquery_client: Resource | None = None,
+) -> list[dict] | None:
+    """
+    Gets BigQuery connections for a project across all locations.
+
+    The BigQuery Connection API does not support a wildcard location, so we
+    discover locations from the BigQuery v2 API (via dataset locations) plus
+    standard multi-region locations, then query each one individually.
+
+    Args:
+        conn_client: The bigqueryconnection v1 API client.
+        project_id: The GCP project ID.
+        bigquery_client: Optional BigQuery v2 API client for location discovery.
+            If not provided, only default locations (us, eu) are queried.
+
+    Returns:
+        list[dict]: List of BigQuery connections
+        None: If the API is not enabled or access is denied
+
+    Raises:
+        HttpError: For errors other than API disabled or permission denied
+    """
+    if bigquery_client is not None:
+        locations = _get_locations(bigquery_client, project_id)
+    else:
+        locations = ["us", "eu"]
+
+    connections: list[dict] = []
+    for location in locations:
+        parent = f"projects/{project_id}/locations/{location}"
+        try:
+            request = (
+                conn_client.projects()
+                .locations()
+                .connections()
+                .list(
+                    parent=parent,
+                )
+            )
+            while request is not None:
+                response = gcp_api_execute_with_retry(request)
+                connections.extend(response.get("connections", []))
+                request = (
+                    conn_client.projects()
+                    .locations()
+                    .connections()
+                    .list_next(
+                        previous_request=request,
+                        previous_response=response,
+                    )
+                )
+        except HttpError as e:
+            if is_api_disabled_error(e) or e.resp.status in (403, 404):
+                logger.warning(
+                    "Could not retrieve BigQuery connections for %s/%s - %s. "
+                    "Skipping location.",
+                    project_id,
+                    location,
+                    e,
+                )
+                continue
+            raise
+
+    return connections
+
+
+def transform_connections(connections_data: list[dict], project_id: str) -> list[dict]:
+    transformed: list[dict] = []
+    for conn in connections_data:
+        # Determine connection type from the oneOf fields in the API response
+        connection_type = None
+        for type_key in (
+            "cloudSql",
+            "aws",
+            "azure",
+            "cloudSpanner",
+            "cloudResource",
+            "spark",
+        ):
+            if type_key in conn:
+                connection_type = type_key
+                break
+
+        cloud_sql = conn.get("cloudSql", {}) or {}
+        aws = conn.get("aws", {}) or {}
+        azure = conn.get("azure", {}) or {}
+        cloud_resource = conn.get("cloudResource", {}) or {}
+        transformed.append(
+            {
+                "name": conn["name"],
+                "friendlyName": conn.get("friendlyName"),
+                "description": conn.get("description"),
+                "connection_type": connection_type,
+                "creationTime": conn.get("creationTime"),
+                "lastModifiedTime": conn.get("lastModifiedTime"),
+                "hasCredential": conn.get("hasCredential"),
+                "cloud_sql_instance_id": cloud_sql.get("instanceId"),
+                "aws_role_arn": aws.get("accessRole", {}).get("iamRoleId"),
+                "azure_app_client_id": azure.get("federatedApplicationClientId"),
+                "service_account_id": cloud_resource.get("serviceAccountId"),
+                "project_id": project_id,
+            },
+        )
+    return transformed
+
+
+@timeit
+def load_bigquery_connections(
+    neo4j_session: neo4j.Session,
+    data: list[dict],
+    project_id: str,
+    update_tag: int,
+) -> None:
+    load(
+        neo4j_session,
+        GCPBigQueryConnectionSchema(),
+        data,
+        lastupdated=update_tag,
+        PROJECT_ID=project_id,
+    )
+
+
+@timeit
+def cleanup_bigquery_connections(
+    neo4j_session: neo4j.Session,
+    common_job_parameters: dict,
+) -> None:
+    GraphJob.from_node_schema(
+        GCPBigQueryConnectionSchema(),
+        common_job_parameters,
+    ).run(neo4j_session)
+
+
+@timeit
+def sync_bigquery_connections(
+    neo4j_session: neo4j.Session,
+    client: Resource,
+    project_id: str,
+    update_tag: int,
+    common_job_parameters: dict,
+    bigquery_client: Resource | None = None,
+) -> None:
+    logger.info("Syncing BigQuery connections for project %s.", project_id)
+    connections_raw = get_bigquery_connections(client, project_id, bigquery_client)
+
+    if connections_raw is not None:
+        connections = transform_connections(connections_raw, project_id)
+        load_bigquery_connections(neo4j_session, connections, project_id, update_tag)
+
+        cleanup_job_params = common_job_parameters.copy()
+        cleanup_job_params["PROJECT_ID"] = project_id
+        cleanup_bigquery_connections(neo4j_session, cleanup_job_params)