In this content we will build the todo application  using the fastapi,mongodb and beanie. project will follow the standard folder structure as well . follow the steps and make the first application in fastapi.

folder  structure is as follow:

TODO-FASTAPI-APP
config          # Database            configuration
  __init__.py
  database.py
models          # MongoDB models using Beanie
  __init__.py
  todo_model.py
repositories    # Data access logic
  todo_repository.py
routes          # API endpoints for CRUD operations
  __init__.py
  todo_route.py
schemas         # Pydantic schemas for request/response validation
  todo_schema.py
services        # Business logic
  todo_service.py
utils           # Utility functions (if needed)
venv            # Virtual environment
main.py         # Application entry point
requirements.txt # Project dependencies

Config: to manage all the configuration related files

#config/database.py
from beanie import init_beanie
from motor.motor_asyncio import AsyncIOMotorClient
from models.todo_model import Todo

MONGO_URI = "mongodb://localhost:27017"
DB_NAME = "todo_db"

# MongoDB Connection URL
# MONGO_URL = "mongodb://localhost:27017/todo_app"
# db_client = motor.motor_asyncio.AsyncIOMotorClient(MONGO_URL)
# database = db_client.get_database("todo_app")

client = None

async def init_db():
    global client
    client = AsyncIOMotorClient(MONGO_URI)
    await init_beanie(database=client[DB_NAME], document_models=[Todo])
    print("? Database connected")

async def close_db():
    global client
    if client:
        client.close()
        print("???? Database connection closed")

Models: to manage all tables/collection in the application

#models/todo_model.py
from beanie import Document
from pydantic import Field
from typing import Optional
from datetime import datetime

class Todo(Document):
    title: str = Field(..., title="Title of the To-Do")
    description: Optional[str] = None
    completed: bool = Field(default=False)
    created_at: datetime = Field(default_factory=datetime.utcnow)

    class Settings:
        name = "todos"

Repositories: To manage database related logic

#respoitories/todo_repository.py
from models.todo_model import Todo
from schemas.todo_schema import TodoCreate,TodoResponse
from typing import List, Optional

async def create_todo(todo: TodoCreate) ->  TodoResponse:
    todo = Todo(**todo.model_dump())
    await todo.insert()
    return TodoResponse(**todo.model_dump())

async def get_all_todos() -> List[Todo]:
    # return await Todo.find(Todo.completed == False).to_list()
    return await Todo.find_all().to_list()

async def get_todo_by_id(todo_id: str) -> Optional[Todo]:
    return await Todo.get(todo_id)

async def update_todo(todo_id: str, update_data: dict) -> bool:
    todo = await Todo.get(todo_id)
    if todo:
        await todo.update({"$set": update_data})
        return True
    return False

async def delete_todo(todo_id: str) -> bool:
    todo = await Todo.get(todo_id)
    if todo:
        await todo.delete()
        return True
    return False

Routes: To manage the API routes

# routes/todo_route.py
from fastapi import APIRouter, HTTPException, status
from services.todo_service import (
    add_todo_service, list_todos_service, get_todo_service, 
    update_todo_service, delete_todo_service
)
from schemas.todo_schema import TodoCreate, TodoUpdate, TodoResponse
from typing import List
from beanie import PydanticObjectId

router = APIRouter()

@router.post("/", response_model=TodoResponse, status_code=status.HTTP_201_CREATED)
async def create_todo(todo: TodoCreate):
    created_todo = await add_todo_service(todo)
    return created_todo

@router.get("/", response_model=List[TodoResponse])
async def get_all_todos():
    return await list_todos_service()

@router.get("/{todo_id}", response_model=TodoResponse)
async def get_todo(todo_id: PydanticObjectId):
    todo = await get_todo_service(todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo

@router.put("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def update_todo(todo_id: PydanticObjectId, todo: TodoUpdate):
    if not await update_todo_service(todo_id, todo):
        raise HTTPException(status_code=404, detail="Todo not found")

@router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(todo_id: PydanticObjectId):
    if not await delete_todo_service(todo_id):
        raise HTTPException(status_code=404, detail="Todo not found")

Schemas: To manage the input/output validation

#schemas/todo_schema.py
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
from beanie import PydanticObjectId

class TodoCreate(BaseModel):
    title: str
    description: Optional[str] = None

class TodoUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    completed: Optional[bool] = None

class TodoResponse(BaseModel):
    id: PydanticObjectId| str = Field(...,)
    title: str
    description: Optional[str]
    completed: bool
    created_at: datetime 

Services: to manage business logic

#services/todo_service.py
from repositories.todo_repository import (
    create_todo, get_all_todos, get_todo_by_id, update_todo, delete_todo
)
from schemas.todo_schema import TodoCreate, TodoUpdate,TodoResponse

async def add_todo_service(todo_data: TodoCreate) -> TodoResponse:
    return await create_todo(todo_data)

async def list_todos_service():
    return await get_all_todos()

async def get_todo_service(todo_id: str):
    return await get_todo_by_id(todo_id)

async def update_todo_service(todo_id: str, todo_data: TodoUpdate):
    return await update_todo(todo_id, todo_data.model_dump(exclude_unset=True))

async def delete_todo_service(todo_id: str):
    return await delete_todo(todo_id)

Utils: if other utils file are available

# utils/pydantic_objectid.py
from pydantic import BaseModel, Field
from bson import ObjectId
from typing import Any, Dict

class PydanticObjectId(str):
    """Custom field to handle ObjectId and convert it to string for Pydantic models."""
    
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, v: Any):
        if isinstance(v, ObjectId):
            return str(v)  # Convert ObjectId to string
        elif isinstance(v, str) and ObjectId.is_valid(v):
            return v  # Return the valid ObjectId string
        raise ValueError(f"Invalid ObjectId: {v}")

    @classmethod
    def json_schema(cls, **kwargs) -> Dict:
        """Override the default json_schema to represent ObjectId as a string in JSON."""
        return {"type": "string"}

main.py : entry point of appliacation

# main.py 
import uvicorn
from fastapi import FastAPI
from config.database import init_db,close_db
from contextlib import asynccontextmanager
from routes.todo_route import router as todo_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()   # Initialize the database
    yield
    await close_db()  # Close the database connection

app = FastAPI(lifespan=lifespan,title="To-Do App with FastAPI and Beanie ODM")


# Include routes
app.include_router(todo_router, prefix="/todos", tags=["Todos"])

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

Activate virtual env and  project by command:

uvicorn main:app --relaod

access appliaction through url : http://localhost:8000/docs#/