• Home
  • πŸš€ FastAPI for Java Developers: A Super-Fast Guide to Building REST APIs

πŸ“Œ Context

You’re a Java developer (likely using Spring Boot, Jakarta EE, etc.), but you need to build APIs fasterβ€”for a prototype, microservice, or edge-service. FastAPI is the Pythonic equivalent of what Spring Boot offers: declarative, fast, and typed.

This guide shows you how to:

  • Create a User model
  • Define schemas (DTO equivalents)
  • Use FastAPI routers and dependency injection
  • Build CRUD APIs using best practices

🧰 Prerequisites

  • Python 3.8+
  • pip or pipenv
  • Basic understanding of Python typing (like Optional, List, str, etc.)
# Install FastAPI and an ASGI server
pip install fastapi uvicorn[standard] pydantic[dotenv] sqlalchemy

πŸ“¦ Project Structure

fastapi-users/
β”‚
β”œβ”€β”€ app/
β”‚   β”œβ”€β”€ main.py               # Entry point
β”‚   β”œβ”€β”€ db.py                 # DB session and base
β”‚   β”œβ”€β”€ models/
β”‚   β”‚   └── user.py           # SQLAlchemy model
β”‚   β”œβ”€β”€ schemas/
β”‚   β”‚   └── user.py           # Pydantic schemas (DTOs)
β”‚   β”œβ”€β”€ crud/
β”‚   β”‚   └── user.py           # DB operations
β”‚   β”œβ”€β”€ api/
β”‚   β”‚   └── user.py           # Route handlers
β”‚   └── core/
β”‚       └── config.py         # Configuration
β”‚
└── requirements.txt

βš™οΈ Step 1: Database and Configuration

app/db.py

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

DATABASE_URL = "sqlite:///./test.db"  # Use PostgreSQL/MySQL in production

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Dependency for DB Session

# app/db.py (continued)
from contextlib import contextmanager

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

πŸ§‘β€πŸ’» Step 2: User Model (Entity Layer)

app/models/user.py

from sqlalchemy import Column, Integer, String
from app.db import Base

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    full_name = Column(String, index=True)

πŸ“€ Step 3: Schemas (DTO Layer)

app/schemas/user.py

from pydantic import BaseModel, EmailStr

# Input schema (like POST request)
class UserCreate(BaseModel):
    username: str
    email: EmailStr
    full_name: str

# Response schema
class UserOut(BaseModel):
    id: int
    username: str
    email: EmailStr
    full_name: str

    class Config:
        orm_mode = True

# For updates
class UserUpdate(BaseModel):
    full_name: str | None = None

πŸ› οΈ Step 4: CRUD Operations (Service Layer)

app/crud/user.py

from sqlalchemy.orm import Session
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate

def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

def get_users(db: Session, skip: int = 0, limit: int = 10):
    return db.query(User).offset(skip).limit(limit).all()

def create_user(db: Session, user: UserCreate):
    db_user = User(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def update_user(db: Session, user_id: int, user: UserUpdate):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user:
        for field, value in user.dict(exclude_unset=True).items():
            setattr(db_user, field, value)
        db.commit()
        db.refresh(db_user)
    return db_user

def delete_user(db: Session, user_id: int):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        db.delete(user)
        db.commit()
    return user

🌐 Step 5: API Router (Controller Layer)

app/api/user.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.db import get_db
from app.schemas.user import UserCreate, UserOut, UserUpdate
from app.crud import user as crud_user

router = APIRouter(prefix="/users", tags=["Users"])

@router.post("/", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def create(user: UserCreate, db: Session = Depends(get_db)):
    return crud_user.create_user(db, user)

@router.get("/{user_id}", response_model=UserOut)
def read(user_id: int, db: Session = Depends(get_db)):
    db_user = crud_user.get_user(db, user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@router.get("/", response_model=list[UserOut])
def read_all(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud_user.get_users(db, skip, limit)

@router.put("/{user_id}", response_model=UserOut)
def update(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    return crud_user.update_user(db, user_id, user)

@router.delete("/{user_id}", status_code=204)
def delete(user_id: int, db: Session = Depends(get_db)):
    crud_user.delete_user(db, user_id)
    return

πŸš€ Step 6: Main App Entry

app/main.py

from fastapi import FastAPI
from app.db import Base, engine
from app.api import user

Base.metadata.create_all(bind=engine)

app = FastAPI(title="FastAPI User CRUD")

app.include_router(user.router)

πŸ§ͺ Step 7: Run and Test

uvicorn app.main:app --reload

Swagger Docs


βœ… Best Practices Recap

PracticeDescription
TypingUse pydantic with full type annotations for clarity and safety.
Layered ArchitectureModels, Schemas, CRUD, API separated clearly.
Dependency InjectionUse Depends for database and other services.
ValidationPydantic schemas enforce validation at entry points.
Auto DocsFastAPI generates Swagger/OpenAPI automatically.
Modular RoutingSeparate routers make APIs clean and scalable.
Database SessionSafe management with dependency injection and context managers.
ORM Mode in SchemasEnables returning SQLAlchemy objects directly.

Leave Comment