DOCSEARCHbyAlgolia
Submit Docs

Cloud SQL Python Connector

repository·main·Indexed 43 days ago

https://github.com/googlecloudplatform/cloud-sql-python-connector

A native Python library providing secure, IAM-authorized connections to Cloud SQL instances (MySQL, PostgreSQL, and SQL Server) using TLS 1.3 encryption. It serves as an alternative to the Cloud SQL Auth Proxy, supporting various drivers including pymysql, pg8000, psycopg, asyncpg, and pytds. The connector integrates with SQLAlchemy, Flask-SQLAlchemy, and FastAPI, and supports features such as Automatic IAM database authentication, DNS-based instance lookup for disaster recovery, and asynchronous connections.

Tokens
10.2K
Snippets
34
Records
40
Agent score
47%

What's inside Cloud SQL Python Connector

  1. Configure credentials for the Cloud SQL Python Connector

    main

    The library uses the Application Default Credentials (ADC) strategy to resolve credentials.

    To use the connector, you must have ADC configured in your environment (e.g., via service account keys or local development authentication). For more details on setting up ADC, consult the google-auth package or the official Google Cloud documentation.

    If you need to explicitly set a specific source for credentials, refer to the Configuring the Connector section in the documentation.

  2. Use DNS names for Advanced Disaster Recovery and Failover

    main

    The connector can resolve instance names via DNS. This allows for automatic failover: when the DNS record (a TXT record containing the instance connection name) changes, the connector detects the change, closes old connections, and directs new connections to the new instance.

    To use this, initialize the Connector with resolver=DnsResolver and pass the domain name instead of the connection name to connect().

    from google.cloud.sql.connector import Connector, DnsResolver
import sqlalchemy

# initialize Cloud SQL Python Connector with `resolver=DnsResolver`
with Connector(resolver=DnsResolver) as connector:
    pool = sqlalchemy.create_engine(
        "mysql+pymysql://",
        creator=lambda: connector.connect(
            "prod-db.mycompany.example.com",  # using DNS name
            "pymysql",
            user="my-user",
            password="my-password",
            db="my-db-name"
        ),
    )

  3. Best Practice: Use Lazy Instantiation for Connector and SQLAlchemy in Cloud Run

    main

    When developing Python applications for Cloud Run using the Cloud SQL Python Connector, it is recommended to use Lazy Instantiation for the Connector and SQLAlchemy Engine objects.

    Implementation Pattern

    Define Connector and Engine as global variables initialized to None. Instantiate them inside your request handlers only when a request is received.

    Benefits

    • Faster Startup: Reduces cold start latency by deferring expensive initialization until the first request.
    • Resource Efficiency: Prevents background processes from running (and incurring costs) when the CPU is throttled during idle periods.
    • Connection Reuse: Global instances are reused across subsequent requests to the same container, preventing connection limit exhaustion.

    Recommended Configuration

    Initialize the Connector with refresh_strategy=lazy to optimize cost and avoid connection errors during CPU throttling.

  4. Build and Push Docker images to Artifact Registry for Cloud Run

    main

    To deploy a Cloud SQL Python Connector application to Cloud Run, you must first build and push your container image to Artifact Registry.

    1. Enable the API:
      gcloud services enable artifactregistry.googleapis.com
    2. Create a repository:
      gcloud artifacts repositories create REPO_NAME \
  --repository-format=docker \
  --location=REGION
    3. Authenticate Docker:
      gcloud auth configure-docker REGION-docker.pkg.dev
    4. Build the image (replace mysql with postgres or sqlserver as needed):
      docker build -t REGION-docker.pkg.dev/PROJECT_ID/REPO_NAME/IMAGE_NAME mysql
    5. Push the image:
      docker push REGION-docker.pkg.dev/PROJECT_ID/REPO_NAME/IMAGE_NAME
    gcloud services enable artifactregistry.googleapis.com

gcloud artifacts repositories create REPO_NAME \
  --repository-format=docker \
  --location=REGION

gcloud auth configure-docker REGION-docker.pkg.dev

docker build -t REGION-docker.pkg.dev/PROJECT_ID/REPO_NAME/IMAGE_NAME mysql

