fastapi-todolist/main.py

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import os
import mysql.connector
from dotenv import load_dotenv

load_dotenv()

app = FastAPI()
app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
)


def get_db_connection():
    connection = mysql.connector.connect(
            host = os.getenv("MYSQL_HOST"),
            port = int(os.getenv("MYSQL_PORT")),
            user = os.getenv("MYSQL_USER"),
            password = os.getenv("MYSQL_PASSWORD"),
            database = os.getenv("MYSQL_DATABASE"),
            )
    return connection


class todocreate(BaseModel):
    task: str

class todoupdate(BaseModel):
    completion: bool  | None = None
    task: str | None = None

class todoresponse(BaseModel):
    id: int
    task: str
    completion: bool
@app.get("/health")
def health():
    return {"status":"ok"}

@app.get("/todos", response_model=list[todoresponse])
def todos_get():
    connection = get_db_connection()
    cursor = connection.cursor(dictionary = True)
    try:
        cursor.execute("SELECT id, task, completion FROM todos")
        results = cursor.fetchall()
        return results
    finally:
            cursor.close()
            connection.close()

@app.post("/todos", response_model=todoresponse)
def todos_create(todo_data:todocreate):
    connection = get_db_connection()
    cursor = connection.cursor(dictionary = True)
    try:
        sql = "INSERT INTO todos (task) VALUES (%s)"
        params = (todo_data.task,)
        cursor.execute(sql,params)
        connection.commit()
        new_id = cursor.lastrowid
        new_todos = {
        "id": new_id,
        "task": todo_data.task, 
        "completion": False
        }
        return new_todos
    except mysql.connector.Error:
        connection.rollback()
        raise
    finally:
        cursor.close()
        connection.close()
@app.patch("/todos/{todo_id}", response_model=todoresponse)
def todos_patch(todo_id:int, update_data:todoupdate):
    connection = get_db_connection()
    cursor = connection.cursor(dictionary = True)
    try:
        sql = """
        UPDATE todos
        SET task = COALESCE(%s,task),
        completion = COALESCE(%s,completion)
        WHERE id = %s
        """
        params = (update_data.task,update_data.completion,todo_id,)
        cursor.execute(sql,params)
        connection.commit()
        select_params = (todo_id,)
        cursor.execute("SELECT id, task, completion FROM todos WHERE id = %s",select_params)
        result = cursor.fetchone() 

        if result is None:
            raise HTTPException(status_code = 404, detail = "Todo not found")
        return result
    except mysql.connector.Error:
        connection.rollback()
        raise
    finally:
        cursor.close()
        connection.close()

@app.delete("/todos/{todo_id}", response_model=todoresponse)
def todos_delete(todo_id:int):
    connection = get_db_connection()
    cursor = connection.cursor(dictionary = True)
    try:
        params = (todo_id,)
        cursor.execute("SELECT id,task,completion FROM todos WHERE id = %s",params)
        result = cursor.fetchone()
        if result is None:
            raise HTTPException(status_code = 404, detail = "Todo not found")
        sql = "DELETE FROM todos WHERE id = %s"
        cursor.execute(sql,params)
        connection.commit()
        return result
    except mysql.connector.Error:
        connection.rollback()
        raise
    finally:
        cursor.close()
        connection.close()