Can the Cloud SQL Python Connector be used with SQLAlchemy?

Question

So I have been trying to deploy an App Engine Standard app using flask, sqlalchemy and Cloud SQL. I believed I had finished all the coding and was ready to deploy when I found out that connecting with the techniques described in here and here simply don't work when deployed on an App Engine instance. So after asking some things, I believe I have been able to stablish a connection between my app and the SQL instance in the Cloud SQL server using the newly released Cloud SQL Python Connector, but I don't how to further use that connection along the SQLAlchemy library.

For example, see my (relevant) code:

config.py

"""Flask configuration."""
from os import environ, path
from dotenv import load_dotenv
from google.cloud.sql.connector import connector

basedir = path.abspath(path.dirname(__file__))
load_dotenv(path.join(basedir, '.env'))

DB_USERNAME = environ.get('DB_USERNAME')
DB_PASSWORD = environ.get('DB_PASSWORD')
DB_ADDRESS = environ.get('DB_ADDRESS')
DB_PORT = environ.get('DB_PORT')
DB_NAME = environ.get('DB_NAME')
DB_CONNECTION_STRING = environ.get('DB_CONNECTION_STRING')
        
class Config:
    """Base config."""
    SECRET_KEY = environ.get('SECRET_KEY')
    STATIC_FOLDER = 'static'
    TEMPLATES_FOLDER = 'templates'
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    connector.connect(
        DB_CONNECTION_STRING, 
        "pymysql",
        user=DB_USERNAME,
        password=DB_PASSWORD,
        db=DB_NAME
    )   

main.py

from flask import Flask, render_template, request, redirect, url_for, flash
from flask_sqlalchemy import SQLAlchemy
import random
import string
import datetime
import json
import re
from config import MAPBOX_TOKEN

#Create App
def create_app():
    app = Flask(__name__)
    app.config.from_object('config.DevConfig')
    return app

app = create_app()
db = SQLAlchemy(app)

# User ORM for SQLAlchemy
class Users(db.Model):
    id = db.Column(db.Integer, primary_key = True, nullable = False)
    ptoken = db.Column(db.String(255), nullable = False, unique = True)
    geodict = db.Column(db.Text, nullable = False)

And while the connector.connect thingy from Google works appropriately, when the code above is run I get the following warning:

..\env\lib\site-packages\flask_sqlalchemy\__init__.py:851: UserWarning: Neither SQLALCHEMY_DATABASE_URI nor SQLALCHEMY_BINDS is set. Defaulting SQLALCHEMY_DATABASE_URI to "sqlite:///:memory:".

To fix this issue I have tried some silly things like:

class Config:
    """Base config."""
    SECRET_KEY = environ.get('SECRET_KEY')
    STATIC_FOLDER = 'static'
    TEMPLATES_FOLDER = 'templates'
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    SQLALCHEMY_DATABASE_URI = connector.connect(
        DB_CONNECTION_STRING, 
        "pymysql",
        user=DB_USERNAME,
        password=DB_PASSWORD,
        db=DB_NAME
    )

But this obviously doesn't work and I just can't find any helpful info online.

So my question is: Can the Cloud SQL Python Connector be used with SQLAlchemy? How can I achieve what I'm trying to do?

Answer 1

What you were looking for is documented here: https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/blob/main/README.md#flask-sqlalchemy

The key information is that you can use SQLALCHEMY_ENGINE_OPTIONS to provide the connector factory function to the SqlAlchemy instance managed by Flask-SqlAlchemy.

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from google.cloud.sql.connector import Connector, IPTypes


# initialize Python Connector object
connector = Connector()

# Python Connector database connection function
def getconn():
    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
    )
    return conn


app = Flask(__name__)

# configure Flask-SQLAlchemy to use Python Connector
app.config['SQLALCHEMY_DATABASE_URI'] = "postgresql+pg8000://"
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
    "creator": getconn
}

# initialize the app with the extension
db = SQLAlchemy()
db.init_app(app)

Answer 2

So first of all, connecting from App Engine to Cloud SQL works just fine. There's an entire page about doing it on in the documentation.

Second, the connector can be used an alternative to that method. Here is an example of it being used with SQL Alchemy.