Code Summary
Below is all of the critical code in this article for those who don’t actually want to read anything:
For local deployment
# Import the connector helper
from google.cloud.sql.connector import Connector, IPTypes
# This is different from the connector but is useful for using SQL in FLask
import sqlalchemy
# initialize Connector object
connector = Connector()
# function to return the database connection
def getconn() -> pymysql.connections.Connection:
conn: pymysql.connections.Connection = connector.connect(
"example-database-340201:us-central1:example-instance",
"pymysql",
user="example-user",
password="example-password",
db="referral_database"
)
return conn
# This is an example function that assumes your database has a table called 'users' with three columns
def get_users():
pool = getconn()
with pool.cursor() as db_conn:
# query database
query = "SELECT * from users"
db_conn.execute(query)
result = db_conn.fetchall()
# Do something with the results
objects_list = []
for row in result:
d = collections.OrderedDict()
d['name'] = row[0]
d['email'] = row[1]
d['phone'] = row[2]
objects_list.append(d)
db_conn.close()
return jsonify(objects_list)For Deployment from Google App Engine
# Declare your database variables (these can be found in the Cloud Console)
db_user = [your username]
db_password = [your password]
db_name = [your database name]
instance_connection_name = [your instance connection] # ie: 'test-database-340201:us-central1:test-database'
db_socket_dir = "/cloudsql"
# Define get connection function
def getconn():
db_config = {
# [START cloud_sql_mysql_sqlalchemy_limit]
# Pool size is the maximum number of permanent connections to keep.
"pool_size": 1,
# Temporarily exceeds the set pool_size if no connections are available.
"max_overflow": 0,
# The total number of concurrent connections for your application will be
# a total of pool_size and max_overflow.
# [END cloud_sql_mysql_sqlalchemy_limit]
# [START cloud_sql_mysql_sqlalchemy_backoff]
# SQLAlchemy automatically uses delays between failed connection attempts,
# but provides no arguments for configuration.
# [END cloud_sql_mysql_sqlalchemy_backoff]
# [START cloud_sql_mysql_sqlalchemy_timeout]
# 'pool_timeout' is the maximum number of seconds to wait when retrieving a
# new connection from the pool. After the specified amount of time, an
# exception will be thrown.
"pool_timeout": 30, # 30 seconds
# [END cloud_sql_mysql_sqlalchemy_timeout]
# [START cloud_sql_mysql_sqlalchemy_lifetime]
# 'pool_recycle' is the maximum number of seconds a connection can persist.
# Connections that live longer than the specified amount of time will be
# reestablished
"pool_recycle": 1800, # 30 minutes
# [END cloud_sql_mysql_sqlalchemy_lifetime]
}
print("UNIX")
return init_unix_connection_engine(db_config)
def init_unix_connection_engine(db_config):
# [START cloud_sql_mysql_sqlalchemy_create_socket]
# Remember - storing secrets in plaintext is potentially unsafe. Consider using
# something like https://cloud.google.com/secret-manager/docs/overview to help keep
# secrets secret.
# pool = sqlalchemy.create_engine("mysql+pymysql://user1:str%48SSy@/nft_valuation_database?unix_socket=/cloudsql/nft-valuation:us-central1:nft-valuation:3306")
pool = sqlalchemy.create_engine(
sqlalchemy.engine.url.URL.create(
drivername="mysql+pymysql",
username=db_user, # e.g. "my-database-user"
password=db_password, # e.g. "my-database-password"
database=db_name, # e.g. "my-database-name",
port="3306",
query={
"unix_socket": "{}/{}".format(
db_socket_dir, # e.g. "/cloudsql"
instance_connection_name) # i.e "<PROJECT-NAME>:<INSTANCE-REGION>:<INSTANCE-NAME>"
}
),
**db_config
)
print(type(pool))
return pool
# This is an example function that assumes your database has a table called 'users' with three columns
def get_users():
pool = getconn()
with pool.connect() as db_conn:
# query database
query = "SELECT * from users"
result = db_conn.execute(query).fetchall()
# Do something with the results
objects_list = []
for row in result:
d = collections.OrderedDict()
d['name'] = row[0]
d['email'] = row[1]
d['phone'] = row[2]
objects_list.append(d)
db_conn.close()
return jsonify(objects_list)What is Flask?
Flask is a lightweight Python web design library that boasts: “Web design, one drop at a time”.
Flask is a popular alternative to Python’s Django which is a much less customizable and more heavy-weight library for web design. The advantage of Flask is that the developer has much more granular control over the web app and what is included or not. Whereas Django will include many modules and libraries that the user may or may not use.
Flask is also well-documented and is suitable for beginner developers as they will get an appreciation of the different aspects of web design.
What is a Web API (Backend)?
Sometimes people refer to a web API as a backend, but often these are the exact same thing. A Web API (Application Programing Interface) is essentially a collection of nodes or endpoints that perform some function or subroutines when accessed.
Often, a web app backend (or Web API) is primarily used to securely communicate with a database. This includes reading, writing, editing, and deleting data. Communication is performed via GET and POST requests using XML or JSON data formatting. Some data processing and security checks are typically written into a Web API as well.
Why Use a Web API?
A Web API is useful as it can be compatible with many different front-end applications despite the operating system. A Web API is also very useful for implementing the security features necessary for maintaining a web service or maintaining data security for users.
Basically, if users will request or manipulate data stored in a server or database, you will need to use a backend or web API.
Additionally, there may be processes that you don’t want to run in a user’s browser. Read up more about the popularity and benefits of RESTful APIs here.
How to Connect a Python-Flask Backend to a Google Cloud SQL Instance
Google Cloud SQL Connector Helper
The Google Cloud SQL Connector Helper is a handy way to quickly and easily connect to a Cloud SQL Instance via encryption using Python or Java.
Please note the following:
- You will need to enable the Cloud SQL Admin API
- The database must already be created
- If you deploy an app to Google App Engine then this connector will no longer be usable.
- You also might need to whitelist the IP address that is accessing the Cloud SQL Instance.
# Import the connector helper
from google.cloud.sql.connector import Connector, IPTypes
# This is different from the connector but is useful for using SQL in FLask
import sqlalchemyYou should initialize the connector object outside of any function (ie globally).
# initialize Connector object
connector = Connector()Connection function that establishes a database connection with a Google Cloud SQL Instance.
# function to return the database connection
def getconn() -> pymysql.connections.Connection:
conn: pymysql.connections.Connection = connector.connect(
"example-database-340201:us-central1:example-instance",
"pymysql",
user="example-user",
password="example-password",
db="referral_database"
)
return connThis is an example function that uses the Google Connector.
# This is an example function that assumes your database has a table called 'users' with three columns
def get_users():
pool = getconn()
with pool.cursor() as db_conn:
# query database
query = "SELECT * from users"
db_conn.execute(query)
result = db_conn.fetchall()
# Do something with the results
objects_list = []
for row in result:
d = collections.OrderedDict()
d['name'] = row[0]
d['email'] = row[1]
d['phone'] = row[2]
objects_list.append(d)
db_conn.close()
return jsonify(objects_list)UNIX Socket Connector (Required if connecting from a Google App Engine Instance)
Once you inevitably deploy your application, you might very well choose to deploy it with Google’s App Engine. You have already chosen to use Google’s Cloud SQL and App Engine is a low-cost, easy way to deploy whatever application you’ve been developing.
However, you should note that you cannot use the Google Cloud SQL Connector Helper once an application is deployed to Google App Engine. This is a nuisance, but not the end of the world.
Let’s look at how to connect to your database from App Engine.
First, let’s declare our configuration variables. You can do so globally if this is a backend application that is not loaded in the browser:
# Declare your database variables (these can be found in the Cloud Console)
db_user = [your username]
db_password = [your password]
db_name = [your database name]
instance_connection_name = [your instance connection] # ie: 'test-database-340201:us-central1:test-database'
db_socket_dir = "/cloudsql"Next, we will define the functions that will create and return our connection:
# Define get connection function
def getconn():
db_config = {
# [START cloud_sql_mysql_sqlalchemy_limit]
# Pool size is the maximum number of permanent connections to keep.
"pool_size": 1,
# Temporarily exceeds the set pool_size if no connections are available.
"max_overflow": 0,
# The total number of concurrent connections for your application will be
# a total of pool_size and max_overflow.
# [END cloud_sql_mysql_sqlalchemy_limit]
# [START cloud_sql_mysql_sqlalchemy_backoff]
# SQLAlchemy automatically uses delays between failed connection attempts,
# but provides no arguments for configuration.
# [END cloud_sql_mysql_sqlalchemy_backoff]
# [START cloud_sql_mysql_sqlalchemy_timeout]
# 'pool_timeout' is the maximum number of seconds to wait when retrieving a
# new connection from the pool. After the specified amount of time, an
# exception will be thrown.
"pool_timeout": 30, # 30 seconds
# [END cloud_sql_mysql_sqlalchemy_timeout]
# [START cloud_sql_mysql_sqlalchemy_lifetime]
# 'pool_recycle' is the maximum number of seconds a connection can persist.
# Connections that live longer than the specified amount of time will be
# reestablished
"pool_recycle": 1800, # 30 minutes
# [END cloud_sql_mysql_sqlalchemy_lifetime]
}
print("UNIX")
return init_unix_connection_engine(db_config)
def init_unix_connection_engine(db_config):
# [START cloud_sql_mysql_sqlalchemy_create_socket]
# Remember - storing secrets in plaintext is potentially unsafe. Consider using
# something like https://cloud.google.com/secret-manager/docs/overview to help keep
# secrets secret.
# pool = sqlalchemy.create_engine("mysql+pymysql://user1:str%48SSy@/nft_valuation_database?unix_socket=/cloudsql/nft-valuation:us-central1:nft-valuation:3306")
pool = sqlalchemy.create_engine(
sqlalchemy.engine.url.URL.create(
drivername="mysql+pymysql",
username=db_user, # e.g. "my-database-user"
password=db_password, # e.g. "my-database-password"
database=db_name, # e.g. "my-database-name",
port="3306",
query={
"unix_socket": "{}/{}".format(
db_socket_dir, # e.g. "/cloudsql"
instance_connection_name) # i.e "<PROJECT-NAME>:<INSTANCE-REGION>:<INSTANCE-NAME>"
}
),
**db_config
)
print(type(pool))
return poolOk! Now we are ok to call the connection from our functions. Note that pool.cursor() is not used, but pool.connect() is.
Let’s look at an example function that does exactly this:
# This is an example function that assumes your database has a table called 'users' with three columns
def get_users():
pool = getconn()
with pool.connect() as db_conn:
# query database
query = "SELECT * from users"
result = db_conn.execute(query).fetchall()
# Do something with the results
objects_list = []
for row in result:
d = collections.OrderedDict()
d['name'] = row[0]
d['email'] = row[1]
d['phone'] = row[2]
objects_list.append(d)
db_conn.close()
return jsonify(objects_list)Conclusion
There are a variety of ways to connect to a Google Cloud SQL Instance that are more fully described in the documentation provided by Google. If you want to dig deeper or have a situation that doesn’t exactly match what is shown here, then check it out.
Thanks for reading and look out for more content from RTL Coding. Hope this article was useful and if you would like to support this blog and this content feel free to donate via Paypal to help support more helpful content.
Sources
- MySQL SQL Dump Options and Examples Documentation
- Amazon Web Service (AWS) SQL Dump Docs
- Microsoft Azure SQL Dump Docs