Google Cloud SQL Python Connector

1.20.1 · active · verified Thu Apr 09

The Google Cloud SQL Python Connector is a client library that helps connect Python applications to Google Cloud SQL databases. It automatically handles authentication, encryption, and secure connection management, acting as a smart proxy for your database connections. As of version 1.20.1, it provides robust and secure connectivity. The library is actively maintained by Google, with frequent updates aligning with Cloud SQL proxy and client library best practices.

Warnings

breaking Major breaking changes occurred in version 1.0.0. The primary connection method was renamed from `create_connection` to `connect`, and the `database_driver` argument was removed in favor of passing the DBAPI module object (e.g., `pymysql`) directly.
Fix: Update your code to use `connector.connect(instance_connection_name, dbapi_module, ...)` and ensure you pass the DBAPI module (e.g., `pymysql`) as the second argument.
breaking The `refresh_delay` parameter in the `Connector` constructor was removed, as its functionality is now handled internally and dynamically by the library.
Fix: Remove the `refresh_delay` parameter and its value from your `Connector` constructor calls. The library will manage token refreshes automatically.
gotcha You must install the appropriate database driver extra (e.g., `[pymysql]`, `[pg8000]`, `[psycopg2]`) when installing the connector, otherwise, connection attempts will fail with module not found errors.
Fix: Install the connector with the correct extra: `pip install cloud-sql-python-connector[<driver>]`. For example, `pip install cloud-sql-python-connector[pymysql]`.
gotcha Incorrect `instance_connection_name` format. The correct format is `PROJECT_ID:REGION:INSTANCE_NAME`. Using just `INSTANCE_NAME` or other variations will lead to connection failures.
Fix: Always provide the full instance connection name in `PROJECT_ID:REGION:INSTANCE_NAME` format, typically retrieved from the Cloud SQL instance details in the Google Cloud Console.

Install

pip install cloud-sql-python-connector[pymysql] For MySQL using PyMySQL
pip install cloud-sql-python-connector[pg8000] For PostgreSQL using pg8000
pip install cloud-sql-python-connector[psycopg2] For PostgreSQL using psycopg2

Imports

Connector

from google.cloud.sql.connector import Connector

IPTypes

from google.cloud.sql.connector import IPTypes

Quickstart

This quickstart demonstrates connecting to a MySQL Cloud SQL instance using PyMySQL. It relies on environment variables for sensitive information and the instance connection name. Ensure you have the `cloud-sql-python-connector[pymysql]` package installed. The connector automatically handles authentication using Application Default Credentials (ADC) or a service account key specified by `GOOGLE_APPLICATION_CREDENTIALS`.

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

view raw JSON →