Shutdown Events

In the previous post, we saw a startup event that verifies if the db is connected as well as if we can execute a query. In this one, we are going to verify that db connection is closed or not on shutdown.
We addon the below lines in main.py file:

..
from db.utils import check_db_connected,check_db_disconnected  #new


def start_application():
    app = FastAPI(title=settings.PROJECT_NAME, version=settings.PROJECT_VERSION)
    ...
    return app


app = start_application()


@app.on_event("shutdown")     #new
async def app_shutdown():
    await check_db_disconnected()

Now, we need to implement the logic for db disconnect. Do you remember we discussed in the previous post an issue that we see for SQLite. SQLite tries to use RowProxy from sqlalchemy but it is removed after version 1.3.24. So, We are going to exclude SQLite for our example. There is workaround for it, We may fork sqlalchemy and make our own modifications and use it but it will become an overhead because we will be stuck with our own version of sqlalchemy and we will need to patch it again and again every time we will need to update our sqlalchemy version. So, I am going to completely exclude sqlite for our shutdown event.
Time to make changes in db > utils.py

...

async def check_db_disconnected():  #new
    try:
        if not str(SQLALCHEMY_DATABASE_URL).__contains__("sqlite"):
            database = databases.Database(SQLALCHEMY_DATABASE_URL)
            if database.is_connected:
                await database.disconnect()
        print("Database is Disconnected (-_-) zZZ")
    except Exception as e:
        raise e

We are all set to test our new feature. Let's try it out.