[WAFFLE] 05-5 Assignment 2 Review

8 minute read

Published: October 05, 2024

In this post, review of assignment 2 are written.

Code explanation

/

// .env.local
DB_DIALECT=mysql
DB_DRIVER=pymysql
DB_HOST=localhost
DB_PORT=3307
DB_USER=wapang-admin
DB_PASSWORD=abcd
DB_DATABASE=wapang

wapang

# main.py

## 라우터 연결
app.include_router(api_router, prefix="/api")

api_router는 wapang.api 모듈에서 가져온 라우터 객체로, API 엔드포인트들을 정의합니다.
prefix=”/api”는 모든 API 엔드포인트 앞에 /api를 추가합니다. 예: /api/users, /api/items.

# api.py

## 라우터 연결
from fastapi import APIRouter

from wapang.app.user.views import user_router
..

api_router = APIRouter()

api_router.include_router(user_router, prefix="/users", tags=["users"])
..

# settings.py

import os

from pydantic_settings import BaseSettings

# TODO 환경 변수로 설정된 환경을 가져옵니다.
# 가능한 값은 "local", "test", "prod" 이며, 기본값은 "local" 입니다.
# 환경 변수는 어떻게 바꿀 수 있을까요?
ENV = os.getenv("ENV", "local") 
assert ENV in ("local", "test", "prod")


class Settings(BaseSettings):
    # TODO property 데코레이터는 메서드를 필드처럼 사용할 수 있게 해줍니다.
    # 예컨데, SETTINGS.is_local() 대신 SETTINGS.is_local 으로 사용할 수 있습니다.
    @property
    def is_local(self) -> bool:
        return ENV == "local"

    @property
    def is_prod(self) -> bool:
        return ENV == "prod"

    @property
    def env_file(self) -> str:
        return f".env.{ENV}"


# TODO SETTINGS 는 런타임에 변경되지 않으므로 미리 초기화해둡니다.
SETTINGS = Settings()

os.getenv(“ENV”, “local”) : 시스템 환경 변수 ENV의 값을 읽고 값이 없으면 “local” 사용
assert : ENV가 3개 중 하나인지 확인

wapang/app/user

# models.py

