FastAPI TodoApp(todos.py)

[주인장]

from fastapi import APIRouter, Depends, HTTPException, Path, status

from pydantic import BaseModel, Field

from models import Todos

from database import SessionLocal

from typing import Annotated

from sqlalchemy.orm import Session

from .auth import get_current_user

router = APIRouter()

def get_db():

    db = (

        SessionLocal()

    )  # 데이터베이스 세션을 생성 - 트랜잭션 단위로 데이터베이스에 접근

    try:

        yield db  # db를 반환

    finally:

        db.close()  # db를 닫음

# 의존성 주입(Dependency Injection) 기능을 활용하여 데이터베이스 세션 객체를 함수에 주입

db_dependency = Annotated[Session, Depends(get_db)]

user_dependency = Annotated[

    dict, Depends(get_current_user)

]  # 사용자 정보를 의존성 주입

# TodoRequest 클래스로 Todo 생성시 검증한 후 Todo 생성 요청을 처리

class TodoRequest(BaseModel):

    title: str = Field(min_length=3)  # 제목은 최소 3자 이상

    description: str = Field(

        min_length=3, max_length=100

    )  # 설명은 최소 3자 이상, 최대 100자 이하

    priority: int = Field(gt=0, lt=6)  # 우선순위는 1~5 사이

    complete: bool = Field(default=False)  # 완료 여부는 기본값이 False

#################################

# 엔드 포인트

#################################

# Todo 테이블에 있는 모든 레코드를 조회하여 반환

# status_code를 사용하여 200 OK 반환

@router.get("/", status_code=status.HTTP_200_OK)

async def read_all(user: user_dependency, db: db_dependency):

    if user is None:

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

            detail="인증되지 않은 사용자입니다.",

        )

    return (

        db.query(Todos).filter(Todos.owner_id == user.get("id")).all()

    )  # 사용자 ID로 조회

# todo_id를 받아 해당 Todo를 반환

# Path parameter를 사용하여 todo_id가 0보다 커야 함

@router.get("/todo/{todo_id}", status_code=status.HTTP_200_OK)

async def read_todo(

    user: user_dependency, db: db_dependency, todo_id: int = Path(gt=0)

):

    if user is None:

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

            detail="인증되지 않은 사용자입니다.",

        )

    todo_model = (

        db.query(Todos)

        .filter(Todos.id == todo_id)

        .filter(Todos.owner_id == user.get("id"))

        .first()

    )

    if todo_model is not None:

        return todo_model

    raise HTTPException(status_code=404, detail="해당 작업이 존재하지 않습니다.")

@router.post("/todo", status_code=status.HTTP_201_CREATED)

async def create_todo(

    user: user_dependency, db: db_dependency, todo_request: TodoRequest

):

    if user is None:

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

            detail="인증되지 않은 사용자입니다.",

        )

    todo_model = Todos(

        **todo_request.model_dump(),

        owner_id=user.get("id")  # 사용자 ID를 Todo 모델에 추가

    )  # TodoRequest 모델을 Todos 모델로 변환

    db.add(todo_model)  # 데이터베이스에 추가

    db.commit()  # 변경사항 저장

@router.put("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)

async def update_todo(

    user: user_dependency,

    db: db_dependency,

    todo_request: TodoRequest,

    todo_id: int = Path(gt=0),

):

    if user is None:

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

            detail="인증되지 않은 사용자입니다.",

        )

    # todo_id에 해당하는 Todo를 조회하기 위해 filter() 함수 사용

    todo_model = (

        db.query(Todos)

        .filter(Todos.id == todo_id)

        .filter(Todos.owner_id == user.get("id"))

        .first()

    )

    if todo_model is None:

        raise HTTPException(status_code=404, detail="해당 작업이 존재하지 않습니다.")

    # 수정을 위해 개별 설정 --> Todos(**todo_request.model_dump()) 사용하면 insert 처리

    todo_model.title = todo_request.title

    todo_model.description = todo_request.description

    todo_model.priority = todo_request.priority

    todo_model.complete = todo_request.complete

    db.add(todo_model)  # 동일한 id로 데이터베이스에 추가하면 수정 처리

    db.commit()

@router.delete("/todo/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)

async def delete_todo(

    user: user_dependency, db: db_dependency, todo_id: int = Path(gt=0)

):

    if user is None:

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

            detail="인증되지 않은 사용자입니다.",

        )

    todo_model = (

        db.query(Todos)

        .filter(Todos.id == todo_id)

        .filter(Todos.owner_id == user.get("id"))

        .first()

    )

    if todo_model is None:

        raise HTTPException(status_code=404, detail="해당 작업이 존재하지 않습니다.")

    db.delete(todo_model)  # 데이터베이스에서 삭제

    # db.query(Todos).filter(Todos.id == todo_id).delete() # 이렇게도 가능

    db.commit()