Database session management in FastAPI - Mini Project: Build & Apply

Database Session Management with FastAPI

📖 Scenario: You are building a simple FastAPI app that connects to a database. To keep your app organized and efficient, you need to manage database sessions properly. This means creating a session to talk to the database, using it for queries, and then closing it when done.

🎯 Goal: Build a FastAPI app that sets up a database session using SQLAlchemy, creates a session dependency, and uses it in a route to fetch data.

📋 What You'll Learn

Create a SQLAlchemy SessionLocal class for database sessions

Create a get_db function that yields a database session and closes it after use

Use Depends(get_db) in a FastAPI route to access the database session

Return a simple JSON response using the database session

💡 Why This Matters

🌍 Real World

Managing database sessions is essential in web apps to ensure efficient and safe database access without leaks or conflicts.

💼 Career

Understanding session management with FastAPI and SQLAlchemy is a key skill for backend developers working with Python web frameworks.

1

Setup SQLAlchemy SessionLocal

Import sessionmaker and create_engine from sqlalchemy. Create an engine with the URL sqlite:///./test.db. Then create a SessionLocal class using sessionmaker with autocommit=False, autoflush=False, and bind it to the engine.

FastAPI

# Your code here to create engine and SessionLocal

Need a hint?

Use create_engine with the SQLite URL and sessionmaker with the right parameters.

2

Create get_db Dependency

Define a function called get_db that creates a session instance from SessionLocal(). Use a try block to yield the session, and in the finally block, close the session.

FastAPI

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    # Your code here

Need a hint?

Use yield inside a try block and close the session in finally.

3

Create FastAPI App and Route

Import FastAPI and Depends from fastapi. Create an app instance called app. Define a GET route at "/items/" with a function read_items that takes a parameter db using Depends(get_db). Inside the function, return a dictionary with key "message" and value "Database session is ready".

FastAPI

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI, Depends

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app = FastAPI()

@app.get("/items/")
def read_items(db = Depends(get_db)):
    # Your code here

Need a hint?

Create the app instance and use Depends(get_db) in the route function parameter.

4

Complete and Run the FastAPI App

Add the code to run the FastAPI app using if __name__ == "__main__" block. Inside it, import uvicorn and call uvicorn.run with "__main__:app", host "127.0.0.1", and port 8000.

FastAPI

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI, Depends

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

app = FastAPI()

@app.get("/items/")
def read_items(db = Depends(get_db)):
    return {"message": "Database session is ready"}

if __name__ == "__main__":
    # Your code here

Need a hint?

Use the standard Python entry point check and call uvicorn.run with the app path and server details.