FastAPI TodoApp(users.py)

[주인장]

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

from pydantic import BaseModel, Field

from models import Todos, Users

from database import SessionLocal

from typing import Annotated

from sqlalchemy.orm import Session

from .auth import get_current_user

from passlib.context import CryptContext  # 암호화 모듈

router = APIRouter(prefix="/user", tags=["user"])

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)

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

bcrypt_context = CryptContext(

    schemes=["bcrypt"], deprecated="auto"

)  # bcrypt 알고리즘을 사용하는 bcrypt_context 객체 생성

class UserVerification(BaseModel):

    password: str

    new_password: str = Field(min_length=4)

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

# 엔드 포인트

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

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

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

    if user is None:

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

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

        )

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

@router.put("/password", status_code=status.HTTP_204_NO_CONTENT)

async def change_password(

    user: user_dependency,

    db: db_dependency,

    user_verification: UserVerification

):

    if user is None:

        raise HTTPException(

            status_code=status.HTTP_401_UNAUTHORIZED,

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

        )

    user_id = user.get("id")

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

    if not bcrypt_context.verify(user_verification.password, user_model.hashed_password):

        raise HTTPException(

            status_code=status.HTTP_400_BAD_REQUEST,

            detail="비밀번호가 일치하지 않습니다."

        )

    user_model.hashed_password = bcrypt_context.hash(user_verification.new_password)  # 새로운 비밀번호로 변경

    db.add(user_model)  # 변경된 사용자 정보를 데이터베이스에 추가  

    db.commit()         # 변경사항을 데이터베이스에 반영