Last active
July 5, 2023 16:06
-
-
Save dhaninugraha/a880a222bc5822ffdd3a0b5a98b96e01 to your computer and use it in GitHub Desktop.
Multiple databases in Flask
This file contains hidden or 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 flask import Flask, render_template, redirect, request, url_for, jsonify | |
from flask_sqlalchemy import SQLAlchemy | |
from sqlalchemy import text | |
app = Flask(__name__) | |
app.config["DEBUG"] = False | |
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False | |
""" | |
My PostgreSQL 10 installation listens on 5434, while 9.6 listens on 5432. | |
On both installations, I set user postgres to use 'trust' authentication for localhost only. | |
Note that SQLALCHEMY_DATABASE_URI still has to be configured. | |
""" | |
app.config["SQLALCHEMY_DATABASE_URI"] = "postgresql://postgres@localhost:5434/test" | |
app.config["SQLALCHEMY_BINDS"] = { | |
"db_pg10": "postgresql://postgres@localhost:5434/test", | |
"db_pg9_6": "postgresql://postgres@localhost:5432/test" | |
} | |
db = SQLAlchemy() | |
db.init_app(app) | |
""" | |
This is where the magic lies. | |
Courtesy of: https://stackoverflow.com/a/41291770 | |
""" | |
e_pg10 = db.get_engine(app, "db_pg10") | |
e_pg9_6 = db.get_engine(app, "db_pg9_6") | |
@app.route("/") | |
def root(): | |
resp = { | |
"data_from_pg10": "", | |
"data_from_pg9_6": "", | |
"error": "" | |
} | |
try: | |
res_from_pg10 = e_pg10.execute(text("select version();")).fetchone() | |
if res_from_pg10 is not None: | |
resp["data_from_pg10"] = res_from_pg10[0] | |
res_from_pg9_6 = e_pg9_6.execute(text("select version();")).fetchone() | |
if res_from_pg9_6 is not None: | |
resp["data_from_pg9_6"] = res_from_pg9_6[0] | |
resp = jsonify(resp) | |
resp.status_code = 200 | |
except Exception, e: | |
resp["error"] = str(e) | |
resp = jsonify(resp) | |
resp.status_code = 500 | |
finally: | |
db.session.close() | |
return resp | |
if __name__ == "__main__": | |
app.run() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment