Repository files navigation

TheCloud SQL Python Connector is a Cloud SQL connector designed for use with thePython language. Using a Cloud SQL connector provides a native alternative to theCloud SQL Auth Proxy whileproviding the following benefits:

• IAM Authorization: uses IAM permissions to control who/what can connect toyour Cloud SQL instances
• Improved Security: uses robust, updated TLS 1.3 encryption andidentity verification between the client connector and the server-side proxy,independent of the database protocol.
• Convenience: removes the requirement to use and distribute SSLcertificates, as well as manage firewalls or source/destination IP addresses.
• (optionally)IAM DB Authentication: provides support forCloud SQL’s automatic IAM DB AuthN feature.

The Cloud SQL Python Connector is a package to be used alongside a database driver.Currently supported drivers are:

• pymysql (MySQL)
• pg8000 (PostgreSQL)
• asyncpg (PostgreSQL)
• pytds (SQL Server)

You can install this library withpip install, specifying the driverbased on your database dialect.

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

There are two different database drivers that are supported for the Postgres dialect:

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

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

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

This package requires the following to successfully make Cloud SQL Connections:

• IAM principal (user, service account, etc.) with theCloud SQL Client role. This IAM principal will be used forcredentials.
• TheCloud SQL Admin API to be enabled within your Google CloudProject. By default, the API will be called in the project associated withthe IAM principal.

This library uses theApplication Default Credentials (ADC) strategy forresolving credentials. Please seethese instructions for how to set your ADC(Google Cloud Application vs Local Development, IAM user vs service account credentials),or consult thegoogle.auth package.

To explicitly set a specific source for the credentials, seeConfiguring the Connector below.

This package provides several functions for authorizing and encryptingconnections. These functions are used with your database driver to connect toyour Cloud SQL instance.

The instance connection name for your Cloud SQL instance is always in theformat "project:region:instance".

To connect to Cloud SQL using the connector, inititalize aConnectorobject and call itsconnect method with the proper input parameters.

TheConnector itself creates connection objects by calling itsconnect method but does not manage database connection pooling. For this reason, it is recommended to use the connector alongside a library that can create connection pools, such asSQLAlchemy. This will allow for connections to remain open and be reused, reducing connection overhead and the number of connections needed.

In the Connector'sconnect method below, input your connection string as the first positional argument and the name of the database driver for the second positional argument. Insert the rest of your connection keyword arguments like user, password and database. You can also set the optionaltimeout orip_type keyword arguments.

To use this connector with SQLAlchemy, use thecreator argument forsqlalchemy.create_engine:

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

The returned connection pool engine can then be used to query and modify the database.

# insert statementinsert_stmt=sqlalchemy.text("INSERT INTO my_table (id, title) VALUES (:id, :title)",)withpool.connect()asdb_conn:# insert into databasedb_conn.execute(insert_stmt,parameters={"id":"book1","title":"Book One"})# query databaseresult=db_conn.execute(sqlalchemy.text("SELECT * from my_table")).fetchall()# commit transaction (SQLAlchemy v2.X.X is commit as you go)db_conn.commit()# Do something with the resultsforrowinresult:print(row)

To close theConnector object's background resources, call itsclose() method as follows:

connector.close()

If you need to customize something about the connector, or want to specifydefaults for each connection to make, you can initialize aConnector object as follows:

fromgoogle.cloud.sql.connectorimportConnector# Note: all parameters below are optionalconnector=Connector(ip_type="public",# can also be "private" or "psc"enable_iam_auth=False,timeout=30,credentials=custom_creds,# google.auth.credentials.Credentialsrefresh_strategy="lazy",# can be "lazy" or "background")

TheConnector object can also be used as a context manager in order toautomatically close and cleanup resources, removing the need for explicitcalls toconnector.close().

Connector as a context manager:

fromgoogle.cloud.sql.connectorimportConnectorimportsqlalchemy# initialize Cloud SQL Python Connector as context managerwithConnector()asconnector:# initialize SQLAlchemy connection pool with Connectorpool=sqlalchemy.create_engine("mysql+pymysql://",creator=lambda:connector.connect("project:region:instance","pymysql",user="my-user",password="my-password",db="my-db-name"        ),    )# insert statementinsert_stmt=sqlalchemy.text("INSERT INTO my_table (id, title) VALUES (:id, :title)",    )# interact with Cloud SQL database using connection poolwithpool.connect()asdb_conn:# insert into databasedb_conn.execute(insert_stmt,parameters={"id":"book1","title":"Book One"})# commit transaction (SQLAlchemy v2.X.X is commit as you go)db_conn.commit()# query databaseresult=db_conn.execute(sqlalchemy.text("SELECT * from my_table")).fetchall()# Do something with the resultsforrowinresult:print(row)

