Motor with FastAPI

Project Structure

app/
├── main.py
├── database.py
├── models.py
└── routes/
    └── users.py

Database Configuration

# database.py
from motor.motor_asyncio import AsyncIOMotorClient
from typing import Optional

class Database:
    client: Optional[AsyncIOMotorClient] = None

db = Database()

async def connect_db():
    db.client = AsyncIOMotorClient("mongodb://localhost:27017")
    print("Connected to MongoDB")

async def close_db():
    if db.client:
        db.client.close()
        print("Closed MongoDB connection")

def get_database():
    return db.client.mydb

Pydantic Models

# models.py
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
from bson import ObjectId

class PyObjectId(ObjectId):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, v, handler):
        if not ObjectId.is_valid(v):
            raise ValueError("Invalid ObjectId")
        return ObjectId(v)

class UserCreate(BaseModel):
    username: str
    email: str

class UserResponse(BaseModel):
    id: str = Field(alias="_id")
    username: str
    email: str
    created_at: datetime

    class Config:
        populate_by_name = True
        json_encoders = {ObjectId: str}

FastAPI Application

# main.py
from fastapi import FastAPI
from contextlib import asynccontextmanager
from database import connect_db, close_db
from routes import users

@asynccontextmanager
async def lifespan(app: FastAPI):
    await connect_db()
    yield
    await close_db()

app = FastAPI(lifespan=lifespan)
app.include_router(users.router, prefix="/users", tags=["users"])

Routes with Motor

# routes/users.py
from fastapi import APIRouter, HTTPException
from database import get_database
from models import UserCreate, UserResponse
from datetime import datetime
from bson import ObjectId

router = APIRouter()

@router.post("/", response_model=UserResponse)
async def create_user(user: UserCreate):
    db = get_database()
    user_dict = user.dict()
    user_dict["created_at"] = datetime.utcnow()

    result = await db.users.insert_one(user_dict)
    created_user = await db.users.find_one({"_id": result.inserted_id})

    return created_user

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: str):
    db = get_database()

    if not ObjectId.is_valid(user_id):
        raise HTTPException(status_code=400, detail="Invalid user ID")

    user = await db.users.find_one({"_id": ObjectId(user_id)})
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return user

@router.get("/", response_model=list[UserResponse])
async def list_users(skip: int = 0, limit: int = 10):
    db = get_database()
    cursor = db.users.find().skip(skip).limit(limit)
    return await cursor.to_list(length=limit)

@router.delete("/{user_id}")
async def delete_user(user_id: str):
    db = get_database()
    result = await db.users.delete_one({"_id": ObjectId(user_id)})

    if result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="User not found")

    return {"message": "User deleted"}

FastAPI Integration

FastAPI Integration

Motor with FastAPI

Project Structure

Database Configuration

Pydantic Models

FastAPI Application

Routes with Motor