Creating a CRUD app with FastAPI + Postgresql

🧠 PostgreSQL Setup

---

✅ 1. Install PostgreSQL (if not already)

sudo apt update
sudo apt install postgresql postgresql-contrib

✅ 2. Start PostgreSQL service

sudo service postgresql start

---

🧑‍💻 3. Create a PostgreSQL User & Database

Step A: Open PostgreSQL shell

sudo -u postgres psql

Step B: Create a user

CREATE USER raj WITH PASSWORD 'yourpassword';

Step C: Create a database

CREATE DATABASE fastapidb OWNER raj;

Step D: Give privileges

GRANT ALL PRIVILEGES ON DATABASE fastapidb TO raj;

Step E: Exit psql

\q

---

📌 4. Database URL Format :

postgresql://raj:yourpassword@localhost/fastapidb

---

🧭 FastAPI Learning Roadmap

📌 Step 1: Introduction & Setup

• ✅ What is FastAPI?

• ✅ Installing FastAPI + Uvicorn

• ✅ Running your first API

📌 Step 2: Core Concepts

• 📥 Path & query parameters (@app.get)

• 🧾 Request bodies (@app.post)

• 🧠 Pydantic models (validation)

• 📤 Response models

📌 Step 3: Intermediate Usage

• 🧩 CRUD with database (SQLAlchemy + PostgreSQL ✅ already covered)

🧠 FastAPI - Deep Introduction

📦 Installing FastAPI

Step 1: Create a virtual environment

python3 -m venv venv
source venv/bin/activate

Step 2: Install FastAPI & Uvicorn

pip install fastapi[all]

• fastapi → the actual framework

• [all] installs optional dependencies like pydantic, uvicorn, and doc support

• uvicorn → lightweight ASGI server to run FastAPI apps

📁 Create your first FastAPI app

📄 File: main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello Raj! Welcome to FastAPI!"}

---

▶️ Run the app

In terminal:

uvicorn main:app --reload

Here:

• main = filename (without .py)

• app = FastAPI instance name inside the file

• --reload = auto-reload on file change (great during development)

  🌐 Visit these URLs in browser:

🔹 @app.get() with Path & Query Parameters (Deep Dive)

---

✅ 1. Path Parameters

📌 Use-case:

When you want to pass a required value in the URL itself, like /items/5, where 5 is the item_id.

---

🔧 Code Example:

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
def get_item(item_id: int):
    return {"item_id": item_id}

---

🧠 Explanation:

• /items/{item_id} means FastAPI is expecting a value at that URL location.

• item_id: int automatically:

  • Converts the path string to an integer

  • Throws an error if the value is not an integer

---

💡 Try:

Visit this in your browser:

http://127.0.0.1:8000/items/101

🔄 Output:

{ "item_id": 101 }

---

❌ If you visit /items/abc (non-integer):

You'll get:

{
  "detail": [
    {
      "loc": ["path", "item_id"],
      "msg": "value is not a valid integer",
      "type": "type_error.integer"
    }
  ]
}

This is auto-generated validation from Pydantic + FastAPI 🤯 — no manual if-else needed!

---

✅ 2. Query Parameters

📌 Use-case:

Optional or filter-based values passed after ? in a URL. Like:

/products?skip=10&limit=5

---

🔧 Code Example:

