cloud-sql-python-connector

import os import pymysql from google.cloud.sql.connector import Connector, IPTypes # Recommended: Use Application Default Credentials (ADC). # Ensure GOOGLE_APPLICATION_CREDENTIALS env var is set or ADC is configured. # Initialize Connector connector = Connector() # Function to get a database connection def get_db_connection(): try: conn = connector.connect( os.environ.get("INSTANCE_CONNECTION_NAME", "your-project:your-region:your-instance"), "pymysql", # Specify the DBAPI module you are using user=os.environ.get("DB_USER", "root"), password=os.environ.get("DB_PASS", ""), db=os.environ.get("DB_NAME", "my_database"), ip_type=IPTypes.PUBLIC if os.environ.get("IP_TYPE", "public").lower() == "public" else IPTypes.PRIVATE ) return conn except Exception as e: print(f"Error creating database connection: {e}") return None # Example Usage: if __name__ == "__main__": # Set environment variables for testing, or rely on actual deployment configs. # os.environ['INSTANCE_CONNECTION_NAME'] = 'your-project:your-region:your-instance' # os.environ['DB_USER'] = 'my_user' # os.environ['DB_PASS'] = 'my_password' # os.environ['DB_NAME'] = 'my_database' # os.environ['IP_TYPE'] = 'public' # or 'private' conn = None try: conn = get_db_connection() if conn: with conn.cursor() as cursor: cursor.execute("SELECT 1 + 1 AS solution;") result = cursor.fetchone() print(f"Database connection successful! Result: {result}") except Exception as e: print(f"An error occurred during database operation: {e}") finally: if conn: conn.close() connector.close() # Important: close the connector when your application exits