FastAPI TodoApp(admin.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(

    prefix='/admin',

    tags=['admin']

)

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)

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

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

# 엔드 포인트

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

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

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

    if user is None or user.get("user_role") != "admin":

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

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

        )

    return (

        db.query(Todos).all()

    )

@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 or user.get("user_role") != "admin":

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

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

        )

    todo_model = db.query(Todos).filter(Todos.id == todo_id).first()

    if todo_model is None:

        raise HTTPException(

            status_code=status.HTTP_404_NOT_FOUND,

            detail="Todo가 존재하지 않습니다."

        )

    db.query(Todos).filter(Todos.id == todo_id).delete()

    db.commit()