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 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".

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.

To connect to Cloud SQL using the connector, inititalize aConnectorobject and call it'sconnect 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()# function to return the database connectiondefgetconn()->pymysql.connections.Connection:conn:pymysql.connections.Connection=connector.connect("project:region:instance","pymysql",user="my-user",password="my-password",db="my-db-name"    )returnconn# create connection poolpool=sqlalchemy.create_engine("mysql+pymysql://",creator=getconn,)

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 it'sclose() method as follows:

connector.close()

Note: For more examples of using SQLAlchemy to manage connection pooling with the connector, please seeCloud SQL SQLAlchemy Samples.

Note for SQL Server users: If your SQL Server instance requires SSL, you need to download the CA certificate for your instance and includecafile={path to downloaded certificate} andvalidate_host=False. This is a workaround for aknown issue.

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,IPTypes# Note: all parameters below are optionalconnector=Connector(ip_type=IPTypes.PUBLIC,enable_iam_auth=False,timeout=30,credentials=custom_creds# google.auth.credentials.Credentials)

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.connectorimportConnectorimportpymysqlimportsqlalchemy# helper function to return SQLAlchemy connection pooldefinit_connection_pool(connector:Connector)->sqlalchemy.engine.Engine:# function used to generate database connectiondefgetconn()->pymysql.connections.Connection:conn=connector.connect("project:region:instance","pymysql",user="my-user",password="my-password",db="my-db-name"        )returnconn# create connection poolpool=sqlalchemy.create_engine("mysql+pymysql://",creator=getconn,    )returnpool# initialize Cloud SQL Python Connector as context managerwithConnector()asconnector:# initialize connection poolpool=init_connection_pool(connector)# 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 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 areIPTypes.PUBLIC (default value),IPTypes.PRIVATE, andIPTypes.PSC.

Example:

fromgoogle.cloud.sql.connectorimportIPTypesconn=connector.connect("project:region:instance","pymysql",ip_type=IPTypes.PRIVATE# use private IP...insertotherkwargs ...)

Note: If specifying Private IP or Private Service Connect, your application must beattached to the proper VPC network to connect to your Cloud SQL instance. For mostapplications this will require the use of aVPC Connector.

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 steps to 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=IPTypes.PRIVATE)

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,IPTypes# initialize Python Connector objectconnector=Connector()# Python Connector database connection functiondefgetconn():conn=connector.connect("project:region:instance-name",# Cloud SQL Instance Connection Name"pg8000",user="my-user",password="my-password",db="my-database",ip_type=IPTypes.PUBLIC# IPTypes.PRIVATE for private IP    )returnconnapp=Flask(__name__)# configure Flask-SQLAlchemy to use Python Connectorapp.config['SQLALCHEMY_DATABASE_URI']="postgresql+pg8000://"app.config['SQLALCHEMY_ENGINE_OPTIONS']= {"creator":getconn}# 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:

fromsqlalchemyimportcreate_enginefromsqlalchemy.engineimportEnginefromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmakerfromgoogle.cloud.sql.connectorimportConnector,IPTypes# helper function to return SQLAlchemy connection pooldefinit_connection_pool(connector:Connector)->Engine:# Python Connector database connection functiondefgetconn():conn=connector.connect("project:region:instance-name",# Cloud SQL Instance Connection Name"pg8000",user="my-user",password="my-password",db="my-database",ip_type=IPTypes.PUBLIC# IPTypes.PRIVATE for private IP        )returnconnSQLALCHEMY_DATABASE_URL="postgresql+pg8000://"engine=create_engine(SQLALCHEMY_DATABASE_URL ,creator=getconn    )returnengine# initialize Cloud SQL Python Connectorconnector=Connector()# create connection pool engineengine=init_connection_pool(connector)# 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.

Django is a high-level Python web framework thatencourages rapid development and clean, pragmatic design.

To use the Cloud SQL Connector with Django, you need to create a custom databasebackend which subclasses the existing MySQL backend. (Note, at present Postgresis not supported because Django requires the psycopg2 driver, which is notcurrently compatible with the connector.)

Create acloudsql directory in your project with a blank__init__.py and abase.py containing the following code:

fromdjango.db.backends.mysqlimportbasefromgoogle.cloud.sql.connectorimportConnectorclassDatabaseWrapper(base.DatabaseWrapper):defget_new_connection(self,conn_params):returnConnector().connect(**conn_params)

Then in your settings.py file, set yourDATABASES setting as follows:

DATABASES= {"default": {"ENGINE":"cloudsql","USER":"...","PASSWORD":"...","NAME":"...","OPTIONS": {"driver":"pymysql","instance_connection_string":"project:region:instance"    }}# Needed because Django does not support PyMySQL out of the boximportpymysqlpymysql.install_as_MySQLdb()

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:

importasyncpgimportsqlalchemyfromsqlalchemy.ext.asyncioimportAsyncEngine,create_async_enginefromgoogle.cloud.sql.connectorimportConnector,create_async_connectorasyncdefinit_connection_pool(connector:Connector)->AsyncEngine:# initialize Connector object for connections to Cloud SQLasyncdefgetconn()->asyncpg.Connection:conn:asyncpg.Connection=awaitconnector.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        )returnconn# 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=getconn,    )returnpoolasyncdefmain():# initialize Connector object for connections to Cloud SQLconnector=awaitcreate_async_connector()# initialize connection poolpool=awaitinit_connection_pool(connector)# 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.

Note: This alternative requires that the running event loop bepassed in as theloop argument toConnector().

importasyncioimportasyncpgimportsqlalchemyfromsqlalchemy.ext.asyncioimportAsyncEngine,create_async_enginefromgoogle.cloud.sql.connectorimportConnectorasyncdefinit_connection_pool(connector:Connector)->AsyncEngine:# initialize Connector object for connections to Cloud SQLasyncdefgetconn()->asyncpg.Connection:conn:asyncpg.Connection=awaitconnector.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            )returnconn# 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=getconn,    )returnpoolasyncdefmain():# initialize Connector object for connections to Cloud SQLloop=asyncio.get_running_loop()asyncwithConnector(loop=loop)asconnector:# initialize connection poolpool=awaitinit_connection_pool(connector)# example queryasyncwithpool.connect()asconn:awaitconn.execute(sqlalchemy.text("SELECT NOW()"))# dispose of connection poolawaitpool.dispose()

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.