FastAPI Startup and Shutdown Events

"Currently in our start_application function in main.py, We are just trusting that our database would be connected and in case the required tables are missing then those tables will be created." But as a developer it is our responsibility to handle the corner cases. We should verify database accessibility as well as if we can execute a query or not.

Let's Implement a logic that runs every time our server reloads and verifies if the database connection is working or not. We are adopting some async logic from now on, lets add these 2 modules in our requirements.txt file and do a pip install -r requirements.txt.

#Databases gives asyncio support for a range of databases.
databases
asyncpg  #For postgres async support

Now, We will be creating a startup event, This event runs after our app is initialized. Since we are making an async function so we need to add async keyword before the def keyword and we need to await the results of async tasks inside this function. In very simple words async tasks are used to resolve blocking requests. It's like if you are copying 80GB of files from your laptop to hard disk, In the meantime you can do other tasks also,You need not sit down and wait for copying to complete. Let's put this async function in our main.py file.

...
from fastapi.staticfiles import StaticFiles
from db.utils import check_db_connected             #new
...


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


app = start_application()


@app.on_event("startup")              #new
async def app_startup():
    await check_db_connected()

and our check_db_connected function in db > utils.py would be:

import databases
from db.session import SQLALCHEMY_DATABASE_URL

async def check_db_connected():
    try:
        if not str(SQLALCHEMY_DATABASE_URL).__contains__('sqlite'):
            database = databases.Database(SQLALCHEMY_DATABASE_URL)
            if not database.is_connected:
                await database.connect()
                await database.execute("SELECT 1")
        print("Database is connected (^_^)")
    except Exception as e:
        print("Looks like there is some problem in connection,see below traceback")
        raise e

# Note for sqlite it might not work, reason:
# For newer version of sqlalchemy (1.3.24 +), We might get 'RowProxy ImportError', It is not recommended to move backward as sqlalchemy newer versions are supporting async while 1.3.* versions are not async compatible.

Done, now every time our unicorn server reloads, the startup event would check if the database is connected or not as well as if we can execute a query or not? See the below image for reference.