Mastering FastAPI: Building a Professional Production-Ready CRUD API

FastAPI has taken the Python world by storm, not just because it’s fast, but because it’s exceptionally developer-friendly. In this guide, we’ll build a production-ready CRUD (Create, Read, Update, Delete) API for a Task Management system.

1. Project Setup

First, let’s set up a virtual environment and install our dependencies:

mkdir fastapi-pro && cd fastapi-pro
python -m venv venv
source venv/bin/activate
pip install fastapi[all] sqlalchemy pydantic-settings

2. The Database Schema (SQLAlchemy)

We’ll use SQLite for simplicity, but the code is easily adaptable for PostgreSQL. Create database.py:

from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./tasks.db"

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

class Task(Base):
    __tablename__ = "tasks"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    description = Column(String)
    is_completed = Column(Boolean, default=False)

Base.metadata.create_all(bind=engine)

3. Data Validation (Pydantic)

Pydantic ensures your API only receives and returns valid data. Create schemas.py:

from pydantic import BaseModel
from typing import Optional

class TaskBase(BaseModel):
    title: str
    description: Optional[str] = None
    is_completed: bool = False

class TaskCreate(TaskBase):
    pass

class TaskResponse(TaskBase):
    id: int
    class Config:
        from_attributes = True # Enables ORM conversion

4. The main API Application

Now, let’s tie it all together in main.py:

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from .database import SessionLocal, Task
from .schemas import TaskCreate, TaskResponse
from typing import List

app = FastAPI(title="Pro Task API")

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

@app.post("/tasks/", response_model=TaskResponse, status_code=21)
def create_task(task: TaskCreate, db: Session = Depends(get_db)):
    db_task = Task(**task.model_dump())
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    return db_task

@app.get("/tasks/", response_model=List[TaskResponse])
def get_tasks(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return db.query(Task).offset(skip).limit(limit).all()

@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int, db: Session = Depends(get_db)):
    task = db.query(Task).filter(Task.id == task_id).first()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@app.put("/tasks/{task_id}", response_model=TaskResponse)
def update_task(task_id: int, updated_task: TaskCreate, db: Session = Depends(get_db)):
    task_query = db.query(Task).filter(Task.id == task_id)
    if not task_query.first():
        raise HTTPException(status_code=404, detail="Task not found")
    task_query.update(updated_task.model_dump())
    db.commit()
    return task_query.first()

@app.delete("/tasks/{task_id}", status_code=24)
def delete_task(task_id: int, db: Session = Depends(get_db)):
    task = db.query(Task).filter(Task.id == task_id).first()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    db.delete(task)
    db.commit()
    return {"message": "Task deleted successfully"}

5. Running the API

Run the server with Uvicorn:

uvicorn main:app --reload

Visit http://127.0.0.1:8000/docs to see your beautiful, interactive Swagger documentation!

Key Takeaways

Type Safety: Every input is validated before it hits your logic.
Auto-Docs: Swagger UI is generated automatically.
Performance: FastAPI is asynchronous by nature, making it extremely efficient for I/O-bound tasks.