How to make SQLAlchemy engine available throughout the flask application?

Question

Sorry for the long question. I tried to keep it well formatted.

How should i structure my application so that I can have the SQLAlchemy engine available throughout the application especially when I am using Mixins and custom Base class in models directory. Currently i am using something like this:

application/
|── svc/
│  |── service/
│  | └─── service.py
│  | └─── service2.py
│  |── postgres/
│  | └─── models/
│  |  | └──── CustomBase.py
│  |  | └──── User.py
│  |──── connect.py
|── scripts/
│  |── mock_data/
│  | └───── postgres_mock_data.py
│  |──── db.py
│  |──── load_data.py
│  |──── commands.sh
|── wsgi.py

My connect.py looks like this:

def connect_postgres(env_config=config['LOCAL']):
    """
    Function that sends host information to SQLAlchemy
    """

    # Omitted variables for brevity 
    connection_string = f'postgresql://{user}:{password}@{host}/{database}'
    engine = create_engine(connection_string, echo=True)

    return engine

service.py also contains:

app = Flask(__name__)
app.register_blueprint(service2, url_prefix='/endpoint2')

wsgi.py looks like this:

from svc.postgres.connect import connect_postgres
from svc.service.service import app as application
config = configparser.ConfigParser()

def startApp(env):
    connect_postgres(config[env])
    application.run(debug=True)

if __name__ == "__main__":
    startApp('LOCAL')

CustomBase.py looks like:

from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.declarative import declarative_base

class CustomBase(object):
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()

    # Ommited for brevity

Base = declarative_base(cls=CustomBase)

User.py

from sqlalchemy import Column, String
from CustomBase import Base

class User(Base):
     name = Column(String)
     fullname = Column(String)
     nickname = Column(String)

---

I start my application with the command python wsgi.py.

Question:

How would I import engine and session in my service.py where the majority of SQL Queries will be performed. I want to avoid doing the following in service.py because it feels like it is redundant since i have already connected to postgres in the beginning of the application.

from svc.postgres.connect import connect_postgres
engine = connect_postgres(config[env])

I think I know I obviously have to do this in service.py once engine is available:

from CustomBase import Base
from sqlalchemy.orm import sessionmaker

# Create all schemas
Base.metadata.create_all(engine)

# create a configured "Session" class
Session = sessionmaker(bind=engine)

# create a Session
session = Session()

I know it is fairly simple to do the same with flask-sqlalchemy but I am bound to use the SQLAlchemy library.

flask-sqlalchemy code:

db = SQLAlchemy(app)

I can use db in all the inherited blueprint apis.

---

EDIT: Potential answer

Should I just include engine in my CustomBase.py file and do a connection again?

CustomBase.py looks like:

from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.declarative import declarative_base
from svc.postgres.connect import connect_postgres

class CustomBase(object):
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()

    # Ommited for brevity

Base = declarative_base(cls=CustomBase)
engine = connect_postgres(config[env])

service.py

from CustomBase import Base, engine
from User import User
from sqlalchemy.orm import sessionmaker

# Create all schemas
Base.metadata.create_all(engine)

# create a configured "Session" class
Session = sessionmaker(bind=engine)

# create a Session
session = Session()

Although it still feels redundant to call connect_postgres function twice or should I just remove it from my wsgi.py (but i will have to hack my way into config variables at this point)?

Answer 1

If you want to create a db object like Flask-SQLAlchemy provided, you can create it with scoped_session:

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

engine = create_engine('sqlite:////tmp/test.db', convert_unicode=True)
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))

Then use it like this:

>>> from yourapplication.database import db_session
>>> from yourapplication.models import User
>>> u = User('admin', 'admin@localhost')
>>> db_session.add(u)
>>> db_session.commit()

Also register a handler to remove the database session when application end:

from yourapplication.database import db_session

@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()

Check out more on Flask's documentation for SQLAlchemy support.