docker push REGION-docker.pkg.dev/PROJECT_ID/REPO_NAME/IMAGE_NAME

  5. Deploy Cloud Run service for MySQL or PostgreSQL (Private IP)

    main

    Deploy a Cloud Run service connecting to a MySQL or PostgreSQL instance via Private IP. This requires configuring VPC networking and setting IP_TYPE=PRIVATE.

    Additional Configuration:

    • Set IP_TYPE=PRIVATE in --set-env-vars.
    • Specify --network and --subnet.
    • Set --vpc-egress=private-ranges-only to ensure traffic stays within the VPC.
    gcloud run deploy SERVICE_NAME \
  --image=REGION-docker.pkg.dev/PROJECT_ID/REPO_NAME/IMAGE_NAME \
  --set-env-vars=DB_USER=DB_USER,DB_IAM_USER=DB_IAM_USER,DB_NAME=DB_NAME,DB_SECRET_NAME=DB_SECRET_NAME,INSTANCE_CONNECTION_NAME=INSTANCE_CONNECTION_NAME,IP_TYPE=PRIVATE \
  --network=VPC_NETWORK \
  --subnet=SUBNET_NAME \
  --vpc-egress=private-ranges-only \
  --region=REGION \
  --update-secrets=DB_PASSWORD=DB_PASSWORD:latest 
    gcloud run deploy SERVICE_NAME \
  --image=REGION-docker.pkg.dev/PROJECT_ID/REPO_NAME/IMAGE_NAME \
  --set-env-vars=DB_USER=DB_USER,DB_IAM_USER=DB_IAM_USER,DB_NAME=DB_NAME,DB_SECRET_NAME=DB_SECRET_NAME,INSTANCE_CONNECTION_NAME=INSTANCE_CONNECTION_NAME,IP_TYPE=PRIVATE \
  --network=VPC_NETWORK \
  --subnet=SUBNET_NAME \
  --vpc-egress=private-ranges-only \
  --region=REGION \
  --update-secrets=DB_PASSWORD=DB_PASSWORD:latest

  6. Configure APIs and IAM permissions for the Connector

    main

    To successfully establish connections to Cloud SQL, ensure the following requirements are met:

    1. IAM Permissions: The IAM principal (user or service account) used by the application must have the Cloud SQL Client role.
    2. API Activation: The Cloud SQL Admin API must be enabled in your Google Cloud Project. By default, the connector calls the API in the project associated with the IAM principal.

  7. Deploy Cloud Run service for SQL Server (Private IP)

    main

    Deploy a Cloud Run service connecting to a Cloud SQL for SQL Server instance via Private IP.

    Note: Cloud SQL for SQL Server does not support IAM database authentication. Use password-based authentication.

    gcloud run deploy SERVICE_NAME \
  --image=REGION-docker.pkg.dev/PROJECT_ID/REPO_NAME/IMAGE_NAME \
  --set-env-vars=DB_USER=DB_USER,DB_NAME=DB_NAME,DB_SECRET_NAME=DB_SECRET_NAME,INSTANCE_CONNECTION_NAME=INSTANCE_CONNECTION_NAME,IP_TYPE=PRIVATE \
  --network=VPC_NETWORK \
  --subnet=SUBNET_NAME \
  --vpc-egress=private-ranges-only \
  --region=REGION \
  --update-secrets=DB_PASSWORD=DB_PASSWORD:latest 
    gcloud run deploy SERVICE_NAME \
  --image=REGION-docker.pkg.dev/PROJECT_ID/REPO_NAME/IMAGE_NAME \
  --set-env-vars=DB_USER=DB_USER,DB_NAME=DB_NAME,DB_SECRET_NAME=DB_SECRET_NAME,INSTANCE_CONNECTION_NAME=INSTANCE_CONNECTION_NAME,IP_TYPE=PRIVATE \
  --network=VPC_NETWORK \
  --subnet=SUBNET_NAME \
  --vpc-egress=private-ranges-only \
  --region=REGION \
  --update-secrets=DB_PASSWORD=DB_PASSWORD:latest

  8. Use Async Drivers (asyncpg) with the Connector

    main

    The connector supports asyncio via create_async_connector() and the connect_async() method. This is recommended for high-performance asynchronous applications.

    Key Methods:

    • create_async_connector(): Returns a Connector object that uses the current thread's running event loop.
    • connect_async(...): Asynchronous version of connect().
    • close_async(): Asynchronous version of close().

    Note: When using Connector as an async context manager, you must pass the running event loop as the loop argument.

    import asyncpg
