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

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

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

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

Hint

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

Practice

1. What is the main purpose of using a get_db function in FastAPI when working with databases?

easy

A. To create and close a database session for each request safely

B. To store user data permanently in memory

C. To handle HTTP requests directly without a database

D. To generate HTML templates for responses

Solution

1. Step 1: Understand the role of get_db

   The get_db function is designed to open a database session when a request starts and close it when the request ends.

2. Step 2: Recognize safe database session management

   This ensures that each request has its own session, preventing conflicts and resource leaks.

3. Final Answer:

   To create and close a database session for each request safely -> Option A

4. Quick Check:

   Database session management = create and close session [OK]

Hint: Remember: get_db opens and closes sessions per request [OK]

Common Mistakes:

• Thinking get_db stores data permanently
• Confusing get_db with HTTP request handling
• Assuming get_db generates HTML

2. Which of the following is the correct way to declare a dependency for a database session in a FastAPI route using Depends?

easy

A. def read_items(db = Depends(Session)):

B. def read_items(db: get_db = Session()):

C. def read_items(db: Session = get_db()):

D. def read_items(db: Session = Depends(get_db)):

Solution

1. Step 1: Understand FastAPI dependency injection syntax

   FastAPI uses Depends to inject dependencies like database sessions into route functions.

2. Step 2: Correct syntax for session injection

   The correct syntax is to type hint the parameter as Session and assign it Depends(get_db) to call the dependency function.

3. Final Answer:

   def read_items(db: Session = Depends(get_db)): -> Option D

4. Quick Check:

   Dependency injection = parameter: Type = Depends(function) [OK]

Hint: Use parameter: Type = Depends(function) for dependencies [OK]

Common Mistakes:

• Calling get_db() directly in parameter default
• Using Depends with a class instead of a function
• Swapping parameter and default values

3. Given this FastAPI route code snippet, what will be the output if the database session is correctly managed?

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

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

@app.get('/items')
def read_items(db: Session = Depends(get_db)):
    items = db.query(Item).all()
    return items

medium

A. An error because the session is not closed

B. An empty list because the query is missing

C. A list of all items from the database

D. A syntax error due to wrong yield usage

Solution

1. Step 1: Analyze the get_db function behavior

   The get_db function creates a session, yields it for use, then closes it safely after the request.

2. Step 2: Understand the route's database query

   The route uses the session to query all Item records and returns them as a list.

3. Final Answer:

   A list of all items from the database -> Option C

4. Quick Check:

   Yielded session + query = list of items [OK]

Hint: Yielded session allows safe query and close after use [OK]

Common Mistakes:

• Assuming session is not closed causing error
• Thinking yield causes syntax error
• Believing query returns empty without data

4. Identify the error in this FastAPI database session management code:

def get_db():
    db = Session()
    yield db
    db.close()

@app.post('/add')
def add_item(item: Item, db: Session = Depends(get_db)):
    db.add(item)
    db.commit()

medium

A. The item parameter should be inside get_db

B. The session is closed after yield, so it may not close if an exception occurs

C. The Depends is used incorrectly in the route

D. The db.commit() is missing

Solution

1. Step 1: Review session closing in get_db

   The db.close() is called after yield without a try-finally block, so if an exception happens, the session may never close.

2. Step 2: Understand proper session cleanup

   Using try-finally ensures the session closes even if errors occur during request handling.

3. Final Answer:

   The session is closed after yield, so it may not close if an exception occurs -> Option B

4. Quick Check:

   Session close needs try-finally for safety [OK]

Hint: Always use try-finally to close sessions safely [OK]

Common Mistakes:

• Ignoring try-finally for session cleanup
• Forgetting to commit changes
• Misplacing Depends usage

5. You want to ensure that your FastAPI app's database sessions are properly managed and that any changes are committed only if no exceptions occur. Which of the following get_db implementations best achieves this?

hard

A. def get_db():
db = Session()
try:
yield db
db.commit()
except:
db.rollback()
raise
finally:
db.close()

B. def get_db():
db = Session()
yield db
db.commit()
db.close()

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

D. def get_db():
db = Session()
yield db
db.rollback()
db.close()

Solution

1. Step 1: Understand transaction management needs

   We want to commit changes only if no errors occur, otherwise rollback to avoid partial changes.

2. Step 2: Analyze each get_db implementation

   def get_db(): db = Session() try: yield db db.commit() except: db.rollback() raise finally: db.close() uses try-except-finally to commit on success, rollback on error, and always close the session, which is the safest approach.

3. Final Answer:

   def get_db(): db = Session() try: yield db db.commit() except: db.rollback() raise finally: db.close() -> Option A

4. Quick Check:

   Commit on success, rollback on error, always close [OK]

Hint: Use try-except-finally to commit, rollback, and close sessions [OK]

Common Mistakes:

• Committing after yield without error handling
• Not rolling back on exceptions
• Closing session without try-finally