e01f3a4d57
Create a simple Todo API with FastAPI and SQLite with CRUD functionality, health check, error handling, and API documentation.
137 lines
3.3 KiB
Python
137 lines
3.3 KiB
Python
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, Path, Query, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.exceptions import TodoException
|
|
from app.database.session import get_db
|
|
from app.models import Todo as TodoModel
|
|
from app.schemas import Todo as TodoSchema
|
|
from app.schemas import TodoCreate, TodoUpdate
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post(
|
|
"/",
|
|
response_model=TodoSchema,
|
|
status_code=status.HTTP_201_CREATED,
|
|
summary="Create a new todo item"
|
|
)
|
|
def create_todo(
|
|
todo_in: TodoCreate,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Create a new todo item.
|
|
"""
|
|
db_todo = TodoModel(**todo_in.model_dump())
|
|
db.add(db_todo)
|
|
db.commit()
|
|
db.refresh(db_todo)
|
|
return db_todo
|
|
|
|
|
|
@router.get(
|
|
"/",
|
|
response_model=List[TodoSchema],
|
|
summary="Get all todo items"
|
|
)
|
|
def read_todos(
|
|
skip: int = Query(0, ge=0, description="Number of items to skip"),
|
|
limit: int = Query(
|
|
100,
|
|
ge=1,
|
|
le=100,
|
|
description="Maximum number of items to return"
|
|
),
|
|
completed: Optional[bool] = Query(None, description="Filter by completed status"),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Get all todo items with optional filtering and pagination.
|
|
"""
|
|
query = db.query(TodoModel)
|
|
|
|
# Apply filter if completed status is provided
|
|
if completed is not None:
|
|
query = query.filter(TodoModel.completed == completed)
|
|
|
|
todos = query.offset(skip).limit(limit).all()
|
|
return todos
|
|
|
|
|
|
@router.get(
|
|
"/{todo_id}",
|
|
response_model=TodoSchema,
|
|
summary="Get a specific todo item"
|
|
)
|
|
def read_todo(
|
|
todo_id: int = Path(..., gt=0, description="The ID of the todo item"),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Get a specific todo item by ID.
|
|
"""
|
|
todo = db.query(TodoModel).filter(TodoModel.id == todo_id).first()
|
|
if todo is None:
|
|
raise TodoException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Todo with ID {todo_id} not found"
|
|
)
|
|
return todo
|
|
|
|
|
|
@router.patch(
|
|
"/{todo_id}",
|
|
response_model=TodoSchema,
|
|
summary="Update a todo item"
|
|
)
|
|
def update_todo(
|
|
todo_id: int = Path(..., gt=0, description="The ID of the todo item"),
|
|
todo_in: TodoUpdate = ...,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Update a todo item.
|
|
"""
|
|
todo = db.query(TodoModel).filter(TodoModel.id == todo_id).first()
|
|
if todo is None:
|
|
raise TodoException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Todo with ID {todo_id} not found"
|
|
)
|
|
|
|
update_data = todo_in.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(todo, field, value)
|
|
|
|
db.add(todo)
|
|
db.commit()
|
|
db.refresh(todo)
|
|
return todo
|
|
|
|
|
|
@router.delete(
|
|
"/{todo_id}",
|
|
status_code=status.HTTP_204_NO_CONTENT,
|
|
response_model=None,
|
|
summary="Delete a todo item"
|
|
)
|
|
def delete_todo(
|
|
todo_id: int = Path(..., gt=0, description="The ID of the todo item"),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Delete a todo item.
|
|
"""
|
|
todo = db.query(TodoModel).filter(TodoModel.id == todo_id).first()
|
|
if todo is None:
|
|
raise TodoException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Todo with ID {todo_id} not found"
|
|
)
|
|
|
|
db.delete(todo)
|
|
db.commit()
|
|
return None |