from google.cloud.sql.connector import Connector, create_async_connector

async def main():
    connector = await create_async_connector()

    async def getconn(instance_connection_name, **kwargs) -> asyncpg.Connection:
        return await connector.connect_async(
            instance_connection_name,
            "asyncpg",
            user="my-user",
            password="my-password",
            db="my-db",
            **kwargs,
        )

    pool = await asyncpg.create_pool(
        "my-project:my-region:my-instance", connect=getconn
    )

    async with pool.acquire() as conn:
        res = await conn.fetch("SELECT NOW()")
    
    await connector.close_async()

  9. How to use the Cloud SQL Python Connector

    main

    To connect to Cloud SQL, initialize a Connector object and call its connect method. The connect method requires the instance connection name (format project:region:instance) as the first argument and the database driver name as the second. Other connection parameters like user, password, and db should be passed as keyword arguments.

    Note that the Connector does not manage connection pooling itself; it is recommended to use it with a pooling library like SQLAlchemy to reuse connections and reduce overhead.

    from google.cloud.sql.connector import Connector
import sqlalchemy

# initialize Connector object
connector = Connector()

# initialize SQLAlchemy connection pool with Connector
pool = sqlalchemy.create_engine(
    "mysql+pymysql://",
    creator=lambda: connector.connect(
        "project:region:instance",
        "pymysql",
        user="my-user",
        password="my-password",
        db="my-db-name"
    ),
)

# interact with database
with pool.connect() as db_conn:
    result = db_conn.execute(sqlalchemy.text("SELECT * from my_table")).fetchall()
    db_conn.commit()

# close the Connector
connector.close()

  10. Enable Automatic IAM Database Authentication

    main

    If your Cloud SQL instance is configured for IAM authentication, you can connect using IAM principals instead of passwords. Set enable_iam_auth=True in the connect method.

    User Formatting Requirements:

    • Postgres: Use the user's email address. For service accounts, use the email without the .gserviceaccount.com suffix.
    • MySQL: Use the user's email address without the @ or domain name (e.g., test-user@gmail.com becomes test-user). For service accounts, use the email without the @project-id.iam.gserviceaccount.com suffix.
    conn = connector.connect(
     "project:region:instance",
     "pg8000",
     user="postgres-iam-user@gmail.com",
     db="my-db-name",
     enable_iam_auth=True,
 )

  11. Install the Cloud SQL Python Connector

    main

    Install the cloud-sql-python-connector package using pip. You must specify an extra to include the database driver compatible with your specific database dialect.

    MySQL

    Use pymysql:

    pip install "cloud-sql-python-connector[pymysql]"

    Postgres

    Choose one of the following supported drivers:

    • pg8000: pip install "cloud-sql-python-connector[pg8000]"
    • psycopg: pip install "cloud-sql-python-connector[psycopg]"
    • asyncpg: pip install "cloud-sql-python-connector[asyncpg]"

    SQL Server

    Use pytds:

    pip install "cloud-sql-python-connector[pytds]"
    pip install "cloud-sql-python-connector[pymysql]"

  12. Use the Connector as a context manager

    main

    To ensure background resources are automatically cleaned up without calling connector.close() or connector.close_async(), use the Connector as a context manager. This works for both synchronous and asynchronous usage.

    from google.cloud.sql.connector import Connector
import sqlalchemy

# initialize Cloud SQL Python Connector as context manager
with Connector() as connector:
    pool = sqlalchemy.create_engine(
        "mysql+pymysql://",
        creator=lambda: connector.connect(
            "project:region:instance",
            "pymysql",
            user="my-user",
            password="my-password",
            db="my-db-name"
        ),
    )
    # ... use pool ...
# Connector is automatically closed here