FastAPI Routing 실습 - 69

실습 문제
1. routers 디렉터리에 users.py 생성
2. users.py 특정 사용자 호출(path parameter) API: GET
3. users.py 특정 사용자 호출(query parameter) API: GET
4. users.py JWT를 생성해서 해당 사용자 암호 업데이트 API: PUT
5. users.py JWT를 생성해서 해당 사용자 삭제 API: DELETE

---

실습 풀이

프로젝트 구조

TodoApp
main.py database.py models.py	TodoApp/routers			TodoApp/company
	auth.py	users.py	todos.py	companyapis.py	dependencies.py

 

1. 라이브러리 및 DB 연결 그리고 에러 함수 생성

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy.orm import Session
from TodoApp.database import SessionLocal
import models
from .auth import get_password_hash, get_current_user, get_user_exception, verify_password


router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "Not found"}}
)


def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()
        
def http_exception(description):
    return HTTPException(status_code=404, detail=description)

def successful_response(description):
    return {
        'status': 200,
        'transaction': description
    }

 

2. GET 요청 API 생성

#전체 사용자 불러오기
@router.get("/")
async def read_all_users(db: Session = Depends(get_db)):
    user_model = db.query(models.Users).all()
    if not user_model:
        http_exception("Invalid user_id")
    return user_model

#특정 사용자 path로 불러오기
@router.get("/{user_id}")
async def user_by_path(user_id: int, db: Session = Depends(get_db)):
    user_model = db.query(models.Users).filter(models.Users.id == user_id).first()
    if user_model is not None:
        return user_model
    return http_exception("Invalid user_id")

#특정 사용자 query로 불러오기
@router.get("/user/")
async def user_by_query(user_id: int, db: Session = Depends(get_db)):
    user_model = db.query(models.Users).filter(models.Users.id == user_id).first()
    if user_model is not None:
        return user_model
    return http_exception("Invalid user_id")

 

3. PUT 요청 API 생성

#유효성 검사
class UserVerification(BaseModel):
    username: str
    password: str
    new_password: str
    
@router.put("/user/password")
async def user_password_change(user_verification: UserVerification, user: dict = Depends(get_current_user),
                               db: Session = Depends(get_db)):
    if user is None:
        raise get_user_exception()

    user_model = db.query(models.Users).filter(models.Users.id == user.get('id')).first()

    if user is not None:
        if user_verification.username == user_model.username and verify_password(user_verification.password,
                                                                                 user_model.hashed_password):
            user_model.hashed_password = get_password_hash(user_verification.new_password)
            db.add(user_model)
            db.commit()
            return successful_response('Update Successful')

 

4. DELETE 요청 API 생성

@router.delete("/user")
async def delete_user(user: dict = Depends(get_current_user), db: Session = Depends(get_db)):

    if user is None:
        raise get_user_exception()

    user_model = db.query(models.Users).filter(models.Users.id == user.get("id")).first()

    if user_model is None:
        return http_exception("Invalid user or request")

    db.query(models.Users).filter(models.Users.id == user.get("id")).delete()
    db.commit()

    return successful_response('Delete Successful')