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 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 >


async def check_db_disconnected():  #new
        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.


Next: Startup Events

Brige the gap between Tutorial hell and Industry. We want to bring in the culture of Clean Code, Test Driven Development.

We know, we might make it hard for you but definitely worth the efforts.



Refund Policy

Follow us on our social media channels to stay updated.

© Copyright 2022-23 Team FastAPITutorial