DEV Community


FastAPI Startup and Shutdown Events

fastapitutorial profile image FastapiTutorial-com Updated on ・2 min read

Alt Text

"Currently in our start_application function in, 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.
asyncpg  #For postgres async support
Enter fullscreen mode Exit fullscreen mode

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 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()
Enter fullscreen mode Exit fullscreen mode

and our check_db_connected function in db > would be:

import databases
from db.session import SQLALCHEMY_DATABASE_URL

async def check_db_connected():
        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.
Enter fullscreen mode Exit fullscreen mode

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.
Alt Text

Discussion (0)

Editor guide