FastAPI Database Testing
In modern web application development, ensuring your database operations work correctly is crucial. This guide will teach you how to effectively test your FastAPI applications that interact with databases. We'll use SQLAlchemy for database operations and pytest for writing clean, maintainable tests.
Why Test Database Operations?
Testing database operations helps ensure:
- Your data models are correctly defined
- CRUD (Create, Read, Update, Delete) operations work as expected
- Business logic that involves database interactions functions correctly
- Changes to your code don't break existing functionality
Prerequisites
Before diving into testing, you should be familiar with:
- FastAPI basics
- SQLAlchemy with FastAPI
- Basic understanding of pytest
Setting Up a Test Environment
1. Creating a Test Database Configuration
First, let's set up a separate test database configuration:
python
# test_database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# Use SQLite in-memory database for testing
TEST_SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
TEST_SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Dependency for tests
def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
2. Setting Up Test Fixtures with pytest
Pytest fixtures help us set up and tear down test environments:
python
# conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from .main import app
from .database import Base
from .test_database import override_get_db
from .dependencies import get_db
# Create test database
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(scope="function")
def test_db():
# Create the database and tables
Base.metadata.create_all(bind=engine)
yield # Run the tests
# Drop the database tables after the tests run
Base.metadata.drop_all(bind=engine)
@pytest.fixture(scope="function")
def client(test_db):
# Dependency override
app.dependency_overrides[get_db] = override_get_db
with TestClient(app) as test_client:
yield test_client
app.dependency_overrides.clear()
Testing Database Models
Let's start with a simple User model:
python
# models.py
from sqlalchemy import Column, Integer, String, Boolean
from .database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
is_active = Column(Boolean, default=True)
Now, let's test that our model works correctly:
python
# test_models.py
import pytest
from sqlalchemy.orm import Session
from .models import User
def test_create_user(test_db):
db = TestingSessionLocal()
user = User(email="[email protected]", hashed_password="hashedpassword")
db.add(user)
db.commit()
# Query to verify user was created
created_user = db.query(User).filter(User.email == "[email protected]").first()
assert created_user is not None
assert created_user.email == "[email protected]"
assert created_user.is_active == True
db.close()
Testing CRUD Operations
Let's create a test file for our user CRUD operations:
python
# test_crud.py
from fastapi.testclient import TestClient
import pytest
def test_create_user(client):
# Test creating a new user
response = client.post(
"/users/",
json={"email": "[email protected]", "password": "testpassword"}
)
assert response.status_code == 200
data = response.json()
assert data["email"] == "[email protected]"
assert "id" in data
user_id = data["id"]
# Test that we can retrieve the user
response = client.get(f"/users/{user_id}")
assert response.status_code == 200
assert response.json()["email"] == "[email protected]"
assert response.json()["id"] == user_id
def test_get_nonexistent_user(client):
# Test getting a user that doesn't exist
response = client.get("/users/999")
assert response.status_code == 404
def test_update_user(client):
# First create a user
response = client.post(
"/users/",
json={"email": "[email protected]", "password": "testpassword"}
)
user_id = response.json()["id"]
# Now update the user
response = client.put(
f"/users/{user_id}",
json={"email": "[email protected]", "is_active": False}
)
assert response.status_code == 200
data = response.json()
assert data["email"] == "[email protected]"
assert data["is_active"] == False
def test_delete_user(client):
# First create a user
response = client.post(
"/users/",
json={"email": "[email protected]", "password": "testpassword"}
)
user_id = response.json()["id"]
# Now delete the user
response = client.delete(f"/users/{user_id}")
assert response.status_code == 200
# Verify user was deleted
response = client.get(f"/users/{user_id}")
assert response.status_code == 404
Testing Database Transactions
Testing transactions is important to ensure your database operations are atomic:
python
# test_transactions.py
import pytest
from sqlalchemy.orm import Session
from .models import User
def test_failed_transaction(test_db):
db = TestingSessionLocal()
# Start a transaction
try:
# Create a valid user
user1 = User(email="[email protected]", hashed_password="hashedpw1")
db.add(user1)
# Try to create an invalid user (missing required field)
user2 = User(email="[email protected]") # Missing hashed_password
db.add(user2)
# This should raise an exception if hashed_password is required
db.commit()
assert False, "Expected an exception but none was raised"
except:
db.rollback()
# Verify that no users were added due to transaction rollback
users = db.query(User).all()
assert len(users) == 0
db.close()
Testing with Dependency Injection
FastAPI's dependency injection system can be used in tests:
python
# test_dependencies.py
from fastapi import Depends, HTTPException
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from .main import app
from .dependencies import get_current_user, get_db
from .models import User
def test_protected_route(client, test_db):
# Setup: Create a user
db = TestingSessionLocal()
user = User(
id=1,
email="[email protected]",
hashed_password="hashedpassword",
is_active=True
)
db.add(user)
db.commit()
db.close()
# Override the get_current_user dependency
async def override_get_current_user():
return {"id": 1, "email": "[email protected]"}
app.dependency_overrides[get_current_user] = override_get_current_user
# Test the protected route
response = client.get("/protected-route/")
assert response.status_code == 200
# Clean up
app.dependency_overrides.clear()
Testing Database Migrations (with Alembic)
If you're using Alembic for database migrations, you can test your migrations:
python
# test_migrations.py
import subprocess
import pytest
import os
@pytest.mark.migration
def test_migrations():
# Run migrations in test environment
result = subprocess.run(
["alembic", "upgrade", "head"],
env={"DATABASE_URL": "sqlite:///./test_migrations.db"},
capture_output=True,
text=True
)
assert result.returncode == 0
# Verify migration was successful (check for expected output)
assert "Running upgrade" in result.stdout
# Clean up
if os.path.exists("test_migrations.db"):
os.remove("test_migrations.db")
Advanced: Testing Database Performance
For more advanced applications, you might want to test database performance:
python
# test_performance.py
import time
import pytest
from sqlalchemy.orm import Session
from .models import User
def test_query_performance(test_db):
db = TestingSessionLocal()
# Add 100 users for performance testing
for i in range(100):
user = User(email=f"user{i}@example.com", hashed_password=f"password{i}")
db.add(user)
db.commit()
# Measure query performance
start_time = time.time()
users = db.query(User).all()
end_time = time.time()
# Assert that the query completes within a reasonable time (e.g., 50ms)
assert end_time - start_time < 0.05
assert len(users) == 100
db.close()
Real-world Example: Testing a Blog API
Let's put everything together by testing a simple blog API:
python
# models.py
from sqlalchemy import Column, Integer, String, ForeignKey, Text
from sqlalchemy.orm import relationship
from .database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
posts = relationship("Post", back_populates="author")
class Post(Base):
__tablename__ = "posts"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
content = Column(Text)
author_id = Column(Integer, ForeignKey("users.id"))
author = relationship("User", back_populates="posts")
Now, let's test the blog API:
python
# test_blog_api.py
import pytest
from fastapi.testclient import TestClient
def test_blog_workflow(client, test_db):
# 1. Register a user
response = client.post(
"/users/",
json={"email": "[email protected]", "password": "securepassword"}
)
assert response.status_code == 200
user_data = response.json()
user_id = user_data["id"]
# 2. Login with the user
response = client.post(
"/token",
data={"username": "[email protected]", "password": "securepassword"}
)
assert response.status_code == 200
token_data = response.json()
access_token = token_data["access_token"]
headers = {"Authorization": f"Bearer {access_token}"}
# 3. Create a blog post
response = client.post(
"/posts/",
headers=headers,
json={"title": "Testing with FastAPI", "content": "This is a test post."}
)
assert response.status_code == 200
post_data = response.json()
post_id = post_data["id"]
# 4. Get the blog post
response = client.get(f"/posts/{post_id}")
assert response.status_code == 200
retrieved_post = response.json()
assert retrieved_post["title"] == "Testing with FastAPI"
assert retrieved_post["content"] == "This is a test post."
assert retrieved_post["author"]["id"] == user_id
assert retrieved_post["author"]["email"] == "[email protected]"
# 5. Update the blog post
response = client.put(
f"/posts/{post_id}",
headers=headers,
json={"title": "Updated Title", "content": "Updated content"}
)
assert response.status_code == 200
updated_post = response.json()
assert updated_post["title"] == "Updated Title"
# 6. Delete the blog post
response = client.delete(f"/posts/{post_id}", headers=headers)
assert response.status_code == 200
# 7. Verify the post was deleted
response = client.get(f"/posts/{post_id}")
assert response.status_code == 404
Best Practices for FastAPI Database Testing
-
Use separate test databases: Never run tests against your production database.
-
Test atomicity: Ensure your transactions are properly committed or rolled back.
-
Isolate tests: Each test should start with a clean database state.
-
Mock external services: If your database operations depend on external services, mock them.
-
Test edge cases: Test with empty values, large datasets, and unexpected inputs.
-
Check for N+1 query problems: Make sure your queries are optimized.
-
Test migrations: Ensure database schema changes won't break your application.
-
Use fixtures efficiently: Reuse setup code with pytest fixtures.
Summary
In this guide, you've learned how to:
- Set up a test environment for FastAPI database testing
- Create pytest fixtures for database testing
- Test database models and CRUD operations
- Handle database transactions in tests
- Use FastAPI's dependency injection in tests
- Test database migrations
- Assess database performance
- Build comprehensive tests for a real-world blog API
Database testing ensures your FastAPI application works correctly with your database layer. By writing thorough tests, you can catch issues early and refactor with confidence.
Additional Resources
- FastAPI Official Documentation
- SQLAlchemy Documentation
- pytest Documentation
- Testing Database Applications by Martin Fowler
Exercises
- Write tests for a user authentication system with password hashing.
- Create tests for a many-to-many relationship between models.
- Test database pagination functionality.
- Implement test coverage reporting for your database tests.
- Write a performance test for a complex query involving multiple table joins.
Happy testing!
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)