note_3.py

"""
note_3.py - User Management API (CRUD operations)

This file demonstrates:
- Pydantic models for data validation
- GET endpoints (all users, single user)
- POST endpoint (create user)
- PUT endpoint (update user)
- Error handling with HTTPException
- Working with JSON files as a database
- Response models

Run with: uvicorn note_3:app --reload
or: fastapi dev note_3.py
"""

from fastapi import FastAPI, HTTPException, status
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, EmailStr, Field
import json
from typing import Optional

app = FastAPI(
    title="User Management API",
    description="Complete user CRUD operations (without DELETE)",
    version="1.0.0"
)

# ---------- Pydantic models ----------
class User(BaseModel):
    """User model with validation"""
    id: int = Field(..., description="Unique user identifier", gt=0)
    username: str = Field(..., min_length=3, max_length=50, description="Username")
    email: str = Field(..., description="User email address")
    age: Optional[int] = Field(None, ge=0, le=150, description="User age")
    is_active: bool = Field(True, description="Whether user is active")
    roles: Optional[list[str]] = Field(None, description="User roles")

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "id": 1,
                    "username": "johndoe",
                    "email": "john@example.com",
                    "age": 30,
                    "is_active": True,
                    "roles": ["user", "admin"]
                }
            ]
        }
    }


class UserUpdate(BaseModel):
    """Model for updating user (all fields optional except ID)"""
    username: Optional[str] = Field(None, min_length=3, max_length=50)
    email: Optional[str] = None
    age: Optional[int] = Field(None, ge=0, le=150)
    is_active: Optional[bool] = None
    roles: Optional[list[str]] = None


# ---------- Paths ----------
HTML_PATH = "html/html_prueba.html"
JSON_PATH = "json/json_for_pydantic.json"


# ---------- Helper functions ----------
def read_users_data() -> dict:
    """Read users from JSON file"""
    try:
        with open(JSON_PATH, "r", encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Database file not found: {JSON_PATH}"
        )
    except json.JSONDecodeError:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Database file contains invalid JSON"
        )


def write_users_data(data: dict) -> None:
    """Write users to JSON file"""
    try:
        with open(JSON_PATH, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=4, ensure_ascii=False)
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to write to database: {str(e)}"
        )


# ---------- HTML endpoint ----------
@app.get("/", response_class=HTMLResponse)
async def root() -> str:
    """
    Serve the main HTML page.
    """
    try:
        with open(HTML_PATH, "r", encoding="utf-8") as f:
            html = f.read()
        return html
    except FileNotFoundError:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="HTML file not found"
        )


# ---------- Return all users ----------
@app.get("/users", response_model=dict)
async def get_users():
    """
    Retrieve all users from the database.
    
    - **Returns**: Dictionary containing all users
    """
    return read_users_data()


# ---------- Return a single user by ID ----------
@app.get("/users/{id_requested}", response_model=User)
async def get_user(id_requested: int):
    """
    Retrieve a specific user by ID.
    
    - **id_requested**: The ID of the user to retrieve
    - **Returns**: User object if found
    - **Raises**: 404 if user not found
    """
    data = read_users_data()

    # Find user by ID
    for user in data["users"]:
        if user["id"] == id_requested:
            return User(**user)

    # Raise 404 if not found
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"User with ID {id_requested} not found"
    )


# ---------- Create a new user ----------
@app.post(
    "/users",
    response_model=User,
    status_code=status.HTTP_201_CREATED
)
async def create_user(new_user: User):
    """
    Create a new user.
    
    - **new_user**: User data (validated by Pydantic)
    - **Returns**: The created user
    - **Raises**: 400 if ID or email already exists
    """
    data = read_users_data()

    # Check for duplicate id or email
    for user in data["users"]:
        if user["id"] == new_user.id:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"User with ID {new_user.id} already exists"
            )
        if user["email"].lower() == new_user.email.lower():
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"User with email {new_user.email} already exists"
            )

    # Add new user
    data["users"].append(new_user.model_dump())
    write_users_data(data)

    return new_user


# ---------- Update a user (PUT) ----------
@app.put("/users/{id_requested}", response_model=User)
async def update_user(id_requested: int, updated_user: User):
    """
    Update a user completely (PUT - replaces entire user).
    
    - **id_requested**: The ID of the user to update
    - **updated_user**: Complete new user data
    - **Returns**: The updated user
    - **Raises**: 404 if user not found, 400 if ID mismatch or email conflict
    """
    data = read_users_data()

    # Verify ID consistency
    if updated_user.id != id_requested:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"ID in path ({id_requested}) doesn't match ID in body ({updated_user.id})"
        )

    # Find user index
    user_index = None
    for index, user in enumerate(data["users"]):
        if user["id"] == id_requested:
            user_index = index
            break

    if user_index is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User with ID {id_requested} not found"
        )

    # Check for email conflicts (excluding current user)
    for index, user in enumerate(data["users"]):
        if index != user_index and user["email"].lower() == updated_user.email.lower():
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"Email {updated_user.email} is already used by another user"
            )

    # Update user
    data["users"][user_index] = updated_user.model_dump()
    write_users_data(data)

    return updated_user


# ---------- Health check ----------
@app.get("/health")
async def health_check():
    """
    Health check endpoint.
    """
    return {
        "status": "healthy",
        "service": "User Management API",
        "version": "1.0.0"
    }


# ---------- Main ----------
if __name__ == "__main__":
    print("To run this app, use:")
    print("  uvicorn note_3:app --reload")
    print("  or")
    print("  fastapi dev note_3.py")