Connecting to Cloud SQL for PostgreSQL with Python via a Unix Socket

Establishes a database connection to a Cloud SQL for PostgreSQL instance using a Unix socket. It utilizes the SQLAlchemy library to create a connection engine, configured specifically for the pg8000 driver. The script retrieves necessary connection parameters like database user, password, database name, and the Unix socket path from environment variables. This approach is particularly useful for applications running in the same Google Cloud region as the Cloud SQL instance, offering a secure and efficient way to communicate with the database without needing to configure IP allowlisting for TCP connections.

Explore further

For detailed documentation that includes this code sample, see the following:

Code sample

Python

To authenticate to Cloud SQL for PostgreSQL, set up Application Default Credentials. For more information, see Set up authentication for a local development environment. 

import os

import sqlalchemy


def connect_unix_socket() -> sqlalchemy.engine.base.Engine:
    """Initializes a Unix socket connection pool for a Cloud SQL instance of Postgres."""
    # Note: Saving credentials in environment variables is convenient, but not
    # secure - consider a more secure solution such as
    # Cloud Secret Manager (https://cloud.google.com/secret-manager) to help
    # keep secrets safe.
    db_user = os.environ["DB_USER"]  # e.g. 'my-database-user'
    db_pass = os.environ["DB_PASS"]  # e.g. 'my-database-password'
    db_name = os.environ["DB_NAME"]  # e.g. 'my-database'
    unix_socket_path = os.environ[
        "INSTANCE_UNIX_SOCKET"
    ]  # e.g. '/cloudsql/project:region:instance'

    pool = sqlalchemy.create_engine(
        # Equivalent URL:
        # postgresql+pg8000://<db_user>:<db_pass>@/<db_name>
        #                         ?unix_sock=<INSTANCE_UNIX_SOCKET>/.s.PGSQL.5432
        # Note: Some drivers require the `unix_sock` query parameter to use a different key.
        # For example, 'psycopg2' uses the path set to `host` in order to connect successfully.
        sqlalchemy.engine.url.URL.create(
            drivername="postgresql+pg8000",
            username=db_user,
            password=db_pass,
            database=db_name,
            query={"unix_sock": f"{unix_socket_path}/.s.PGSQL.5432"},
        ),
        # ...
    )
    return pool

What's next

To search and filter code samples for other Google Cloud products, see the Google Cloud sample browser.