The Connector'srefresh_strategy argument can be set to"lazy" to configurethe Python Connector to retrieve connection info lazily and as-needed.Otherwise, a background refresh cycle runs to retrive the connection infoperiodically. This setting is useful in environments where the CPU may bethrottled outside of a request context, e.g., Cloud Run, Cloud Functions, etc.

To set the refresh strategy, set therefresh_strategy keyword argument wheninitializing aConnector:

connector=Connector(refresh_strategy="lazy")

The Cloud SQL Python Connector can be used to connect to Cloud SQL instancesusing both public and private IP addresses, as well asPrivate Service Connect (PSC). To specify which IP address type to connectwith, set theip_type keyword argument when initializing aConnector() or whencallingconnector.connect().

Possible values forip_type are"public" (default value),"private", and"psc".

Example:

conn=connector.connect("project:region:instance","pymysql",ip_type="private"# use private IP...insertotherkwargs ...)

Connections usingAutomatic IAM database authentication are supported when using Postgres or MySQL drivers.First, make sure toconfigure your Cloud SQL Instance to allow IAM authenticationandadd an IAM database user.

Now, you can connect using user or service account credentials instead of a password.In the call to connect, set theenable_iam_auth keyword argument to true and theuser argument to the appropriately formatted IAM principal.

Postgres: For an IAM user account, this is the user's email address. For a service account, it is the service account's email without the.gserviceaccount.com domain suffix.

MySQL: For an IAM user account, this is the user's email address, without the @ or domain name. For example, fortest-user@gmail.com, set theuser argument totest-user. For a service account, this is the service account's email address without the@project-id.iam.gserviceaccount.com suffix.

Example:

conn=connector.connect("project:region:instance","pg8000",user="postgres-iam-user@gmail.com",db="my-db-name",enable_iam_auth=True, )

Active Directory authentication for SQL Server instances is currently only supported on Windows.First, make sure to followthese stepsto set up a Managed AD domain and join your Cloud SQL instance to the domain.See here for more info on Cloud SQL Active Directory integration.

Once you have followed the steps linked above, you can run the following code to return a connection object:

conn=connector.connect("project:region:instance","pytds",db="my-db-name",active_directory_auth=True,server_name="public.[instance].[location].[project].cloudsql.[domain]",)

Or, if using Private IP:

conn=connector.connect("project:region:instance","pytds",db="my-db-name",active_directory_auth=True,server_name="private.[instance].[location].[project].cloudsql.[domain]",ip_type="private")

The connector can be configured to use DNS to look up an instance. This wouldallow you to configure your application to connect to a database instance, andcentrally configure which instance in your DNS zone.

Add a DNS TXT record for the Cloud SQL instance to aprivate DNS serveror a private Google Cloud DNS Zone used by your application.

For example: suppose you wanted to use the domain nameprod-db.mycompany.example.com to connect to your database instancemy-project:region:my-instance. You would create the following DNS record:

• Record type:TXT
• Name:prod-db.mycompany.example.com – This is the domain name used by the application
• Value:my-project:my-region:my-instance – This is the Cloud SQL instance connection name

Configure the connector to resolve DNS names by initializing it withresolver=DnsResolver and replacing the instance connection name with the DNSname inconnector.connect:

fromgoogle.cloud.sql.connectorimportConnector,DnsResolverimportpymysqlimportsqlalchemy# initialize Cloud SQL Python Connector with `resolver=DnsResolver`withConnector(resolver=DnsResolver)asconnector:# initialize SQLAlchemy connection pool with Connectorpool=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"        ),    )# ... use SQLAlchemy engine normally

When the connector is configured using a domain name, the connector willperiodically check if the DNS record for an instance changes. When the connectordetects that the domain name refers to a different instance, the connector willclose all open connections to the old instance. Subsequent connection attemptswill be directed to the new instance.

For example: suppose application is configured to connect using thedomain nameprod-db.mycompany.example.com. Initially the private DNSzone has a TXT record with the valuemy-project:region:my-instance. Theapplication establishes connections to themy-project:region:my-instanceCloud SQL instance.

Then, to reconfigure the application to use a different databaseinstance, change the value of theprod-db.mycompany.example.com DNS recordfrommy-project:region:my-instance tomy-project:other-region:my-instance-2

The connector inside the application detects the change to thisDNS record. Now, when the application connects to its database using thedomain nameprod-db.mycompany.example.com, it will connect to themy-project:other-region:my-instance-2 Cloud SQL instance.

The connector will automatically close all existing connections tomy-project:region:my-instance. This will force the connection pools toestablish new connections. Also, it may cause database queries in progressto fail.

The connector will poll for changes to the DNS name every 30 seconds by default.You may configure the frequency of the connections using the Connector'sfailover_period argument (i.e.Connector(failover_period=60). When this isset to 0, the connector will disable polling and only check if the DNS recordchanged when it is creating a new connection.

The Python Connector can be used alongside popular Python web frameworks suchas Flask, FastAPI, etc, to integrate Cloud SQL databases within yourweb applications.

Flask-SQLAlchemyis an extension forFlaskthat adds support forSQLAlchemy to yourapplication. It aims to simplify using SQLAlchemy with Flask by providinguseful defaults and extra helpers that make it easier to accomplishcommon tasks.

You can configure Flask-SQLAlchemy to connect to a Cloud SQL database fromyour web application through the following:

fromflaskimportFlaskfromflask_sqlalchemyimportSQLAlchemyfromgoogle.cloud.sql.connectorimportConnector# initialize Python Connector objectconnector=Connector()app=Flask(__name__)# configure Flask-SQLAlchemy to use Python Connectorapp.config['SQLALCHEMY_DATABASE_URI']="postgresql+pg8000://"app.config['SQLALCHEMY_ENGINE_OPTIONS']= {"creator":lambda:conn=connector.connect("project:region:instance-name",# Cloud SQL Instance Connection Name"pg8000",user="my-user",password="my-password",db="my-database",ip_type="public"# "private" for private IP    )}# initialize the app with the extensiondb=SQLAlchemy()db.init_app(app)

For more details on how to use Flask-SQLAlchemy, check out theFlask-SQLAlchemy Quickstarts

FastAPI is a modern, fast (high-performance),web framework for building APIs with Python based on standard Python type hints.

You can configure FastAPI to connect to a Cloud SQL database fromyour web application usingSQLAlchemy ORMthrough the following:

importsqlalchemyfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmakerfromgoogle.cloud.sql.connectorimportConnector# initialize Cloud SQL Python Connectorconnector=Connector()# create connection pool engineengine=sqlalchemy.create_engine("postgresql+pg8000://",creator=lambda:connector.connect("project:region:instance-name",# Cloud SQL Instance Connection Name"pg8000",user="my-user",password="my-password",db="my-database",ip_type="public"# "private" for private IP    ),)# create SQLAlchemy ORM sessionSessionLocal=sessionmaker(autocommit=False,autoflush=False,bind=engine)Base=declarative_base()

To learn more about integrating a database into your FastAPI application,follow along theFastAPI SQL Database guide.

The Cloud SQL Connector is compatible withasyncio to improve the speedand efficiency of database connections through concurrency. You can use allnon-asyncio drivers through theConnector.connect_async function, in additionto the following asyncio database drivers:

• asyncpg (Postgres)

The Cloud SQL Connector has a helpercreate_async_connector function that isrecommended for asyncio database connections. It returns aConnectorobject that uses the current thread's running event loop. This is differentthanConnector() which by default initializes a new event loop in abackground thread.

Thecreate_async_connector allows all the same input arguments as theConnector object.

Once aConnector object is returned bycreate_async_connector you can callitsconnect_async method, just as you would theconnect method:

importasyncpgfromgoogle.cloud.sql.connectorimportConnector,create_async_connectorasyncdefmain():# initialize Connector object for connections to Cloud SQLconnector=create_async_connector()# creation function to generate asyncpg connections as the 'connect' argasyncdefgetconn(instance_connection_name,**kwargs)->asyncpg.Connection:returnawaitconnector.connect_async(instance_connection_name,"asyncpg",user="my-user",password="my-password",db="my-db",**kwargs,# ... additional asyncpg args        )# initialize connection poolpool=awaitasyncpg.create_pool("my-project:my-region:my-instance",connect=getconn    )# acquire connection and query Cloud SQL databaseasyncwithpool.acquire()asconn:res=awaitconn.fetch("SELECT NOW()")# close Connectorawaitconnector.close_async()

importsqlalchemyfromsqlalchemy.ext.asyncioimportAsyncEngine,create_async_enginefromgoogle.cloud.sql.connectorimportConnector,create_async_connectorasyncdefmain():# initialize Connector object for connections to Cloud SQLconnector=awaitcreate_async_connector()# The Cloud SQL Python Connector can be used along with SQLAlchemy using the# 'async_creator' argument to 'create_async_engine'pool=create_async_engine("postgresql+asyncpg://",async_creator=lambda:connector.connect_async("project:region:instance",# Cloud SQL instance connection name"asyncpg",user="my-user",password="my-password",db="my-db-name"# ... additional database driver args        ),    )# example queryasyncwithpool.connect()asconn:awaitconn.execute(sqlalchemy.text("SELECT NOW()"))# close Connectorawaitconnector.close_async()# dispose of connection poolawaitpool.dispose()

For more details on additional database arguments with anasyncpg.Connection, please visit theofficial documentation.

An alternative to using thecreate_async_connector function is initializingaConnector as an async context manager, removing the need for explicitcalls toconnector.close_async() to cleanup resources.

importasyncpgfromgoogle.cloud.sql.connectorimportConnector,create_async_connectorasyncdefmain():# initialize Connector object for connections to Cloud SQLloop=asyncio.get_running_loop()asyncwithConnector(loop=loop)asconnector:# creation function to generate asyncpg connections as the 'connect' argasyncdefgetconn(instance_connection_name,**kwargs)->asyncpg.Connection:returnawaitconnector.connect_async(instance_connection_name,"asyncpg",user="my-user",password="my-password",db="my-db",**kwargs,# ... additional asyncpg args            )# create connection poolpool=awaitasyncpg.create_pool("my-project:my-region:my-instance",connect=getconn        )# acquire connection and query Cloud SQL databaseasyncwithpool.acquire()asconn:res=awaitconn.fetch("SELECT NOW()")

importasyncioimportasyncpgimportsqlalchemyfromsqlalchemy.ext.asyncioimportAsyncEngine,create_async_enginefromgoogle.cloud.sql.connectorimportConnectorasyncdefmain():# initialize Connector object for connections to Cloud SQLloop=asyncio.get_running_loop()asyncwithConnector(loop=loop)asconnector:# The Cloud SQL Python Connector can be used along with SQLAlchemy using the# 'async_creator' argument to 'create_async_engine'pool=create_async_engine("postgresql+asyncpg://",async_creator=lambda:connector.connect_async("project:region:instance",# Cloud SQL instance connection name"asyncpg",user="my-user",password="my-password",db="my-db-name"# ... additional database driver args            ),        )# example queryasyncwithpool.connect()asconn:awaitconn.execute(sqlalchemy.text("SELECT NOW()"))# dispose of connection poolawaitpool.dispose()

The Cloud SQL Python Connector uses the standardPython logging modulefor debug logging support.

Add the below code to your application to enable debug logging with the Cloud SQLPython Connector:

importlogginglogging.basicConfig(format="%(asctime)s [%(levelname)s]: %(message)s")logger=logging.getLogger(name="google.cloud.sql.connector")logger.setLevel(logging.DEBUG)

For more details on configuring logging, please refer to thePython logging docs.

This project usessemantic versioning, and uses thefollowing lifecycle regarding support for a major version:

Active - Active versions get all new features and security fixes (thatwouldn’t otherwise introduce a breaking change). New major versions areguaranteed to be "active" for a minimum of 1 year.Deprecated - Deprecated versions continue to receive security and criticalbug fixes, but do not receive new features. Deprecated versions will be publiclysupported for 1 year.Unsupported - Any major version that has been deprecated for >=1 year isconsidered publicly unsupported.

We follow thePython Version Support Policy used by Google CloudLibraries for Python. Changes in supported Python versions will beconsidered a minor change, and will be listed in the release notes.

This project aims for a minimum monthly release cadence. If no newfeatures or fixes have been added, a new PATCH version with the latestdependencies is released.

We welcome outside contributions. Please see ourContributing Guide for details on how best to contribute.