todolistapi-h6p2fk/main.py

from fastapi import FastAPI, Depends, HTTPException, status, Response
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 logging

# Import application config
from app.config import (
    API_TITLE,
    API_DESCRIPTION,
    API_VERSION,
    CORS_ORIGINS,
    HOST,
    PORT,
    DEBUG,
    HEALTH_CHECK_PATH,
    DETAILED_HEALTH_CHECK_PATH,
    HEALTH_CHECK_INCLUDE_DB,
    get_settings,
)

# Import database models and config
from app.database.config import get_db, create_tables, SessionLocal
from app.database.models import Todo as TodoModel

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 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=API_TITLE,
    description=API_DESCRIPTION,
    version=API_VERSION,
    docs_url="/docs",
    redoc_url="/redoc",
    openapi_url="/openapi.json",
)

# Configure CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# Health endpoints
@app.get(HEALTH_CHECK_PATH, tags=["Health"], status_code=200)
async def health_check():
    """
    Simple health check endpoint that always returns healthy.
    This is used by container orchestration systems to verify the app is running.

    The health check will always return a 200 OK status to indicate the application
    is running, even if some components (like the database) might be unavailable.
    """
    logger.info(f"Health check requested at {HEALTH_CHECK_PATH}")
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}


@app.get(DETAILED_HEALTH_CHECK_PATH, tags=["Health"])
async def detailed_health_check(response: Response):
    """
    Detailed health check endpoint that verifies API and database connectivity.

    This endpoint provides more detailed information about the state of various
    components of the application, including the database connection status.
    """
    logger.info(f"Detailed health check requested at {DETAILED_HEALTH_CHECK_PATH}")

    # Start with basic info
    health_data = {
        "status": "healthy",  # App is running, so it's healthy from an orchestration perspective
        "services": {
            "api": "running",
        },
        "timestamp": datetime.now().isoformat(),
        "config": {
            "db_check_enabled": HEALTH_CHECK_INCLUDE_DB,
            "environment": get_settings().get("APP_ENV"),
        },
    }

    # Check database if enabled
    if HEALTH_CHECK_INCLUDE_DB:
        db_status = "unknown"
        error_message = None

        # Create a new session for the health check
        db = SessionLocal()
        try:
            # Test database connection
            db.execute("SELECT 1").first()
            db_status = "connected"
        except Exception as e:
            db_status = "disconnected"
            error_message = str(e)
            # Don't change HTTP status code - we want health check to be 200 OK
            # to ensure container orchestration doesn't kill the app
        finally:
            db.close()

        health_data["services"]["database"] = db_status

        if error_message:
            health_data["services"]["database_error"] = error_message

    return health_data


# 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


# Application startup and shutdown events
@app.on_event("startup")
async def startup_event():
    """
    Function that runs when the application starts.
    """
    logger.info("Starting Todo List API")
    logger.info(f"API Version: {API_VERSION}")
    logger.info(f"Environment: {get_settings().get('APP_ENV')}")
    logger.info(f"Debug mode: {DEBUG}")

    # Log all available routes
    routes = []
    for route in app.routes:
        routes.append(f"{route.path} [{', '.join(route.methods)}]")

    logger.info(f"Available routes: {len(routes)}")
    for route in sorted(routes):
        logger.info(f"  {route}")


@app.on_event("shutdown")
async def shutdown_event():
    """
    Function that runs when the application shuts down.
    """
    logger.info("Shutting down Todo List API")


# For local development
if __name__ == "__main__":
    uvicorn.run("main:app", host=HOST, port=PORT, reload=DEBUG, log_level="info")