Using flask-sqlalchemy and sqlalchemy together, why and is it safe

Question

I have been working on a flask app that is database intensive for about 4 months. I am pretty new to python, flask and sqlalchemy but very familiar with web and database programming.

I mainly use core SQL for the hard database bits but have used the ORM as well. The main methods I use with core SQL are the text() and bindparam() functions to give me database independence via sqlalchemy dialects.

I have found myself having to intermix my module imports to get the stuff I want. Here is a simple but typical import.

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import bindparam
from sqlalchemy.sql import text

The SQLAlchemy object then gets used in the ORM bits of my code and I believe it also handles sessions as per this. It is initialised something like this.

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = appdb
app.config['SQLALCHEMY_BINDS'] = {'meta': foodb, 'data': bardb}
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

I then use the db variable to access all the flask_sqlalchemy stuff.

The direct sqlalchemy stuff gets used in my core SQL and looks something like this.

What I am finding now is that I am intermixing the code, for example like this.

eng = db.get_engine(bind='meta')
cxn = eng.connect()
sql = text('select * from foo') 
rows = cxn.execute(sql).fetchall()

It seems to me that if I have decided to use flask_sqlalchemy I shouldn't have to import sqlalchemy stuff seperately, it should be somewhere in flask_sqlalchemy. Intermixing makes me nervous about side-effects.

So I have two questions.

1. Is intermixing like this safe?
2. Can I get text() and bindparam() and other sqlalchemy stuff from flask_sqlalchemy somehow? If so how?

Answer 1

The following answer to my own question encapsulates some of the stuff I have learned the hard way over the last 4 months.

I am a database programmer and want to use SQLAlchemy core SQL to talk to my databases, not the ORM. The code below is given in that context.

1. Binds and using multiple databases.
2. Getting column names.
3. Handling IN clauses.
4. Engines v sessions.
5. Transactions.

I hope it helps someone.

'''
THE SETUP
'''
maindb = 'dialect+driver://username:password@host:port/main database'
foodb = 'dialect+driver://username:password@host:port/foo database'
bardb = 'dialect+driver://username:password@host:port/bar database'

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = maindb
app.config['SQLALCHEMY_BINDS'] = {'foodb': foodb, 'bardb': bardb}
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)


'''
THE CODE USING AN ENGINE
'''
# Using an engine (dbms = database management system).
dbms = db.get_engine(bind='foodb')
cxn = dbms.connect()
sql = db.text('SELECT * FROM foo WHERE foo_id = :id')
parms = {'id': 4}
rows = cxn.execute(sql, parms)

# Get column names, works for all the examples
print(rows.keys())

# Loop through rows and print the column value
for row in rows:
    print(row[1]) # Access by position
    print(row['foo_name']) # Access by column name
cxn.close

# Providing a list to an IN clause (bindparam with expanding=True)
sql = db.text('select * from foo where foo_id in :ids and foo_name like :nm')
parms = {'ids':[3,4,5], 'nm':'C%'}
for key in parms.keys(): # Apply bindparam to special cases like IN lists
    if type(parms[key]) == type([]): # Parameter value is a list
        sql = sql.bindparams(db.bindparam(key, expanding = True))
rows = cxn.execute(sql, parms)
print([row['foo_name'] for row in rows])

# Do a database transaction with an engine
txn = cxn.begin()
try:      
    sql = db.text('update foo set foo_name = :nm where foo_id = :id')
    parms = {'id': 4, 'nm':'mr foo bar'}
    cxn.execute(sql, parms)
    txn.commit()
except:
    txn.rollback()
    raise
finally:
    if cxn != None:
        cxn.close()

'''
THE CODE USING A SESSION
'''
# Using a session.
dbms = db.get_engine(bind='foodb')
ssn = db.session(bind=dbms)
sql = db.text('SELECT * FROM foo WHERE foo_id = :id')
parms = {'id': 4}
rows = ssn.execute(sql, parms)
print([row['foo_name'] for row in rows])

# Do a database transaction with a session
breakpoint()
try:      
    sql = db.text('update foo set foo_name = :nm where foo_id = :id')
    parms = {'id': 4, 'nm':'ms bar foo'}
    ssn.execute(sql, parms)
    ssn.commit()
except:
    ssn.rollback()
    raise
finally:
    if ssn != None:
        ssn.close()