FastAPI - MiniProject - 학습성과

프로젝트를 하면서 알게 된 것

from_attributes = True

• Pydantic V1에서는 orm_mode = True
• ORM(예: Tortoise ORM, SQLAlchemy 등) 모델을 Pydantic 모델로 변환할 때 필요한 옵션

# schema
from pydantic import BaseModel, ConfigDict

class UserResponse(BaseModel):
    id: int
    username: str
    email: str

    model_config = ConfigDict(from_attributes=True)
    
   or 
   
   # pydantic v1
    class Config: 
        from_attributes = True

from app.models.user import User  # Tortoise ORM 모델
from app.schemas.user import UserResponse  # Pydantic 모델

user = await User.get(id=1)
return UserResponse.model_validate(user)

• ORM 모델은 dict가 아니라 객체(클래스 인스턴스) 이기 때문에
  • “객체의 속성(attribute)”을 읽을 수 있게 해주는 설정이 필요

---

Tortoise-ORM

# app/models/user
from tortoise import fields
from tortoise.models import Model

class User(Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(50, unique=True)
    email = fields.CharField(100, unique=True)
    password = fields.CharField(255)
    created_at = fields.DatetimeField(auto_now_add=True)

    def __str__(self):
        return self.username

# app/repositories/user_repo.py
from app.models.user import User

class UserRepository:
    @staticmethod
    async def create_user(username: str, email: str, password: str):
        return await User.create(username=username, email=email, password=password)

    @staticmethod
    async def get_by_username(username: str):
        return await User.get_or_none(username=username)

• User
  • 데이터베이스의 users 테이블과 직접 연결된 Tortoise-ORM 모델 클래스
• class UserRepository:
  • 이 클래스는 데이터베이스에 접근하는 로직만 담당하는 계층
  • "비즈니스 로직”(service)과 “DB 조작 로직”(repository)을 분리하기 위해 사용
• @staticmethod
  • 인스턴스를 만들지 않고 바로 호출할 수 있도록 하는 정적 메서드
  • UserRepository() 객체를 만들지 않아도 호출 가능
    • await UserRepository.create_user(...) 가능
• return await User.create(username=username, email=email, password=password)
  • User.create() : Tortoise-ORM의 비동기 DB 삽입 함수
    • DB에 새로운 유저 레코드를 추가하고, 생성된 User 객체를 반환
    • await 키워드는 비동기 I/O(즉, DB 작업이 완료될 때까지 기다림)를 의미
• return await User.get_or_none(username=username)
  • User.get_or_none

Tortoise ORM의 ORM 패턴과 비동기 흐름

await User.get_or_none(username=username)
await User.create(**data)
await User.filter(email=email).update(...)

• get_or_none() : 존재하지 않으면 None 반환 (try/except 필요 X)
• filter().update() : SQL UPDATE 직접 수행 (성능 좋음)
• Prefetch 나 select_related() : 관계 데이터 미리 불러오기 (JOIN)
  • Tortoise는 “비동기 ORM”이기 때문에, await를 잊으면 작동은 하지만 결과가 틀릴 수 있음

PostgreSQL과 ORM의 관계

• Tortoise ORM은 결국 PostgreSQL 쿼리를 자동 생성하는 도구

개념	ORM 표현	SQL 표현
SELECT	User.all()	SELECT * FROM user;
WHERE	User.filter(username="kihoon")	WHERE username='kihoon'
JOIN	select_related("profile")	JOIN profile ON ...

---

app/core/security.py

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.now() + (expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)

• pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  • CryptContext
    • passlib 라이브러리의 객체로, 비밀번호를 안전하게 암호화/검증하는 도구
  • schemes=["bcrypt"]:
    • bcrypt 알고리즘을 사용하겠다는 뜻 (현재 가장 널리 쓰이는 안전한 해싱 방식 중 하나)
  • deprecated="auto":
    • 이전 버전의 해싱 알고리즘을 자동으로 “낡은 방식”으로 처리하도록 지정
    • 나중에 bcrypt에서 다른 알고리즘으로 교체할 때도 자연스럽게 마이그레이션 가능

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

• 사용자가 로그인할 때 입력한 평문 비밀번호(plain_password)를
  • DB에 저장된 암호화된 비밀번호(hashed_password)와 비교함, 내부적으로 bcrypt 검증이 일어남
    • 즉, 평문을 복호화하지 않고, 다시 같은 방식으로 해싱한 뒤 결과를 비교
    • 일치: True / 불일치: False

def get_password_hash(password):
    return pwd_context.hash(password)

• 회원가입 시 평문 비밀번호를 해싱하여 DB에 저장하기 위한 함수
• 같은 문자열이라도 매번 랜덤한 salt 가 포함되어 해시 결과가 다르게 나옴

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    expire = datetime.now() + (expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)

• 사용자가 로그인 성공 시 발급받는 JWT 토큰을 생성
  • to_encode = data.copy() : 입력 데이터 복사
  • expire = datetime.now() + (expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES))
    • expires_delta가 직접 주어지면 그걸 쓰고, 없으면 .env에서 설정한 기본값 사용.
  • to_encode.update({"exp": expire}) : payload에 만료시간 추가
  • jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    • SECRET_KEY로 서명하여 안전하게 암호화된 JWT 문자열을 생성

