Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Save qlawmarq/1ece4dac14a998189cf98f1658de0a16 to your computer and use it in GitHub Desktop.
Save qlawmarq/1ece4dac14a998189cf98f1658de0a16 to your computer and use it in GitHub Desktop.
Example of using PyTDS and Cloud SQL Python Connector to connect SQL server of Cloud SQL
from google.cloud.sql.connector import Connector, IPTypes
import pytds
import os
##### SQL connecter #############################################
#################################################################
# Grabbing the user and password from environment variables
db_name = os.getenv("APPLICATION_DB")
db_user = os.getenv("DATABASE_USERNAME")
db_password = os.getenv("DATABASE_PASSWORD")
def get_cloudsql_connection():
ip_type = IPTypes.PRIVATE if os.environ.get("PRIVATE_IP") else IPTypes.PUBLIC
connector = Connector(ip_type)
conn = connector.connect(
# You need to get your DB instance info.
"your-project-name:australia-southeast1:your-project-name",
"pytds",
user=db_user,
password=db_password,
db=db_name,
# You need to download the CA certificate for your DB instance.
# https://issuetracker.google.com/issues/184867147
validate_host=False,
cafile='key/server-ca.pem'
)
conn.autocommit = False
conn.as_dict = False
return conn
def query_get(sql, param):
conn = get_cloudsql_connection()
with conn.cursor() as cur:
cur.execute(sql, param)
results = [row for row in cur.fetchall()]
return results
def test_connection():
id = 1
# You can test any query you like!
test = query_get("SELECT * FROM User WHERE User.Id = %s ",(id,))
return test
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment