🧠 What You'll Learn
- 🏗️ How to structure a real-world FastAPI project
- 🛢️ Use MySQL/MariaDB with SQLAlchemy ORM
- 🧾 Perform full CRUD operations
- 📦 Handle exceptions globally
- ✅ Standardize all HTTP responses
- 🐍 Use
.envconfig, dependency injection, and clean code practices
📁 Folder Structure (Production-Ready)
fastapi_crud_app/
├── app/
│ ├── __init__.py
│ ├── api/
│ │ └── v1/
│ │ ├── __init__.py
│ │ └── routes/
│ │ └── item.py
│ ├── core/
│ │ ├── __init__.py
│ │ └── config.py
│ ├── crud/
│ │ └── item.py
│ ├── db/
│ │ ├── base.py
│ │ ├── base_class.py
│ │ ├── models/
│ │ │ └── item.py
│ │ └── session.py
│ ├── exceptions/
│ │ ├── custom_exceptions.py
│ │ └── handlers.py
│ ├── schemas/
│ │ └── item.py
│ ├── utils/
│ │ └── response.py
│ └── main.py
├── .env
├── requirements.txt
└── README.md
⚙️ Setup and Dependencies
✅ Install
pip install fastapi uvicorn sqlalchemy pymysql python-dotenv
✅ .env
DATABASE_URL=mysql+pymysql://user:password@localhost:3306/fastapi_db
🔧 Configuration – app/core/config.py
from dotenv import load_dotenv
import os
load_dotenv()
class Settings:
DATABASE_URL: str = os.getenv("DATABASE_URL")
settings = Settings()
🛢️ Database – app/db/session.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
engine = create_engine(settings.DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
📦 Base Class – app/db/base_class.py
from sqlalchemy.ext.declarative import as_declarative, declared_attr
@as_declarative()
class Base:
id: int
__name__: str
@declared_attr
def __tablename__(cls) -> str:
return cls.__name__.lower()
📁 Models – app/db/models/item.py
from sqlalchemy import Column, Integer, String
from app.db.base_class import Base
class Item(Base):
id = Column(Integer, primary_key=True, index=True)
title = Column(String(255), nullable=False)
description = Column(String(255), nullable=True)
🧬 Base Import – app/db/base.py
from app.db.models import item
from app.db.base_class import Base
📑 Schemas – app/schemas/item.py
from pydantic import BaseModel
class ItemBase(BaseModel):
title: str
description: str
class ItemCreate(ItemBase):
pass
class ItemUpdate(ItemBase):
pass
class ItemOut(ItemBase):
id: int
class Config:
orm_mode = True
🛠️ CRUD Logic – app/crud/item.py
from sqlalchemy.orm import Session
from app.db.models.item import Item
from app.schemas.item import ItemCreate, ItemUpdate
def create_item(db: Session, item: ItemCreate):
db_item = Item(**item.dict())
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
def get_item(db: Session, item_id: int):
return db.query(Item).filter(Item.id == item_id).first()
def get_items(db: Session, skip=0, limit=100):
return db.query(Item).offset(skip).limit(limit).all()
def update_item(db: Session, item_id: int, item: ItemUpdate):
db_item = get_item(db, item_id)
if db_item:
for key, value in item.dict().items():
setattr(db_item, key, value)
db.commit()
db.refresh(db_item)
return db_item
def delete_item(db: Session, item_id: int):
db_item = get_item(db, item_id)
if db_item:
db.delete(db_item)
db.commit()
return db_item
🌐 API Routes – app/api/v1/routes/item.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.schemas.item import ItemCreate, ItemOut, ItemUpdate
from app.crud import item as crud
from app.db.session import SessionLocal
from app.utils.response import APIResponse
from app.exceptions.custom_exceptions import NotFoundException
router = APIRouter()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@router.post("/", response_model=APIResponse)
def create_item(data: ItemCreate, db: Session = Depends(get_db)):
item = crud.create_item(db, data)
return APIResponse(status="success", message="Item created", data=item)
@router.get("/", response_model=APIResponse)
def get_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = crud.get_items(db, skip, limit)
return APIResponse(status="success", message="Item list", data=items)
@router.get("/{item_id}", response_model=APIResponse)
def get_item(item_id: int, db: Session = Depends(get_db)):
item = crud.get_item(db, item_id)
if not item:
raise NotFoundException(name="Item")
return APIResponse(status="success", message="Item found", data=item)
@router.put("/{item_id}", response_model=APIResponse)
def update_item(item_id: int, data: ItemUpdate, db: Session = Depends(get_db)):
updated = crud.update_item(db, item_id, data)
if not updated:
raise NotFoundException(name="Item")
return APIResponse(status="success", message="Item updated", data=updated)
@router.delete("/{item_id}", response_model=APIResponse)
def delete_item(item_id: int, db: Session = Depends(get_db)):
deleted = crud.delete_item(db, item_id)
if not deleted:
raise NotFoundException(name="Item")
return APIResponse(status="success", message="Item deleted", data=deleted)
🚨 Custom Exceptions – app/exceptions/custom_exceptions.py
class NotFoundException(Exception):
def __init__(self, name: str):
self.name = name
🌐 Global Exception Handler – app/exceptions/handlers.py
from fastapi import Request
from fastapi.responses import JSONResponse
from app.exceptions.custom_exceptions import NotFoundException
def not_found_exception_handler(request: Request, exc: NotFoundException):
return JSONResponse(
status_code=404,
content={
"status": "error",
"message": f"{exc.name} not found",
"data": None
}
)
🔄 Response Utility – app/utils/response.py
from typing import Optional, Any
from pydantic import BaseModel
class APIResponse(BaseModel):
status: str
message: str
data: Optional[Any] = None
🚀 Main App – app/main.py
from fastapi import FastAPI
from app.api.v1.routes import item
from app.db.base import Base
from app.db.session import engine
from app.exceptions.handlers import not_found_exception_handler
from app.exceptions.custom_exceptions import NotFoundException
app = FastAPI(title="FastAPI CRUD App")
# Register exception handlers
app.add_exception_handler(NotFoundException, not_found_exception_handler)
# Create tables
Base.metadata.create_all(bind=engine)
# Routes
app.include_router(item.router, prefix="/api/v1/items", tags=["Items"])
🧪 Test It!
uvicorn app.main:app --reload
Open docs: http://localhost:8000/docs
🧰 Ready for Production?
- ✅ Add Alembic for migrations
- ✅ Use Gunicorn + UvicornWorker
- ✅ Add JWT Auth for secure APIs
- ✅ Containerize with Docker
- ✅ Use logging and monitoring tools
Top comments (0)