Module 5 — CRUD Operations & In-Memory DB
In this module, you'll build full CRUD APIs (Create, Read, Update, Delete) using a simple in-memory database (Python dictionary). This is perfect for learning before connecting to SQL or NoSQL databases.
1. Creating a Model & In-Memory DB
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from uuid import uuid4
app = FastAPI()
class Item(BaseModel):
name: str
price: float
description: str | None = None
# In-memory database
db = {}
2. CREATE (POST)
@app.post("/items")
async def create_item(item: Item):
item_id = str(uuid4())
db[item_id] = item
return {"id": item_id, "item": item}
3. READ ALL (GET)
@app.get("/items")
async def list_items():
return db
4. READ ONE (GET by ID)
@app.get("/items/{item_id}")
async def get_item(item_id: str):
if item_id not in db:
raise HTTPException(status_code=404, detail="Item not found")
return db[item_id]
5. UPDATE (PUT)
@app.put("/items/{item_id}")
async def update_item(item_id: str, item: Item):
if item_id not in db:
raise HTTPException(status_code=404, detail="Item not found")
db[item_id] = item
return {"id": item_id, "updated_item": item}
6. DELETE (DELETE)
@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
if item_id not in db:
raise HTTPException(status_code=404, detail="Item not found")
del db[item_id]
return {"deleted": item_id}
7. Complete CRUD File (Copy-Paste Ready)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from uuid import uuid4
app = FastAPI()
class Item(BaseModel):
name: str
price: float
description: str | None = None
db = {}
@app.post("/items")
async def create_item(item: Item):
item_id = str(uuid4())
db[item_id] = item
return {"id": item_id, "item": item}
@app.get("/items")
async def list_items():
return db
@app.get("/items/{item_id}")
async def get_item(item_id: str):
if item_id not in db:
raise HTTPException(status_code=404, detail="Item not found")
return db[item_id]
@app.put("/items/{item_id}")
async def update_item(item_id: str, item: Item):
if item_id not in db:
raise HTTPException(status_code=404, detail="Item not found")
db[item_id] = item
return {"id": item_id, "updated_item": item}
@app.delete("/items/{item_id}")
async def delete_item(item_id: str):
if item_id not in db:
raise HTTPException(status_code=404, detail="Item not found")
del db[item_id]
return {"deleted": item_id}
No comments:
Post a Comment