FastAPI + SQLAlchemy: Rapidly Building Database-Driven APIs

Why Choose FastAPI + SQLAlchemy?¶

In Python Web development, FastAPI is widely popular for its high performance, automatic API documentation generation, and concise syntax. SQLAlchemy, as a powerful ORM (Object-Relational Mapping) tool, allows developers to operate databases using Python classes, avoiding the need to write complex SQL statements directly. Combining these two enables quick construction of efficient and maintainable database-driven APIs, making it ideal for beginners to get started.

Preparation: Install Dependencies¶

First, ensure your Python environment is installed. Open your terminal and run the following command to install the required libraries:

pip install fastapi uvicorn sqlalchemy pydantic

FastAPI: Core web framework for building APIs
Uvicorn: ASGI server for running FastAPI applications
SQLAlchemy: ORM tool for database interactions
Pydantic: Data validation and serialization library (dependent on FastAPI for request/response handling)

Project Structure Design¶

We use a modular structure to split different functionalities into independent files for easier maintenance:

myapp/
├── main.py          # FastAPI main program, defines API routes
├── database.py      # Database connection configuration
├── models.py        # SQLAlchemy ORM models (database table structures)
├── schemas.py       # Pydantic models (data validation/serialization)
└── crud.py          # CRUD operations (Create, Read, Update, Delete)

1. Database Configuration (database.py)¶

This part handles database connections and session management. We’ll use SQLite as an example (no additional server required, suitable for beginners):

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

# Database connection URL (SQLite requires only a file path)
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

# Create SQLAlchemy engine (entry point for database connections)
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}  # SQLite multi-thread compatibility
)

# Create local session class (for independent sessions per request)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base model class (all ORM models must inherit from this)
Base = declarative_base()

2. Database Models (models.py)¶

Define database table structures using SQLAlchemy (ORM models), taking a user table as an example:

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

class User(Base):
    __tablename__ = "users"  # Database table name

    id = Column(Integer, primary_key=True, index=True)  # Primary key
    name = Column(String, index=True)                   # Name (index for query optimization)
    email = Column(String, unique=True, index=True)     # Email (unique, indexed)
    age = Column(Integer, nullable=True)                # Age (optional)

Explanation:
- Base inherits from SQLAlchemy’s declarative_base and serves as the parent class for all models
- __tablename__ specifies the database table name
- Column defines table fields with specified types (e.g., Integer, String) and constraints (e.g., primary key, unique)

3. Pydantic Models (schemas.py)¶

FastAPI uses Pydantic to define data validation and serialization rules (separating request and response formats):

from pydantic import BaseModel

# Request data validation for creating a user (excludes ID, which is auto-generated by the database)
class UserCreate(BaseModel):
    name: str
    email: str
    age: int | None = None  # Optional field

# Response format for user data (includes all fields)
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    age: int | None = None

    class Config:
        orm_mode = True  # Allows creating instances from ORM models (SQLAlchemy objects)

Explanation:
- UserCreate is used to receive data from the frontend (e.g., POST requests) and validate data integrity
- UserResponse formats the returned query results. orm_mode=True enables conversion from SQLAlchemy model instances

4. CRUD Operations (crud.py)¶

Implement database CRUD operations (Create, Read, Update, Delete):

from sqlalchemy.orm import Session
from models import User
from schemas import UserCreate, UserResponse

# Create user
def create_user(db: Session, user: UserCreate) -> UserResponse:
    db_user = User(name=user.name, email=user.email, age=user.age)
    db.add(db_user)
    db.commit()       # Commit transaction
    db.refresh(db_user)  # Refresh to get database-generated ID
    return db_user

# Get a single user
def get_user(db: Session, user_id: int) -> UserResponse | None:
    return db.query(User).filter(User.id == user_id).first()

# Get user list
def get_users(db: Session, skip: int = 0, limit: int = 100) -> list[UserResponse]:
    return db.query(User).offset(skip).limit(limit).all()

# Update user
def update_user(db: Session, user_id: int, user_update: UserCreate) -> UserResponse | None:
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        return None
    # Update fields
    db_user.name = user_update.name
    db_user.email = user_update.email
    db_user.age = user_update.age
    db.commit()
    db.refresh(db_user)
    return db_user

# Delete user
def delete_user(db: Session, user_id: int) -> bool:
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        return False
    db.delete(db_user)
    db.commit()
    return True

5. Main Program (main.py)¶

Integrate all modules and define API routes:

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from database import engine, SessionLocal, Base
from models import User
from schemas import UserCreate, UserResponse
from crud import create_user, get_user, get_users, update_user, delete_user

# Create database tables (executed on first run, requires importing all models)
Base.metadata.create_all(bind=engine)

app = FastAPI(title="User API Example")

# Dependency: Get database session (automatically create/close sessions per request)
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()  # Ensure session closure to prevent connection leaks

# API Routes
@app.post("/users/", response_model=UserResponse)
def create_new_user(user: UserCreate, db: Session = Depends(get_db)):
    """Create a new user"""
    return create_user(db=db, user=user)

@app.get("/users/{user_id}", response_model=UserResponse | None)
def read_user(user_id: int, db: Session = Depends(get_db)):
    """Get a single user"""
    db_user = get_user(db=db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.get("/users/", response_model=list[UserResponse])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    """Get user list"""
    users = get_users(db=db, skip=skip, limit=limit)
    return users

@app.put("/users/{user_id}", response_model=UserResponse | None)
def update_existing_user(user_id: int, user: UserCreate, db: Session = Depends(get_db)):
    """Update user information"""
    return update_user(db=db, user_id=user_id, user_update=user)

@app.delete("/users/{user_id}", response_model=bool)
def delete_existing_user(user_id: int, db: Session = Depends(get_db)):
    """Delete user"""
    success = delete_user(db=db, user_id=user_id)
    if not success:
        raise HTTPException(status_code=404, detail="User not found")
    return success

Running and Testing¶

Start the Application: Run the following command in the terminal:

   uvicorn main:app --reload

main:app means importing the app instance from main.py
--reload enables hot-reloading, restarting automatically after code changes

Test the API:
- Visit Swagger UI at: http://localhost:8000/docs to interactively test all endpoints
- Or test using curl:

     # Create a user
     curl -X POST "http://localhost:8000/users/" -H "Content-Type: application/json" -d '{"name":"Alice","email":"alice@example.com","age":25}'

     # Get user list
     curl "http://localhost:8000/users/"

Summary of Core Advantages¶

FastAPI: Automatic API documentation, async support, robust data validation
SQLAlchemy: ORM simplifies database operations without writing raw SQL
Modular Design: Separates routes, models, validation, and business logic for easy extension
Dependency Injection: Automatically manages database sessions to prevent connection leaks

With the above steps, you’ve completed the development of a basic database-driven API. Next, you can try extending functionality, such as adding user permissions, implementing multi-table relationships (e.g., order tables and user tables), etc. Practice is the best way to learn programming—modify the code and explore more possibilities with FastAPI + SQLAlchemy!