---

app/core/dependencies.py

• JWT 토큰을 해석해 현재 로그인한 사용자 정보를 얻는 함수
  • “현재 사용자 가져오기(get_current_user)” 의존성

from fastapi import Depends, HTTPException, status
from jose import jwt, JWTError
from app.core.config import settings
from app.repositories.user_repo import UserRepository

async def get_current_user(token: str):
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = await UserRepository.get_by_username(username)
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return user

• async def get_current_user(token: str):
  • 매개변수로 token(JWT 액세스 토큰 문자열)을 받음
    • 이 토큰은 보통 HTTP 요청의 Authorization 헤더에서 전달
    • ex: "Bearer eyJhbGciOi..."
• payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
  • jwt.decode() : 토큰을 해석(decode) 하여 안에 들어있는 정보를 추출합니다.
  • settings.SECRET_KEY : 토큰을 서명할 때 사용된 비밀키 (발급 시와 같아야 함).
  • settings.ALGORITHM : 암호화 알고리즘 (예: "HS256").
    • 디코드 성공하면,
      • "sub": "kihoon", # username
      • "exp": 1730000000 # 만료 시간 (timestamp)
• username: str = payload.get("sub")
  • JWT 토큰 안에는 보통 "sub" 키에 사용자 식별자(username)를 저장

except JWTError:
    raise HTTPException(status_code=401, detail="Invalid token")

• 비밀키가 다르거나, 토큰이 만료되었거나, 형식이 잘못된 경우 JWTError 가 발생

user = await UserRepository.get_by_username(username)
if not user:
    raise HTTPException(status_code=401, detail="User not found")
return user

• Tortoise ORM을 사용하는 UserRepository 클래스에서 username
  • 실제 DB의 사용자 정보를 가져옵니다.
• DB에 사용자가 존재하지 않으면 → 또다시 401 오류 발생.
• 정상적으로 찾으면 user 객체(모델 인스턴스)를 반환

---

app/api/v1/diary.py

from fastapi import APIRouter, Depends, HTTPException
from app.core.dependencies import get_current_user
from app.models.diary import Diary

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

@router.put("/{diary_id}")
async def update_diary(diary_id: int, content: str, current_user=Depends(get_current_user)):
    diary = await Diary.get_or_none(id=diary_id)
    if not diary:
        raise HTTPException(status_code=404, detail="Diary not found")

    if diary.user_id != current_user.id:
        raise HTTPException(status_code=403, detail="Not authorized")

    diary.content = content
    await diary.save()
    return {"message": "Updated successfully"}

• 로그인한 사용자가 자신의 일기만 수정할 수 있게 하는 권한 처리 포함 CRUD
• Depends(get_current_user)
  • JWT 토큰을 통해 현재 로그인한 사용자 정보를 가져오는 의존성 주입.
• Diary
  • Tortoise ORM의 일기 모델 클래스, DB의 diary 테이블과 연결됨.

@router.put("/{diary_id}")
async def update_diary(diary_id: int, content: str, current_user=Depends(get_current_user)):

• diary_id: 수정할 일기의 ID
• content: 새로 수정할 일기 내용 (쿼리 파라미터나 JSON body로 전달)
• current_user: Depends(get_current_user)
  • 덕분에 자동으로 로그인한 사용자 정보가 들어옴 (JWT 토큰에서 추출된 User 객체)

diary = await Diary.get_or_none(id=diary_id)
if not diary:
    raise HTTPException(status_code=404, detail="Diary not found")

• Diary.get_or_none()
  • Tortoise ORM 메서드 → 지정된 id를 가진 일기 레코드를 가져옴. 없으면 None 반환.
    • 없을 경우 404 Not Found 예외 발생.

if diary.user_id != current_user.id:
    raise HTTPException(status_code=403, detail="Not authorized")

• 일기의 작성자(diary.user_id)와 현재 로그인한 사용자(current_user.id)가 다르면
  • 수정 권한이 없으므로 403 Forbidden 반환.

diary.content = content
await diary.save()

• await diary.save() : DB에 변경사항 반영 (UPDATE SQL 실행).

---

