google-cloud-alloydb-connector

import os import psycopg2 from google.cloud.alloydb.connector import Connector # --- Environment Variables (Recommended for Production) --- # Replace with your actual AlloyDB instance connection details or set as ENV vars. # Ensure the service account running this code has 'AlloyDB Connection User' role. PROJECT_ID = os.environ.get('GOOGLE_CLOUD_PROJECT', 'your-project-id') REGION = os.environ.get('ALLOYDB_REGION', 'us-central1') CLUSTER_ID = os.environ.get('ALLOYDB_CLUSTER_ID', 'your-cluster-id') INSTANCE_ID = os.environ.get('ALLOYDB_INSTANCE_ID', 'your-instance-id') DB_USER = os.environ.get('ALLOYDB_DB_USER', 'postgres') DB_PASS = os.environ.get('ALLOYDB_DB_PASS', 'your-db-password') # Use Secret Manager in production DB_NAME = os.environ.get('ALLOYDB_DB_NAME', 'postgres') # Full instance connection name format: projects/<PROJECT>/locations/<REGION>/clusters/<CLUSTER>/instances/<INSTANCE> INSTANCE_CONNECTION_NAME = ( f"projects/{PROJECT_ID}/locations/{REGION}/clusters/{CLUSTER_ID}/instances/{INSTANCE_ID}" ) def main(): connector = None conn = None try: # 1. Initialize the AlloyDB connector. It uses google.auth.default() for credentials. connector = Connector() print(f"Attempting to connect to AlloyDB instance: {INSTANCE_CONNECTION_NAME}") # 2. Connect to the AlloyDB instance using psycopg2. # The connector automatically handles the secure proxy connection. conn: psycopg2.Connection = connector.connect( INSTANCE_CONNECTION_NAME, "psycopg2", # Specify the database driver user=DB_USER, password=DB_PASS, db_name=DB_NAME, ) # 3. Execute a simple query to verify the connection. with conn.cursor() as cursor: cursor.execute("SELECT version();") version = cursor.fetchone()[0] print(f"Successfully connected to AlloyDB. PostgreSQL version: {version}") except ImportError: print("ERROR: PostgreSQL driver 'psycopg2-binary' not found. Please install with: ") print(" pip install google-cloud-alloydb-connector[psycopg2]") except Exception as e: print(f"An error occurred during connection: {e}") # Common errors: if "PERMISSION_DENIED" in str(e): print("Hint: Ensure the service account has 'AlloyDB Connection User' role.") elif "Invalid instance connection name" in str(e): print("Hint: Check the format of INSTANCE_CONNECTION_NAME.") finally: # 4. Clean up resources. if conn: conn.close() if connector: connector.close() # Important: close the connector to release resources if __name__ == "__main__": main()