Falcon SQLAlchemy Integration

api.py

import falcon
import sqlalchemy

from .middleware import SQLAlchemyMiddleware
from .db import session

def create_app(config: dict):
    session.configure(bind=sqlalchemy.create_engine(config.get('DATABASE_URL', 'sqlite://')))
    return falcon.API(middleware=[SQLAlchemyMiddleware(session)])

conftest.py

# This isn't strictly nessecary, but is nice. It will automatically 
# rollback the db for every test. 
# 
# Works for dialects which support nested transactions, like Postgresql

from .db import session
from .app import create_app

@pytest.fixture(scope='function', autouse=True)
def session():
    """Manage the SQLA session for tests"""
    
    # Configure the session, but calling create_app. 
    # You can move this to another fixture, which is rather handy
    app = create_app()
    
    # Get a connection outside of the session
    conn = session.bind.connect()
    
    # Begin a transaction outside of the session scope
    trans = conn.begin()
    
    # Create an instance of a session
    sess = session()
    
    # Start the nested transaction
    sess.begin_nested()

    # Reopen the SAVEPOINT when it closes but listening for when the app tries to commit
    # or rollback.
    @sa.event.listens_for(sess, "after_transaction_end")
    def restart_savepoint_after_trans_end(session, transaction):
        session.expire_all()
        if transaction.nested and not transaction._parent.nested:
            session.begin_nested()

    # Run the test
    yield sess
    
    # Close out the session
    sess.close()
    
    # Rollback the entire transaction
    trans.rollback()

db.py

import sqlalchemy as sa
session = sa.orm.scoping.scoped_session(sa.orm.session.sessionmaker())

middleware.py

class SQLAlchemyMiddleware:
    def __init__(self, session):
        self.session = session

    def process_response(self, req, resp, resource, req_succeeded):
        if req_succeeded:
            self.session.commit()
        else:
            self.session.rollback()