HTTPBearer / OAuth2PasswordBearer

• 공통점
  • FastAPI의 fastapi.security 모듈에서 제공하는 “보안 스킴(Security Scheme)” 클래스
  • HTTP 요청 헤더(Authorization)에서 토큰을 추출하는 역할을 함
    • Authorization: Bearer <토큰>

HTTPBearer

• “헤더에서 Bearer 토큰 문자열을 꺼내주는 역할만” 함
  • 토큰의 내용이 JWT인지, 유효한지 등은 직접 검증해야 함

from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer

security = HTTPBearer()

@app.get("/secure")
async def secure_endpoint(credentials=Depends(security)):
    token = credentials.credentials  # 단순히 Authorization 헤더에서 토큰 꺼냄
    if token != "mysecrettoken":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"msg": "Access granted"}

• 단순히 Authorization: Bearer <token> 을 읽음.
  • Swagger UI에서 “Authorize” 버튼을 누르면 Bearer 토큰 입력칸이 뜸.
  • 토큰의 검증 로직은 전부 직접 작성해야 함.

OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

• "요청의 HTTP 헤더에서 토큰을 자동으로 추출해주는 의존성"
  • FastAPI는 OAuth2PasswordBearer를 통해
    • Authorization: Bearer <토큰> 형식의 헤더를 자동으로 감지하고, 해당 토큰을 함수에 전달

OAuth2PasswordBearer(tokenUrl="/auth/login") 흐름

• 사용자가 로그인 요청을 보냄 -> /auth/login에서 아이디와 비밀번호를 검증
  • JWT 토큰(Access Token)을 발급받음
    • 이후 API 요청 시, 사용자는 헤더에 다음처럼 토큰을 포함해서 요청을 보냄
      • Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6...
  • 즉, OAuth2PasswordBearer이 작동해서,
    • FastAPI가 이 헤더를 자동으로 인식하고 토큰만 추출
• tokenUrl="/auth/login"의 의미
  • “토큰을 발급받는 엔드포인트 경로” 를 명시적으로 알려주는 설정값
  • 즉, Swagger 문서(/docs)에서 "Authorize" 버튼을 눌렀을 때
    • FastAPI가 어디로 로그인 요청을 보내야 할지 이 값을 보고 알아냄

예시

# 1

@router.post("/logout")
async def logout(token: str = Depends(oauth2_scheme)):
    return await AuthService.logout(token)

# 요청 헤더의 Bearer ... 토큰이 token 변수로 들어오게 됨

### 요약
GET /diary
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6...

이렇게 클라이언트가 요청을 보내면 Depends(oauth2_scheme) 덕분에
"eyJhbGciOi..." 부분만 token 인자로 자동으로 들어감

# 2

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    # 여기서 토큰 검증 (JWT decode)
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"username": username}

---

JWT

# auth_service.py
@staticmethod
    async def login(username: str, password: str):
        user = await UserRepository.get_by_username(username)
        if not user or not verify_password(password, user.password_hash):
            raise HTTPException(status_code=401, detail="Invalid credentials")

        access_token = create_access_token(data={"sub": user.username})
        return {"access_token": access_token, "token_type": "bearer"}

# dependencies.py
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

• auth_service.py에서 로그인할 때 넣었던 sub값을 dependencies.py에서 JWT토큰을 검증
  • jwt.decode()를 통해 SECRET_KEY와 ALGORITHM으로 디코딩하면
    • 원래 들어있던 데이터(payload)를 볼 수 있음

---

---

자잘한 내용

Diary.get_or_none(id=diary_id)

• diary = await Diary.get_or_none(id=diary_id)
  

current_user: User = Depends(get_current_user)

from fastapi import Depends
from app.models.user import User
from app.core.dependencies import get_current_user

@router.get("/myinfo")
async def get_my_info(current_user: User = Depends(get_current_user)):
    return {"username": current_user.username}

• “이 엔드포인트가 실행되기 전에, 먼저 get_current_user() 함수를 실행해서
  • 그 결과를 current_user 변수에 자동으로 넣어줘.”

---

Header

• 클라이언트(예: 브라우저, Postman, 프론트엔드 앱)가 서버로 요청을 보낼 때
  • 추가적인 메타데이터나 인증 정보를 함께 전달하는 방법

GET /users HTTP/1.1
Host: example.com
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6Ikp...
Content-Type: application/json
User-Agent: Mozilla/5.0

• Authorization, Content-Type, User-Agent 같은 부분이 헤더(Header)
  • 요청의 본문(body)과는 별개로 서버에
  • “이 요청은 어떤 종류의 데이터고, 누가 보낸 것이며, 어떤 인증을 거쳤는지” 등을 알려주는 역할

---