google-cloud-bigquery-connection

import os from google.cloud import bigquery_connection_v1 # It is highly recommended to set the GOOGLE_APPLICATION_CREDENTIALS environment variable # to the path of your service account key file for authentication. # For example (Linux/macOS): export GOOGLE_APPLICATION_CREDENTIALS="/path/to/keyfile.json" # Or set programmatically, but generally discouraged for production: # os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/path/to/keyfile.json" def list_bigquery_connections(project_id: str, location: str): """Lists all BigQuery connections in a given project and location.""" client = bigquery_connection_v1.BigQueryConnectionServiceClient() # The parent resource for connections is formatted as 'projects/{project}/locations/{location}' parent = f"projects/{project_id}/locations/{location}" print(f"Listing connections in project '{project_id}' at location '{location}':") try: # API call to list connections for connection in client.list_connections(parent=parent): print(f" Connection Name: {connection.name}") print(f" Connection Type: {connection.cloud_resource.connection_type.name}") print(f" Description: {connection.friendly_name or 'N/A'}") # You can access other connection properties like `cloud_sql` or `aws` here except Exception as e: print(f"Error listing connections: {e}") print("Please ensure the BigQuery Connection API is enabled for your project ") print("and the service account has the 'BigQuery Connection User' role or equivalent permissions.") # Replace with your actual project ID and desired location # It's best practice to get project_id from the environment or Google Cloud metadata. PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "your-gcp-project-id") LOCATION = "us" # Example: "us", "eu", "asia-east1" if PROJECT_ID == "your-gcp-project-id": print("WARNING: Please set the 'GOOGLE_CLOUD_PROJECT' environment variable ") print(" or replace 'your-gcp-project-id' with your actual project ID.") print(" Authentication and API calls will likely fail without proper setup.") else: list_bigquery_connections(PROJECT_ID, LOCATION)