Migrate Todo CRUD to SQLAlchemy ORM

main.py

@@ -1,11 +1,11 @@
 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()
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from database import engine
+from models import Todo
+from sqlalchemy.exc import SQLAlchemyError
 
 app = FastAPI()
 app.add_middleware(
@@ -16,18 +16,6 @@ app.add_middleware(
         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
 
@@ -45,83 +33,54 @@ def health():
 
 @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()
+    with Session(engine) as session:
+        statement = select(Todo)
+        todos = session.scalars(statement).all()
+        return todos
 
 @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()
+    with Session(engine) as session:
+        try:
+            new_todo = Todo(task = todo_data.task)
+            session.add(new_todo)
+            session.commit()
+            session.refresh(new_todo)
+            return new_todo
+        except SQLAlchemyError:
+            session.rollback()
+            raise
+
+
 @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() 
+    with Session(engine) as session:
+        try:
+            todo = session.get(Todo, todo_id)
+            if todo is None:
+                raise HTTPException(status_code = 404, detail = "Todo not found")
+            if update_data.task is not None:
+                todo.task = update_data.task
+            if update_data.completion is not None:
+                todo.completion = update_data.completion
+            session.commit()
+            session.refresh(todo)
+            return todo
+        except SQLAlchemyError:
+            session.rollback()
+            raise
 
-        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()
+    with Session(engine) as session:
+        try:
+            todo = session.get(Todo, todo_id)
+            if todo is None:
+                raise HTTPException(status_code = 404, detail = "Todo not found")
+            session.delete(todo)
+            session.commit()
+            return todo
+        except SQLAlchemyError:
+            session.rollback()
+            raise