class User(Base):
    __tablename__ = "user"

    id: Mapped[intpk]
    username: Mapped[str] = mapped_column(String(20), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
    password: Mapped[str] = mapped_column(String(20))
    address: Mapped[str | None] = mapped_column(String(100))
    ..

sqlalchemy.orm을 이용하여 User table 정의

# store.py

class UserStore:
    def __init__(self, session: Annotated[Session, Depends(get_db_session)]) -> None:
        self.session = session

    def add_user(self, username: str, password: str, email: str) -> User:
        if self.get_user_by_username(username):
            raise UsernameAlreadyExistsError()

        if self.get_user_by_email(email):
            raise EmailAlreadyExistsError()

        user = User(username=username, password=password, email=email)
        self.session.add(user)
        return user

    def get_user_by_username(self, username: str) -> User | None:
        return self.session.scalar(select(User).where(User.username == username))
..

UserStore 클래스 : 사용자 관련 데이터베이스 작업을 담당하는 클래스입니다.
- init : FastAPI의 Depends(get_db_session)를 통해 SQLAlchemy Session 객체를 주입받습니다.
- add_user : 새 사용자를 데이터베이스에 추가합니다.
- get_user_by_username : 사용자 이름으로 사용자 데이터를 조회합니다. SQLAlchemy select 구문으로 User 테이블에서 username이 일치하는 사용자 조회.

# service.py

class UserService:
    def __init__(self, user_store: Annotated[UserStore, Depends()]) -> None:
        self.user_store = user_store

    def add_user(self, username: str, password: str, email: str):
        self.user_store.add_user(username=username, password=password, email=email)

    def get_user_by_username(self, username: str) -> User | None:
        return self.user_store.get_user_by_username(username)
    ..

# views.py

user_router = APIRouter()


def login_with_header(
    user_service: Annotated[UserService, Depends()],
    x_wapang_username: Annotated[str, Header()] = "",
    x_wapang_password: Annotated[str, Header()] = "",
) -> User:
    user = user_service.get_user_by_username(x_wapang_username)
    if not user or user.password != x_wapang_password:
        raise HTTPException(
            status_code=HTTP_401_UNAUTHORIZED, detail="Invalid credentials"
        )
    return user


@user_router.post("/signup", status_code=HTTP_201_CREATED)
def signup(
    signup_request: UserSignupRequest, user_service: Annotated[UserService, Depends()]
):
    user_service.add_user(
        signup_request.username, signup_request.password, signup_request.email
    )
    return "Success"

..

login_with_header : 헤더 정보를 사용하여 사용자를 인증합니다.

wapang/app/user/dto

# requests.py

USERNAME_PATTERN = re.compile(r"^[a-zA-Z0-9_-]{3,20}$")


def validate_username(value: str | None) -> str | None:
    if value is None:
        return value
    if not re.match(USERNAME_PATTERN, value):
        raise InvalidFieldFormatError()
    return value
..

class UserSignupRequest(BaseModel):
    username: Annotated[str, AfterValidator(validate_username)]
    # TODO 과제1 에서는 연습을 위해 외부 라이브러리를 사용하지 말라고 말씀드렸지만,
    # 실전에서는 당연히 좋은 라이브러리가 있다면 적절히 사용하는 것이 좋습니다.
    # 개발자라면 누구나 알법한 명언이 있죠. "바퀴를 재발명하지 마세요."
    email: EmailStr
    password: Annotated[str, AfterValidator(validate_password)]


class UserUpdateRequest(BaseModel):
    email: EmailStr | None = None
    address: Annotated[str | None, AfterValidator(validate_address)] = None
    phone_number: Annotated[str | None, AfterValidator(validate_phone_number)] = None

validate_username : 이름 유효성 검증 함수
UserSignupRequest : 회원가입 요청 데이터를 처리하는 DTO.
- username: validate_username 함수로 검증.
- email: Pydantic의 EmailStr 타입으로 이메일 형식을 자동 검증.
UserUpdateRequest : 사용자 정보 업데이트 요청 데이터를 처리하는 DTO.

# responses.py

class MyProfileResponse(BaseModel):
    username: str
    email: str
    address: str | None
    phone_number: str | None

    @staticmethod
    def from_user(user: User) -> "MyProfileResponse":
        return MyProfileResponse(
            username=user.username,
            email=user.email,
            address=user.address,
            phone_number=user.phone_number,
        )

wapang/common

item, order, store, user에서 공통으로 사용하는 error와 validator 정의

# utils.py

def validate_phone_number(value: str | None) -> str | None:
    if value is None:
        return None
    if not value.startswith("010") or not len(value) == 11 or not value.isdigit():
        raise InvalidFieldFormatError()
    return value

# errors.py
class WapangHttpException(HTTPException):
    def __init__(self, status_code: int, detail: str) -> None:
        super().__init__(status_code=status_code, detail=detail)


class InvalidFieldFormatError(WapangHttpException):
    def __init__(self) -> None:
        super().__init__(HTTP_400_BAD_REQUEST, "Invalid field format")

wapang/database

# connection.py
from typing import Any, Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

from wapang.database.settings import DB_SETTINGS


class DatabaseManager:
    def __init__(self):
        # TODO pool 이 뭘까요?
        # pool_recycle 은 뭐고 왜 28000으로 설정해두었을까요?
        self.engine = create_engine(
            DB_SETTINGS.url,
            pool_recycle=28000,
            pool_pre_ping=True,
        )
        self.session_factory = sessionmaker(bind=self.engine, expire_on_commit=False)


# TODO 아래 함수를 이용해서 Session 종속성을 주입받을 수 있습니다.
# 이렇게 했을 때의 장점은 무엇일까요? 다른 방법은 없을까요?
# 한 번 생각해보세요.
def get_db_session() -> Generator[Session, Any, None]:
    session = DatabaseManager().session_factory() #1
    try:
        yield session #2
        session.commit() #3
    except Exception as e:
        session.rollback() #3
        raise e
    finally:
        session.close() #4

create_engine : 데이터베이스 연결을 관리하는 SQLAlchemy의 핵심 객체.
get_db_session : yield 키워드를 사용해 세션 객체를 라우터나 서비스로 주입. 함수가 종료되면 finally 블록에서 세션이 자동으로 닫힙니다.
1. 세션 생성 : DatabaseManager의 session_factory를 통해 세션을 생성.
2. yield로 주입 : 세션을 호출하는 라우터나 서비스로 전달.
3. 커밋 또는 롤백 :
  - 성공 : 작업 완료 후 session.commit() 호출.
  - 예외 : session.rollback()으로 처리.
4. 세션 닫기 : 모든 작업이 끝난 뒤 session.close()로 세션 리소스를 정리.

# settings.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from wapang.settings import SETTINGS


class DatabaseSettings(BaseSettings):
    dialect: str = ""
    driver: str = ""
    host: str = ""
    port: int = 0
    user: str = ""
    password: str = ""
    database: str = ""

    @property
    def url(self) -> str:
        return f"{self.dialect}+{self.driver}://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}"

    model_config = SettingsConfigDict(
        case_sensitive=False,
        env_prefix="DB_",
        env_file=SETTINGS.env_file,
    )


DB_SETTINGS = DatabaseSettings()

Project Flow

Overall Flow

클라이언트 요청:
- 클라이언트가 특정 API 엔드포인트(예: /api/users/me)로 HTTP 요청을 보냅니다.
- 요청에는 HTTP 메서드(GET, POST, PATCH 등), 헤더, 본문 데이터 등이 포함됩니다.
FastAPI 라우터:
- 요청은 FastAPI 라우터(views.py)에서 정의된 엔드포인트로 라우팅됩니다.
- 엔드포인트는 요청을 처리하기 위해 서비스 계층(Service Layer) 및 데이터 계층(Data Layer)과 통신합니다.
서비스 계층:
- service.py는 비즈니스 로직을 처리하며, 데이터 계층(store.py)과 상호작용합니다.
- 예: 사용자 조회, 추가, 업데이트 요청을 처리.
데이터 계층:
- store.py는 데이터베이스 작업을 직접적으로 수행하며, SQLAlchemy를 통해 데이터베이스와 연결됩니다.
데이터베이스 연결:
- connection.py에서 정의된 get_db_session을 통해 데이터베이스 세션을 생성하고 관리합니다.
- 데이터베이스 설정은 settings.py에서 로드됩니다.
응답 생성:
- 서비스 계층에서 처리된 결과를 DTO(responses.py)로 변환하여 클라이언트에 응답합니다.

import relationship

main.py 역할: FastAPI 애플리케이션의 진입점. 라우터(api.py)를 포함하여 API 엔드포인트를 정의. 예외 처리, 미들웨어 등을 설정. 호출 관계: api.py → 프로젝트 전반의 라우터 통합.
api.py 역할: 여러 도메인별 라우터(user/views.py, store/views.py, item/views.py)를 하나의 APIRouter 객체로 통합. 각 도메인의 엔드포인트를 /api/<도메인> 형태로 제공. 호출 관계: user/views.py, store/views.py 등에서 정의된 라우터를 통합.
user/views.py 역할: 사용자 관련 API 엔드포인트 정의. 예: /signup, /me, /me(PATCH). 주요 호출 흐름: 라우터: signup: 사용자 추가 요청 처리. 호출 관계: UserService.add_user → UserStore.add_user → 데이터베이스. me: 현재 사용자 프로필 조회. 호출 관계: UserService.get_user_by_username → UserStore.get_user_by_username → 데이터베이스. update_me: 사용자 정보 업데이트. 호출 관계: UserService.update_user → UserStore.update_user → 데이터베이스.
user/service.py 역할: 비즈니스 로직 처리. 사용자 관련 데이터 조작 요청을 데이터 계층(store.py)으로 전달. 호출 관계: store.py: add_user → UserStore.add_user. get_user_by_username → UserStore.get_user_by_username. update_user → UserStore.update_user.
user/store.py 역할: 데이터베이스와 직접 통신. SQLAlchemy를 사용하여 사용자 데이터를 추가, 조회, 업데이트. 호출 관계: connection.py: get_db_session으로 데이터베이스 세션을 받아 데이터 작업 수행. SQLAlchemy ORM: 데이터베이스와 직접 상호작용.
database/connection.py 역할: SQLAlchemy 엔진과 세션을 생성 및 관리. FastAPI 의존성 주입(Depends)을 통해 세션을 주입. 호출 관계: store.py, get_db_session 호출.
database/settings.py 역할: 데이터베이스 연결 정보를 관리. 환경 변수 또는 .env 파일에서 설정 값을 로드. 호출 관계: connection.py: 데이터베이스 URL을 제공(DB_SETTINGS.url).
user/dto/requests.py 역할: 요청 데이터를 정의하는 DTO. 입력 데이터 유효성을 검증. 호출 관계: views.py: UserSignupRequest, UserUpdateRequest 사용.
user/dto/responses.py 역할: 응답 데이터를 정의하는 DTO. 데이터베이스 모델(User)을 응답 모델(MyProfileResponse)로 변환. 호출 관계: views.py: MyProfileResponse.from_user 호출.
common/utils.py 역할: 공통 유틸리티 함수 제공. 예: 전화번호 유효성 검증. 호출 관계: requests.py: validate_phone_number를 통해 전화번호 검증 수행.
common/errors.py 역할: 사용자 정의 예외 클래스 정의. 예: InvalidFieldFormatError, UserUnsignedError. 호출 관계: 유효성 검증 로직 및 데이터 계층에서 예외 발생 시 사용.

Example

회원가입 흐름 (POST /signup)
1. 클라이언트 → /api/users/signup 요청 전송.
2. views.py:
  - 요청 데이터를 UserSignupRequest로 검증.
  - UserService.add_user 호출.
3. service.py:
  - UserStore.add_user 호출.
4. store.py:
  - 데이터베이스에 사용자 정보 추가.
프로필 조회 흐름 (GET /me)
1. 클라이언트 → /api/users/me 요청 전송.
2. views.py:
  - 헤더 기반 인증(login_with_header) 수행.
  - 인증된 사용자 객체를 받아 MyProfileResponse.from_user로 변환 후 반환.

Conclusion

main.py → api.py → views.py → service.py → store.py → connection.py → database/settings.py

Share on

Twitter Facebook LinkedIn

Hoseok LEE

[WAFFLE] 05-5 Assignment 2 Review

Code explanation

/

wapang

wapang/app/user

wapang/app/user/dto

wapang/common

wapang/database

Project Flow

Overall Flow

import relationship

Example

Conclusion

Share on

Leave a Comment

You May Also Enjoy

Future Blog Post

Blog Post number 100

Blog Post number 4

Blog Post number 3