FastAPI Database Introduction

What You'll Learn

In this guide, you'll learn how to work with databases in FastAPI applications. We'll cover the fundamentals of database integration, explore different database options, and implement basic database operations.

Introduction to FastAPI with Databases

Modern web applications often need to store, retrieve, and manipulate data. FastAPI provides excellent support for integrating with databases, allowing your application to persist data between requests and across application restarts.

FastAPI itself doesn't include a database engine or ORM (Object-Relational Mapping) tool directly. Instead, it's designed to work seamlessly with various database solutions, giving you the flexibility to choose the one that best fits your project requirements.

Database Options for FastAPI

When working with FastAPI, you have several database options:

1. SQLAlchemy - A powerful and widely-used Python ORM that supports multiple database engines
2. Tortoise ORM - An async ORM inspired by Django's ORM, designed specifically for Python asyncio
3. Databases - A simple async database library that works well with SQLAlchemy Core
4. Motor - An async driver for MongoDB
5. Beanie - An ODM (Object Document Mapper) for MongoDB built on top of Motor

For this introduction, we'll focus on SQLAlchemy, as it's the most widely used option in the FastAPI ecosystem.

Setting Up a Database with FastAPI

Let's walk through setting up a basic FastAPI application with SQLAlchemy.

1. Install Required Packages

First, install the necessary packages:

bash

pip install fastapi sqlalchemy psycopg2-binary

Or if you prefer using an async approach:

bash

pip install fastapi sqlalchemy psycopg2-binary databases

2. Database Connection Setup

Let's create a basic project structure:

fastapi_db_project/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── database.py
│   ├── models.py
│   └── schemas.py

Now, let's set up our database connection in database.py:

python

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Replace with your actual database URL
SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"

# For SQLite, you could use:
# SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# Dependency to get DB session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3. Define Database Models

Next, let's define our database models in models.py:

python

from sqlalchemy import Boolean, Column, Integer, String
from .database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    username = Column(String, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

4. Create Pydantic Schemas

Now let's define our Pydantic schemas in schemas.py for data validation and serialization:

python

from pydantic import BaseModel, EmailStr

class UserBase(BaseModel):
    email: EmailStr
    username: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool

    class Config:
        orm_mode = True

5. FastAPI Application Setup

Finally, let's implement our FastAPI application in main.py:

python

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from . import models, schemas
from .database import engine, get_db

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    # Check if user already exists
    db_user = db.query(models.User).filter(models.User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    
    # Create hashed password (in production, use a proper hashing algorithm)
    hashed_password = user.password + "_hashed"  # Not for production!
    
    # Create new user
    db_user = models.User(
        email=user.email,
        username=user.username,
        hashed_password=hashed_password
    )
    
    # Add to DB
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    
    return db_user

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(models.User).filter(models.User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = db.query(models.User).offset(skip).limit(limit).all()
    return users

Testing Your Database Integration

You can test your application using FastAPI's automatic interactive docs. Run your application with Uvicorn:

bash

uvicorn app.main:app --reload

Then navigate to http://localhost:8000/docs in your browser. You should see the Swagger UI interface where you can:

1. Create a new user with POST /users/
2. Fetch a user by ID with GET /users/{user_id}
3. List all users with GET /users/

Async Database Operations

For high-performance applications, you might want to use asynchronous database operations. Here's how you could adapt the above example to use databases for async operations:

python

# database.py with async support
from databases import Database
from sqlalchemy import create_engine, MetaData
from sqlalchemy.ext.declarative import declarative_base

DATABASE_URL = "postgresql://user:password@localhost/dbname"

# Database instance for async operations
database = Database(DATABASE_URL)

# SQLAlchemy setup for model definitions and migrations
metadata = MetaData()
engine = create_engine(DATABASE_URL)
Base = declarative_base(metadata=metadata)

And then in your main application:

python

from fastapi import FastAPI
import asyncio
from .database import database, engine, metadata
from . import models

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.post("/users/", response_model=schemas.User)
async def create_user(user: schemas.UserCreate):
    query = models.users.insert().values(
        email=user.email,
        username=user.username,
        hashed_password=user.password + "_hashed"  # Not for production!
    )
    user_id = await database.execute(query)
    
    return {**user.dict(), "id": user_id, "is_active": True}

Best Practices

When working with databases in FastAPI, keep these best practices in mind:

1. Use dependency injection for database sessions to ensure proper cleanup
2. Create separate models and schemas - SQLAlchemy models for database tables and Pydantic schemas for API requests and responses
3. Use migrations for database schema changes in production (Alembic is commonly used with SQLAlchemy)
4. Properly hash passwords - never store plain text passwords
5. Consider using async for high-performance applications
6. Use connection pooling for efficient database connections

Summary

In this introduction, you've learned:

• How to set up a database connection in FastAPI
• How to create database models and Pydantic schemas
• How to implement CRUD operations in FastAPI endpoints
• The basics of async database operations
• Best practices for working with databases in FastAPI

Connecting FastAPI with a database opens up a wide range of possibilities for your applications, from simple data storage to complex data management systems.

Additional Resources

• FastAPI SQL Databases documentation
• SQLAlchemy documentation
• Alembic migrations with SQLAlchemy

Exercise

Try extending the example application with:

1. Update and delete endpoints for user management
2. A new model for user posts with a relationship to users
3. Implement filtering and pagination for listing users

By completing these exercises, you'll gain a deeper understanding of database operations in FastAPI applications.