How do I use flask_sqlalchemy and a GCP TLS protected DB connector?

Question

I'd like to use flask_sqlalchemy with GCP's db connector which does automatic TLS setup as described here: https://cloud.google.com/sql/docs/postgres/connect-connectors#python_1

I tried their example code on my laptop with credentials set via GOOGLE_APPLICATION_CREDENTIALS and it worked. It was able to connect to the remote DB on GCP, I checked that it was TLS encrypted with wireshark, it successfully queried data from the DB. I ran that with sqlalchemy 1.4.47.

I want to use the GCP connector with a flask + flask_sqlalchemy app that uses SQLALCHEMY_BINDS. Using sqlalchemy 1.4.47 flask_sqlalcehmy 2.5.1 and code like this:

...

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

...   

db: SQLAlchemy = SQLAlchemy()

def create_web_app():
    app = Flask('ex')
    ...

    ip_type = IPTypes.PUBLIC
    connector = Connector()

    def getconn() -> pg8000.dbapi.Connection:
        conn: pg8000.dbapi.Connection = connector.connect(
            app.config["GCPTSLPROTECTED_INSTANCE_CONNECTION_NAME"],
            "pg8000",
            user=app.config["GCPTSLPROTECTED_DB_USER"],
            password=app.config["GCPTSLPROTECTED_DB_PASS"],
            db=app.config["GCPTSLPROTECTED_DB_NAME"],
            ip_type=ip_type,
        )
        return conn

    app.config["SQLALCHEMY_BINDS"] = {
          "gcptslprotected": {
             # This URI looks funny but is what GCP docs request and works in the test script
             "url": make_url("postgresql+pg8000://"), 
             "creator": getconn }
    }

    ...
    # SQLALCHEMY_URI for a default db connection also gets created and works fine
    app.config['SQLALCHEMY_URI'] = "sqlite:///./tests/data/test.db"
    
    db.init_app(app)
    ...

The above code runs fine but then a route in the flask app uses the DB this this error is raised:

File "/home/bdc34/.pyenv/versions/browse310/lib/python3.10/site-packages/sqlalchemy/engine/create.py", line 516, in create_engine
u, plugins, kwargs = u._instantiate_plugins(kwargs)
AttributeError: 'dict' object has no attribute '_instantiate_plugins'

You might notice that I use make_url() where as the GCP example script does not have that. I did that to get around an exception raised by flask_sqlalchemy function apply_driver_hacks() which is not the most confidence inspiring name.

Answer 1

I found that flask_sqlalchemy says in the docs that it will passe the dict in the SQLALCHEMY_BINDS to sqlalchemy creat_engine() but it does not seem to correctly do this.

So I looked for a place in flask_sqlalchemy.SQLAlchemy to override something to make it work and I found that set_options calls apply_driver_hacks and that was an okay place.

I did:

class BrowseSQLAlchemy(SQLAlchemy):
    """Overrides how flask_sqlalchemy handles options 
    that need to be passed to create_engine.

    In spite of documentation to the contrary, flask_sqlalchemy 
    does not seem to be able to handle a dict as the value for a 
    SQLALCHEMY_BINDS.
    """

    def apply_driver_hacks(self, app, sa_url, options): # type: ignore
        if not isinstance(sa_url, dict):
            return super().apply_driver_hacks(app, sa_url, options)

        url = make_url(sa_url["url"])
        options.update(sa_url)
        options.pop("url")
        return url, options

 db: SQLAlchemy = BrowseSQLAlchemy()

I came to these conclusions after looking at the flask_sqlalchemy code. It can be deduced that get_options() and apply_driver_hasks() are not intended to accept a Dict for the sa_url.