Last active
October 26, 2022 01:23
-
-
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
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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