Building a Python FastAPI CRUD API with MVC Structure

In this guide, we will build a Customer Management CRUD API using FastAPI. We’ll structure the application using the Model-View-Controller (MVC) pattern. The API will use SQLAlchemy ORM to interact with a MySQL database and Pydantic models for validation. We’ll also implement a REST API response wrapper to ensure consistent responses.

Goals of This Guide

• Organize code using MVC structure.
• Use a REST API response wrapper with fields like status, message, data, and error.
• Implement validations and handle errors gracefully.
• Use a single Pydantic model for create, update, and delete operations.
• Generate a UUID for the primary key.

Project Structure

Here’s how we’ll structure the FastAPI project:

fastapi_crud_mvc/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── controllers/
│   │   ├── __init__.py
│   │   └── customer_controller.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── customer.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── customer_schema.py
│   ├── db/
│   │   ├── __init__.py
│   │   └── database.py
│   └── utils/
│       ├── __init__.py
│       └── response_wrapper.py
└── requirements.txt

Step 1: Install Dependencies

Create a virtual environment and install necessary packages:

pip install fastapi uvicorn sqlalchemy mysqlclient pymysql pydantic

Step 2: Database Configuration (db/database.py)

First, create the database connection:

# db/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "mysql+pymysql://<username>:<password>@localhost:3306/<database_name>"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Base = declarative_base()

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

Replace <username>, <password>, and <database_name> with your MySQL credentials.

Step 3: Create the ORM Model (models/customer.py)

Define the Customer table schema using SQLAlchemy:

# models/customer.py
import uuid
from sqlalchemy import Column, String
from sqlalchemy.dialects.mysql import CHAR
from ..db.database import Base

class Customer(Base):
    __tablename__ = "customers"

    id = Column(CHAR(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    name = Column(String(100), nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    address = Column(String(255), nullable=True)

Step 4: Create the Pydantic Schema (schemas/customer_schema.py)

Create a single CustomerSchema for create, update, and response:

# schemas/customer_schema.py
from pydantic import BaseModel, EmailStr
from typing import Optional

class CustomerSchema(BaseModel):
    name: Optional[str] = None
    email: Optional[EmailStr] = None
    address: Optional[str] = None

    class Config:
        orm_mode = True

• CustomerSchema is used for both create and update requests by making fields optional.

Step 5: REST API Response Wrapper (utils/response_wrapper.py)

To ensure consistent API responses, define a wrapper function:

# utils/response_wrapper.py
def api_response(data=None, message="Success", status=True, error=None):
    return {
        "status": status,
        "message": message,
        "data": data,
        "error": error,
    }

Step 6: Controller (controllers/customer_controller.py)

Define the CRUD operations in the controller layer:

# controllers/customer_controller.py
from fastapi import APIRouter, HTTPException, Depends
from sqlalchemy.orm import Session
from ..models.customer import Customer
from ..schemas.customer_schema import CustomerSchema
from ..db.database import get_db
from ..utils.response_wrapper import api_response

router = APIRouter()

# CREATE Customer
@router.post("/customers/")
def create_customer(customer: CustomerSchema, db: Session = Depends(get_db)):
    if db.query(Customer).filter(Customer.email == customer.email).first():
        raise HTTPException(status_code=400, detail="Email already registered")
    
    new_customer = Customer(**customer.dict())
    db.add(new_customer)
    db.commit()
    db.refresh(new_customer)
    return api_response(data=new_customer, message="Customer created successfully")

# READ All Customers
@router.get("/customers/")
def get_customers(db: Session = Depends(get_db)):
    customers = db.query(Customer).all()
    return api_response(data=customers, message="All customers retrieved")

# READ Single Customer
@router.get("/customers/{customer_id}")
def get_customer(customer_id: str, db: Session = Depends(get_db)):
    customer = db.query(Customer).filter(Customer.id == customer_id).first()
    if customer is None:
        raise HTTPException(status_code=404, detail="Customer not found")
    return api_response(data=customer, message="Customer retrieved successfully")

# UPDATE Customer
@router.put("/customers/{customer_id}")
def update_customer(customer_id: str, customer_update: CustomerSchema, db: Session = Depends(get_db)):
    customer = db.query(Customer).filter(Customer.id == customer_id).first()
    if not customer:
        raise HTTPException(status_code=404, detail="Customer not found")
    
    for field, value in customer_update.dict(exclude_unset=True).items():
        setattr(customer, field, value)
    
    db.commit()
    db.refresh(customer)
    return api_response(data=customer, message="Customer updated successfully")

# DELETE Customer
@router.delete("/customers/{customer_id}")
def delete_customer(customer_id: str, db: Session = Depends(get_db)):
    customer = db.query(Customer).filter(Customer.id == customer_id).first()
    if not customer:
        raise HTTPException(status_code=404, detail="Customer not found")
    
    db.delete(customer)
    db.commit()
    return api_response(message="Customer deleted successfully")

Step 7: Register Controller in main.py

Create the main FastAPI app and include the customer router:

# main.py
from fastapi import FastAPI
from .controllers.customer_controller import router as customer_router

app = FastAPI()

app.include_router(customer_router, prefix="/api", tags=["Customers"])

@app.get("/")
def root():
    return {"message": "Welcome to the FastAPI CRUD API"}

Step 8: Run the FastAPI Application

Run the app using Uvicorn:

uvicorn app.main:app --reload

Step 9: Testing the API

Visit Swagger UI at:

http://127.0.0.1:8000/docs

You can test all the CRUD endpoints from the interactive API documentation.

Sample API Requests and Responses

1. Create a Customer (POST /api/customers/)

Request:

{
  "name": "John Doe",
  "email": "john@example.com",
  "address": "123 Main St"
}

Response:

{
  "status": true,
  "message": "Customer created successfully",
  "data": {
    "id": "b6a25c97-5d69-4f41-82d3-1a4d5f4731d8",
    "name": "John Doe",
    "email": "john@example.com",
    "address": "123 Main St"
  },
  "error": null
}

2. Get All Customers (GET /api/customers/)

Response:

{
  "status": true,
  "message": "All customers retrieved",
  "data": [
    {
      "id": "b6a25c97-5d69-4f41-82d3-1a4d5f4731d8",
      "name": "John Doe",
      "email": "john@example.com",
      "address": "123 Main St"
    }
  ],
  "error": null
}

Conclusion

In this post, we created a Python FastAPI CRUD API using the MVC pattern. We used SQLAlchemy ORM for database operations, Pydantic for request validation, and MySQL as our database. Additionally, we implemented a REST API response wrapper for consistent responses. This structure makes the project scalable, maintainable, and production-ready!