27. MongoDB

pip install motor  # async MongoDB driver

Connection Setup

# database.py
from motor.motor_asyncio import AsyncIOMotorClient

MONGO_URI = "mongodb://localhost:27017"
client = AsyncIOMotorClient(MONGO_URI)
db = client["fastapi_tutorial"]
items_collection = db["items"]

Pydantic Models

from pydantic import BaseModel
from typing import Optional
from bson import ObjectId

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = False

class ItemInDB(Item):
    id: str  # MongoDB ObjectId as string

Async CRUD

from fastapi import FastAPI, HTTPException
from database import items_collection

app = FastAPI()

@app.post("/items/", status_code=201)
async def create_item(item: Item):
    result = await items_collection.insert_one(item.dict())
    created = await items_collection.find_one({"_id": result.inserted_id})
    created["id"] = str(created.pop("_id"))
    return created

@app.get("/items/{item_id}")
async def get_item(item_id: str):
    item = await items_collection.find_one({"_id": ObjectId(item_id)})
    if not item:
        raise HTTPException(404)
    item["id"] = str(item.pop("_id"))
    return item