from fastapi import FastAPI, Depends, HTTPException, status from fastapi.middleware.cors import CORSMiddleware from sqlalchemy.orm import Session import uvicorn from typing import List, Optional from pydantic import BaseModel from datetime import datetime # Import database models and config from app.database.config import get_db, create_tables from app.database.models import Todo as TodoModel # Create tables if they don't exist # In production, you should use Alembic migrations instead create_tables() # Pydantic models for request and response class TodoBase(BaseModel): title: str description: Optional[str] = None completed: bool = False class TodoCreate(TodoBase): pass class TodoResponse(TodoBase): id: int created_at: datetime updated_at: Optional[datetime] = None class Config: from_attributes = True # Updated for Pydantic v2 (replaces orm_mode) # Create the FastAPI app app = FastAPI( title="Todo List API", description="A simple Todo List API built with FastAPI", version="0.1.0", ) # Configure CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Health endpoint @app.get("/health", tags=["Health"]) async def health_check(db: Session = Depends(get_db)): """ Health check endpoint to verify the API is running and database connection is working. """ try: # Test database connection db.execute("SELECT 1").first() return { "status": "healthy", "database": "connected", "timestamp": datetime.now().isoformat(), } except Exception as e: return { "status": "unhealthy", "database": "disconnected", "error": str(e), "timestamp": datetime.now().isoformat(), } # Root endpoint @app.get("/", tags=["Root"]) async def root(): """ Root endpoint that redirects to the API documentation. """ return {"message": "Welcome to Todo List API. Visit /docs for documentation."} # Todo API endpoints @app.post( "/todos", response_model=TodoResponse, status_code=status.HTTP_201_CREATED, tags=["Todos"], ) def create_todo(todo: TodoCreate, db: Session = Depends(get_db)): """ Create a new todo item. """ db_todo = TodoModel(**todo.model_dump()) db.add(db_todo) db.commit() db.refresh(db_todo) return db_todo @app.get("/todos", response_model=List[TodoResponse], tags=["Todos"]) def read_todos(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): """ Get all todo items with pagination. """ todos = db.query(TodoModel).offset(skip).limit(limit).all() return todos @app.get("/todos/{todo_id}", response_model=TodoResponse, tags=["Todos"]) def read_todo(todo_id: int, db: Session = Depends(get_db)): """ Get a specific todo item by ID. """ db_todo = db.query(TodoModel).filter(TodoModel.id == todo_id).first() if db_todo is None: raise HTTPException(status_code=404, detail="Todo not found") return db_todo @app.put("/todos/{todo_id}", response_model=TodoResponse, tags=["Todos"]) def update_todo(todo_id: int, todo: TodoCreate, db: Session = Depends(get_db)): """ Update a todo item. """ db_todo = db.query(TodoModel).filter(TodoModel.id == todo_id).first() if db_todo is None: raise HTTPException(status_code=404, detail="Todo not found") # Update todo item fields for key, value in todo.model_dump().items(): setattr(db_todo, key, value) db.commit() db.refresh(db_todo) return db_todo @app.delete( "/todos/{todo_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, tags=["Todos"], ) def delete_todo(todo_id: int, db: Session = Depends(get_db)): """ Delete a todo item. """ db_todo = db.query(TodoModel).filter(TodoModel.id == todo_id).first() if db_todo is None: raise HTTPException(status_code=404, detail="Todo not found") db.delete(db_todo) db.commit() return None # For local development if __name__ == "__main__": uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)