Initial TodoList MySQL delivery

2026-07-16 15:42:31 +08:00
commit fb0a53676f
11 changed files with 723 additions and 0 deletions
@@ -0,0 +1,127 @@
+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()