@app.get("/products")
def list_products(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

---

💡 Try:

Visit:

http://127.0.0.1:8000/products?skip=5&limit=2

🔄 Output:

{ "skip": 5, "limit": 2 }

---

✅ What's Happening Behind the Scenes?

FastAPI is:

• Reading query strings (?skip=5&limit=2)

• Matching them with function parameters

• Automatically validating types (int)

• Using default values if they’re not provided

---

🧠 Bonus: Combine Path + Query

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

URL:

http://127.0.0.1:8000/items/5?q=hello

Result:

{
  "item_id": 5,
  "q": "hello"
}

---

🧾 Step 3: Request Body with Pydantic Models (POST requests)

---

🔍 What is a Request Body?

When a client sends data inside the body of a request — not in the URL — for example when creating a new user, item, product, etc.

This usually happens with POST or PUT methods.

---

📦 Pydantic: The Power Engine

FastAPI uses Pydantic models (Python classes) to:

• ✅ Automatically validate incoming JSON

• ✅ Convert it to Python objects

• ✅ Auto-generate Swagger docs

---

✅ Step-by-Step Example

---

📄 main.py

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# Step 1: Define a model for the data you expect
class Item(BaseModel):
    name: str
    description: str
    price: float
    in_stock: bool = True  # default value

# Step 2: Accept that model as a parameter in your POST endpoint
@app.post("/items/")
def create_item(item: Item):
    return {
        "name": item.name,
        "desc": item.description,
        "price": item.price,
        "available": item.in_stock
    }

---

▶️ Run and test:

uvicorn main:app --reload

Visit:

http://127.0.0.1:8000/docs

Click on POST /items/ → Try it out → Paste:

{
  "name": "Mouse",
  "description": "Wireless Logitech mouse",
  "price": 499.99
}

🔄 Response:

{
  "name": "Mouse",
  "desc": "Wireless Logitech mouse",
  "price": 499.99,
  "available": true
}

🎯 Goal of Response Models in FastAPI

Even though your function might return a lot of data (or even sensitive data), you can use response models to:

• ✅ Return only the fields you want

• ✅ Validate and shape the output

• ✅ Auto-generate clean, accurate docs

---

✅ Example Use Case

Suppose we store passwords or other sensitive fields in the DB — we should never send those back in API responses.

So we define 2 models:

• 🧾 ItemCreate → for request body (input)

• 📤 ItemResponse → for response model (output)

---

🧪 Step-by-Step Example

📄 main.py

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# Input model (request)
class ItemCreate(BaseModel):
    name: str
    description: str
    price: float
    secret_code: str  # 🔐 to be hidden in response

# Output model (response)
class ItemResponse(BaseModel):
    name: str
    description: str
    price: float

@app.post("/items/", response_model=ItemResponse)
def create_item(item: ItemCreate):
    # Normally you would save to DB here...
    return item  # Will be filtered by response_model

---

✅ Try it in /docs

POST /items/ with:

{
  "name": "Keyboard",
  "description": "Mechanical RGB keyboard",
  "price": 2500.0,
  "secret_code": "12345"
}

🟢 Response:

{
  "name": "Keyboard",
  "description": "Mechanical RGB keyboard",
  "price": 2500.0
}

✅ secret_code is not returned, even though the function returned it — because response_model=ItemResponse filtered it out!

---

🧠 Behind the Scenes

---

🔐 Use Case in Real Projects

If you have a User model with password field:

class UserIn(BaseModel):
    username: str
    password: str  # don't return this

class UserOut(BaseModel):
    username: str

@app.post("/register", response_model=UserOut)
def register_user(user: UserIn):
    # hash password and save to DB
    return user  # password will be filtered from response

---

Let’s now build the full CRUD project with FastAPI + PostgreSQL using the modular structure — one file at a time, with clear walkthroughs.

📂 Project Structure Overview

fastapi_postgres_crud/
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
└── routers/
    └── items.py

✅ Step 1: database.py – DB Connection Setup

📄 File: database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 🔧 Update your own credentials here
DATABASE_URL = "postgresql://raj:yourpassword@localhost/yourdb"

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

Base = declarative_base()

✅ Your Task:
Replace "yourpassword" and "yourdb" with your actual PostgreSQL credentials.

💡 What this does:

• Connects to your PostgreSQL DB

• Prepares SessionLocal for querying

• Creates a Base class to define models

---

Great! Now let’s move on to the next file: defining our table structure using SQLAlchemy ORM.

✅ Step 2: models.py – Define DB Table (SQLAlchemy Model)

📄 File: models.py

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

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    description = Column(String)
    price = Column(Float, nullable=False)
    in_stock = Column(Boolean, default=True)

---

🧠 Explanation:

• Item is our DB model mapped to the items table.

• Each class attribute is a column in the table.

• Base is inherited from database.py to register the model.

• nullable=False ensures name and price are required.

📌 Tip: This table will only be created when we run this in main.py — we’ll get there soon:

models.Base.metadata.create_all(bind=engine)

---

✅ Step 3: schemas.py – Define Input/Output Pydantic Models

📄 File: schemas.py

from pydantic import BaseModel

# Input model (for creating a new item)
class ItemCreate(BaseModel):
    name: str
    description: str
    price: float
    in_stock: bool = True

# Output model (for response)
class ItemResponse(BaseModel):
    id: int
    name: str
    description: str
    price: float
    in_stock: bool

    class Config:
        from_attributes = True  # allows SQLAlchemy to Pydantic conversion

---

🧠 Explanation:

• ItemCreate:

  • What the client must send in the body (e.g., POST /items)

  • Doesn’t include id because that’s auto-generated

• ItemResponse:

  • Sent back to the client (e.g., GET /items)

  • Includes id because it’s created by the DB

• from_attributes = True:

  • Tells Pydantic to accept ORM objects (like SQLAlchemy models)

✅ Once you save this file, let me know and we’ll proceed to the heart of data handling: crud.py — where we write functions to interact with the database.

---

✅ Step 4: crud.py – Database Logic (Create + Read)

📄 File: crud.py

from sqlalchemy.orm import Session
import models
import schemas

# Create a new item
def create_item(db: Session, item: schemas.ItemCreate):
    db_item = models.Item(**item.dict())  # Convert Pydantic to SQLAlchemy
    db.add(db_item)
    db.commit()
    db.refresh(db_item)  # Get updated DB record (with ID)
    return db_item

# Get all items
def get_items(db: Session):
    return db.query(models.Item).all()

# Get single item by ID
def get_item_by_id(db: Session, item_id: int):
    return db.query(models.Item).filter(models.Item.id == item_id).first()

# Delete item
def delete_item(db: Session, item_id: int):
    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if db_item:
        db.delete(db_item)
        db.commit()
    return db_item

# Update item
def update_item(db: Session, item_id: int, updated_item: schemas.ItemCreate):
    db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if db_item:
        for key, value in updated_item.dict().items():
            setattr(db_item, key, value)
        db.commit()
        db.refresh(db_item)
    return db_item

📌 This is not FastAPI code yet — this is just database access. The routes will call these functions.

✅ Let me know when this is saved and I’ll move to step 5: routers/items.py, where we finally expose these as actual API endpoints!

---

✅ Step 5: routers/items.py – Define Routes (API Endpoints)

📄 File: routers/items.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import SessionLocal
import schemas, crud
from typing import List

router = APIRouter(
    prefix="/items",
    tags=["Items"]
)

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

@router.post("/", response_model=schemas.ItemResponse)
def create(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    return crud.create_item(db, item)

@router.get("/", response_model=List[schemas.ItemResponse])
def read_all(db: Session = Depends(get_db)):
    return crud.get_items(db)

@router.get("/{item_id}", response_model=schemas.ItemResponse)
def read_one(item_id: int, db: Session = Depends(get_db)):
    db_item = crud.get_item_by_id(db, item_id)
    if not db_item:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

@router.put("/{item_id}", response_model=schemas.ItemResponse)
def update(item_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)):
    updated = crud.update_item(db, item_id, item)
    if not updated:
        raise HTTPException(status_code=404, detail="Item not found")
    return updated

@router.delete("/{item_id}", response_model=schemas.ItemResponse)
def delete(item_id: int, db: Session = Depends(get_db)):
    deleted = crud.delete_item(db, item_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Item not found")
    return deleted

Perfect! Let’s now wrap it up with the final piece: main.py, where we launch the FastAPI app and connect everything together.

✅ Step 6: main.py – App Entry Point

📄 File: main.py

from fastapi import FastAPI
import models
from database import engine
from routers import items

# Create all tables (runs once on startup)
models.Base.metadata.create_all(bind=engine)

# Initialize app
app = FastAPI()

# Register the items router
app.include_router(items.router)

---

🧠 What’s Happening Here:

---

✅ Launch the App

Go to your project root and run:

uvicorn main:app --reload

Then open: http://127.0.0.1:8000/docs
🎉 You’ll see your full CRUD API in Swagger!

📌 Final Checklist Recap