예비군 동미참 다녀오느라 업로드가 조금 느렸습니다.. PyTest 라이브러리를 활용해 이 때까지 개발했던 서비스 코드들에 대한 테스트 코드를 작성해보겠습니다.

서비스 코드에 몇 가지 변경사항이 있으니, 그대로 따라해보실 분들께서는 레포지토리의 test-puzzle_auth 브랜치를 참고해주시길 바랍니다.

1. Pytest 세팅

pip install pytest-asyncio

pytest-asyncio 라이브러리를 설치하면 비동기로 테스트를 진행할 수 있습니다.

테스트 폴더 구조

ROOT
ㄴ test/
  ㄴ __init__.py
  ㄴ conftest.py
    ㄴ domain1/
      ㄴ conftest.py
      ㄴ ...
    ㄴ domain2/
      ㄴ conftest.py
      ㄴ ...

    ...

위와 같은 형식으로 테스트 코드를 추가해나가겠습니다.

Test 환경 설정

테스트 할 때 실사용중인 DB에서 진행하면 안 되겠죠? 그리고 각 테스트 케이스에 독립성을 부여하기 위해 Test DB를 따로 사용하겠습니다.

# app/config.py

# Test Settings, Settings Class에 추가
TESTING = True if os.getenv("TESTING") == "true" else False
TEST_DB_NAME = os.getenv("TEST_DB_NAME")

# app/database.py

engine = create_engine(
    "mysql+pymysql://{username}:{password}@{host}:{port}/{name}".format(
        username=settings.DB_USERNAME,
        password=settings.DB_PASSWORD,
        host=settings.DB_HOST,
        port=settings.DB_PORT,
        name=settings.TEST_DB_NAME if settings.TESTING else settings.DB_NAME,
    )
)

테스트 중인지 아닌지를 서버세팅에 추가하고, 그 값에 따라 어떤 DB를 사용할 지 나누어주었습니다.

# test/conftest.py

@pytest_asyncio.fixture(scope="session")
def app():
    if not settings.TESTING:
        raise SystemError("Testing Environment setting must be 'TRUE'")

    return main.app


@pytest_asyncio.fixture
async def session():
    db = next(get_db())
    try:
        yield db
    finally:
        db.close()


@pytest_asyncio.fixture
async def client(app):
    async with AsyncClient(app=app, base_url="http://test/api/v1") as ac:
        models.Base.metadata.drop_all(bind=engine, tables=tables_to_drop)
        models.Base.metadata.create_all(bind=engine)
        yield ac

1. 테스트 설정이 아니라면 pytest를 실행시켜도 SystemError가 발생
2. session fixture로 test db에 접근
3. client fixture로 각 테스트케이스마다 DB를 초기화 시켜 독립성 유지

app fixture의 scope가 "session"으로 pytest를 실행할 때 최초 호출 후 캐시에서 불러오기 때문에 한 번만 실행된다는 점이 눈여겨볼만한 요소입니다.

2. Puzzle Test Case 작성

퍼즐 생성 테스트

퍼즐 생성 엔드포인트를 테스트하기 위해서는 WordInfo(단어정보) 테이블에 데이터가 들어가 있어야합니다.

하지만 우리가 위에서 작성한 conftest의 client fixture로 테스트를 진행하면 늘 DB가 초기화 된 상태로 진행되기 때문에 데이터가 들어있지 않습니다.

가장 간단해보이는 방법으로는 wordinfo fixture를 만들어 Test DB에 단어 데이터를 넣어주는 수가 있습니다만, 이 방법을 사용할 시에는 삽입해주는 데이터 양에 따라서 테스트 케이스 하나에 많은 비용이 들어갈 것 입니다.

단어 데이터를 모킹해주는 방법은 어떨까요? 테이블의 구조와 로직을 생각해보면 퍼즐 생성 후 저장하는 과정에서 WordInfo.id가 외래키로 걸려있으므로 테이블에 데이터를 삽입하는 과정이 불가피합니다.

그래서 어떻게 시간을 줄일거냐?

나름대로의 꼼수를 부려봤습니다. 현재 로직을 생각해보면 WordInfo 테이블에 단어 데이터를 정제하면서 삽입한 이후에 데이터의 수정을 요구하지 않습니다.

따라서, WordInfo 테이블에 데이터 삽입을 한번만 실행하고 client에서 WordInfo의 데이터를 드랍시키지 않는 방법을 떠올렸습니다.

Pytest를 시작했을 때 TEST DB에 WordInfo 데이터 유무에 따라 데이터 삽입을 결정하도록 하는 로직을 그림으로 표현해봤습니다. 아래 코드는 위 그림을 구현한 것입니다.

pip install pandas

단어 데이터를 csv로 관리하고 DB에 추가하기 위해 pandas 라이브러리를 설치해줬습니다.

test/conftest.py

@pytest_asyncio.fixture(scope="session")
async def inserted_wordinfo_into_db(app):
    with engine.connect() as connection:
        result = connection.execute(text("SELECT COUNT(*) FROM wordinfo"))
        row_count = result.scalar()

        if row_count > 494047:
            connection.execute(text("DELETE FROM wordinfo"))
            connection.execute(text("ALTER TABLE wordinfo AUTO_INCREMENT = 1"))
        if row_count == 0:
            df = pd.read_csv("test/example/wordinfo_backup.csv")
            df.to_sql("wordinfo", con=engine, index=False, if_exists="append")


@pytest_asyncio.fixture
async def client(app, inserted_wordinfo_into_db):
    async with AsyncClient(app=app, base_url="http://test/api/v1") as ac:
        tables_to_drop = [
            table for table in models.Base.metadata.sorted_tables if table.name != "wordinfo"
        ]

        models.Base.metadata.drop_all(bind=engine, tables=tables_to_drop)
        models.Base.metadata.create_all(bind=engine)
        yield ac

단어 데이터 확인 fixture의 scope를 session으로 설정해 이 후 다른 테스트 케이스에서 client를 불러도 데이터 확인 로직이 한 번만 실행하도록 구성했습니다.

또한, client fixture에서 drop하는 테이블에서 WordInfo 테이블만 제외시켜 데이터를 보존하는 동시에 각 테스트 케이스의 독립성을 유지해줬습니다.

이제 테스트 케이스를 작성해봅시다.

# test/puzzle/test_puzzle.py

@pytest.mark.asyncio
async def test_create_puzzle(client):
    r = await client.get("/puzzle")
    data = r.json()

    assert r.status_code == 200
    assert type(data.get("map", None)) == list


@pytest.mark.asyncio
async def test_create_puzzle_error_with_size(client):
    r = await client.get("/puzzle?size=11")

    assert r.status_code == 422

터미널창에서 pytest -v를 입력해 확인해본 결과 잘 나왔습니다.

퍼즐 조회 테스트

퍼즐 조회 테스트를 하기 위해서는 퍼즐 데이터가 존재해야겠죠? puzzle fixture를 작성해줍시다.

# test/puzzle/conftest.py

@pytest_asyncio.fixture
async def puzzle(session):
    with open("test/example/example_map.json", "r", encoding="utf-8") as f:
        json_data = json.load(f)
    _map = json_data.get("map", None)
    _answers = json_data.get("desc", None)

    map_row = models.Puzzle(puzzle=_map)
    session.add(map_row)
    session.flush()

    insert_data = [
        {"puzzle_id": map_row.id, "word_id": desc["id"], "num": desc["num"]} for desc in _answers
    ]

    session.bulk_insert_mappings(models.PuzzleAnswer, insert_data)
    session.commit()

퍼즐 샘플이 담긴 json 파일을 불러와 TEST DB에 저장하는 fixture입니다.

@pytest.mark.asyncio
async def test_get_puzzle(client, puzzle):
    r = await client.get("/puzzle/1")

    assert r.status_code == 200


@pytest.mark.asyncio
async def test_get_puzzle_error_with_wrong_id(client):
    r = await client.get("/puzzle/2")

    assert r.status_code == 404

3. Auth Test Case 작성

일반 로그인

# test/auth/conftest.py

@pytest_asyncio.fixture
async def user(session):
    user_dict = {
        "email": "test@test.com",
        "nickname": "testname",
    }
    raw_password = "test1234"
    salt_value = bcrypt.gensalt()
    password = bcrypt.hashpw(raw_password.encode(), salt_value)

    user_dict.update({"password": password})

    user_row = models.User(**user_dict)
    session.add(user_row)
    session.commit()

    return user_row

conftest에 user fixture를 만들어줍시다.

# test/auth/test_auth.py

@pytest.mark.asyncio
async def test_general_register(client, session):
    body = {
        "email": "test@test.com",
        "password": "test1234",
        "nickname": "testname",
    }
    r = await client.post("/auth/general-register", data=json.dumps(body))

    assert r.status_code == 200

    user = session.query(models.User).filter(models.User.email == body.get("email")).first()

    assert body.get("email") == user.email
    assert body.get("nickname") == user.nickname

@pytest.mark.asyncio
async def test_general_login(client, user):
    body = {"email": "test@test.com", "password": "test1234"}
    r = await client.post("/auth/general-login", data=json.dumps(body))

    assert r.status_code == 200

@pytest.mark.asyncio
async def test_check_duplicated_email(client, user):
    email = "test@test.com"
    r = await client.get(f"/auth/duplicated?email={email}")
    data = r.json()

    assert r.status_code == 200
    assert data.get("is_duplicated")

일반 회원가입과 로그인 테스트 및 중복 이메일 테스트 코드입니다. 이제 응답에 딸려온 토큰이 잘 생성되었는지 테스트 해봅시다.

# test/auth/conftest.py

@pytest_asyncio.fixture
async def token(user):
    jwt_service = JWTService()
    user_dict = user.as_dict()
    user_dict.pop("password")
    user_dict.pop("created_at")
    user_dict.pop("updated_at")
    access_token = jwt_service.create_access_token(user_dict)

    return access_token

user fixture를 주입받아 token을 생성하는 fixture입니다.

# test/auth/test_auth.py

@pytest.mark.asyncio
async def test_get_user_by_token(client, token):
    cookies = {"access": token}
    r = await client.get("/auth/get-user", cookies=cookies)
    data = r.json()

    assert r.status_code == 200
    assert data.get("email", None) == "test@test.com"

토큰 정보로 유저 정보를 반환하는 테스트 코드입니다.

소셜 로그인

소셜 로그인은 일반 로그인과 다르게 외부 API를 이용하는 로직이 있습니다. 이는 모킹을 이용해 테스트해보겠습니다.

파이썬에서 기본으로 제공하는 unittest 라이브러리의 patch 기능을 사용할 수 있습니다.

테스트 할 엔드포인트의 함수를 모킹하여 원하는 동작을 하도록 바꿔주는 기능이 patch입니다.

# test/auth/mock.py

async def mock_post_token_request(*args, **kwargs):
    return {"access_token": "mock_access_token"}

async def mock_get_userinfo(*args, **kwargs):
    return {"email": "test@test.com", "name": "testuser"}

위 함수는 구글 API에서 Access Token을 받아오는 메소드, 아래 함수는 구글 API에서 유저정보를 받아오는 메소드를 모킹할 함수입니다.

@patch("app.api.v1.auth.service.GoogleOAuthService.get_token", new=mock.mock_post_token_request)
@patch("app.api.v1.auth.service.GoogleOAuthService.get_userinfo", new=mock.mock_get_userinfo)

테스트 코드에 위와 같은 데코레이터를 붙여주면 모킹 기능이 작동합니다.

# test/auth/conftest.py

@pytest_asyncio.fixture
async def oauth_google_user(session):
    user = models.User(email="test@test.com", nickname="test")
    session.add(user)
    session.flush()

    oauth_info = models.OAuth(user_id=user.id, email=user.email, provider="google")
    session.add(oauth_info)
    session.commit()

소셜 로그인 유저 fixture를 작성한 뒤 테스트 코드도 작성해봅시다.

# test/auth/test_auth.py

@pytest.mark.asyncio
@patch("app.api.v1.auth.service.GoogleOAuthService.get_token", new=mock.mock_post_token_request)
@patch("app.api.v1.auth.service.GoogleOAuthService.get_userinfo", new=mock.mock_get_userinfo)
async def test_google_register(client, session):
    code = "dummy code"
    r = await client.get(f"/auth/oauth-register/google/callback?code={code}")

    assert r.status_code == 200

    user = session.query(models.User).filter(models.User.email == "test@test.com").first()

    assert "test" == user.nickname

    oauth_user = session.query(models.OAuth).filter(models.OAuth.id == user.id).first()

    assert oauth_user.email == user.email


@pytest.mark.asyncio
@patch("app.api.v1.auth.service.GoogleOAuthService.get_token", new=mock.mock_post_token_request)
@patch("app.api.v1.auth.service.GoogleOAuthService.get_userinfo", new=mock.mock_get_userinfo)
async def test_google_register_failed_by_duplicated_email(client, user):
    code = "dummmy code"
    r = await client.get(f"/auth/oauth-register/google/callback?code={code}")
    data = r.json()

    assert r.status_code == 400
    assert data.get("detail") == "중복된 이메일입니다."


@pytest.mark.asyncio
@patch("app.api.v1.auth.service.GoogleOAuthService.get_token", new=mock.mock_post_token_request)
@patch("app.api.v1.auth.service.GoogleOAuthService.get_userinfo", new=mock.mock_get_userinfo)
async def test_google_login(client, oauth_google_user):
    code = "dummmy code"
    r = await client.get(f"/auth/oauth-register/google/callback?code={code}")
    data = r.json()

    assert r.status_code == 200
    assert data.get("email") == "test@test.com"
    assert data.get("nickname") == "test"

테스트가 잘 되는 군요.

이번 포스트에서는 FastAPI 서버에 회원가입/로그인과 소셜 로그인을 구현하고 JWT로 인증하는 기능을 작성해보도록 하겠습니다.

JWT와 쿠키를 사용하는 이유

참고 링크 잘 정리되어있는 블로그 글이 있어서 링크를 걸어두도록 하겠습니다.

프로젝트 구조 변경

역할 별로 폴더를 나누던 구조에서 기능 별로 폴더를 나누는 구조로 변경했습니다.

변경 전

ROOT
ㄴ app/
  ㄴ utils/
   ㄴ puzzle.py
   ㄴ ...
 ㄴ api/
  ㄴ v1/
   ㄴ puzzle.py
   ㄴ ...
 ㄴ schema/

...

변경 후

ROOT
ㄴ app/
  ㄴ api/
   ㄴ v1/
    ㄴ puzzle/
     ㄴ controller.py
     ㄴ schema.py
     ㄴ ...
    ㄴ auth/
     ㄴ controller.py
     ㄴ service.py
     ㄴ ...

...

1. 회원가입/로그인 구현

# jwt.py
# models.py
class BaseDate(BaseMin):
    created_at = Column(DATETIME, default=func.now())
    updated_at = Column(DATETIME, default=func.now(), onupdate=func.now())

class User(BaseDate, Base):
    __tablename__ = "user"

    email = Column(VARCHAR(20), unique=True, nullable=False)
    password = Column(VARCHAR(70), nullable=True)
    nickname = Column(VARCHAR(10), nullable=False)
    solved = Column(Integer, default=0)

    def as_dict(self):
        return {column.name: getattr(self, column.name) for column in self.__table__.columns}

User 테이블을 추가해주겠습니다.

JWT 암호화 및 해독을 담당할 관련 서비스와 쿠키에 토큰을 달아줄 서비스를 작성하겠습니다.

class JWTService:
    """
    JWT 암호화 및 해독에 필요한 기능을 제공하는 클래스
    """

    def __init__(self):
        self.algorithm = settings.ALGORITHM
        self.secret_key = settings.SECRET_KEY
        self.access_expire_time = settings.ACCESS_EXPIRE_TIME
        self.refresh_expire_time = settings.REFRESH_EXPIRE_TIME

    def _encode(self, data: dict, expires_delta: int) -> str:
        """
        JWT 토큰으로 암호화한다.

        Args:
            data (dict): 암호화 할 데이터
            expires_delta (int): JWT 만료기간
        Returns:
            str: JWT 토큰
        """
        to_encode = data.copy()
        expire = datetime.now(ZoneInfo("Asia/Seoul")) + timedelta(minutes=expires_delta)
        to_encode.update({"exp": expire})
        return jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm)

    def _decode(self, token: str) -> Optional[Dict]:
        """
        JWT 토큰을 해독한다.

        Args:
            token (str): JWT 토큰
        Returns:
            Dict or None: 해독 성공 시 정보를 담은 사전, 실패 시 None
        """
        try:
            return jwt.decode(token, self.secret_key, algorithms=self.algorithm)
        except JWTError:
            return None

    def _create_token(self, data: dict, expires_delta: int) -> str:
        """
        JWT 토큰 암호화 래퍼함수
        """
        return self._encode(data, expires_delta)

    def create_access_token(self, data) -> str:
        """
        Access Token 생성 함수
        """
        return self._create_token(data, self.access_expire_time)

    def create_refresh_token(self, data) -> str:
        """
        Refresh Token 생성 함수
        """
        return self._create_token(data, self.refresh_expire_time)

    def check_is_expired(self, token: str) -> Optional[Dict]:
        """
        JWT 토큰 해독 래퍼함수
        """
        payload = self._decode(token)

        now = datetime.timestamp(datetime.now(ZoneInfo("Asia/Seoul")))
        if payload and payload["exp"] < now:
            return None

        return payload

# cookie.py
class Cookie:
    """
    응답 객체의 쿠키 관련 작업을 담당하는 클래스
    """

    async def attach_token_into_cookie(
        self, response: JSONResponse, access_token: str, refresh_token: str
    ) -> JSONResponse:
        """
        쿠키에 Access, Refresh Token을 부착하는 메소드

        Args:
            response (JSONResponse): 쿠키 생성해줄 response 객체
            access_token (str): Access Token JWT
            refresh_token (str): Refresh Token JWT
        Returns:
            JSONResponse: 쿠키가 부착된 JSONResponse 객체
        """
        response.set_cookie(
            key="access",
            domain="localhost",
            samesite="None",
            value=access_token,
        )
        response.set_cookie(
            key="refresh",
            domain="localhost",
            samesite="None",
            value=refresh_token,
        )
        return response

config.py와 .env 파일에 필요한 설정들을 추가되어 있는게 전제입니다. 위에 작성된 두 클래스를 Auth 서비스에 주입해 내부 메소드처럼 사용할 예정입니다.

# auth.py
class AuthBase(ABC):
    """
    AuthService의 기초가 되는 추상클래스
    """

    def __init__(self):
        self.jwt_service = JWTService()
        self.cookie_service = Cookie()

    @abstractmethod
    async def register(self):
        """
        회원가입을 담당하는 추상 메소드
        """
        pass

    @abstractmethod
    async def login(self):
        """
        로그인을 담당하는 추상 메소드
        """
        pass

    async def _get_login_response(self, user) -> JSONResponse:
        """
        로그인 성공 후 응답을 생성하는 메소드
        Args:
            user (User): 로그인한 유저의 User 객체
        Returns:
            JSONResponse: Access, Refresh Token이 쿠키에 들어간 응답
        """
        self.user_dict = user.as_dict()
        self.user_dict.pop("password")
        self.user_dict.pop("created_at")
        self.user_dict.pop("updated_at")

        response = JSONResponse(content=self.user_dict)
        return await self._attach_token(response, self.user_dict)

    async def _attach_token(self, response, data) -> JSONResponse:
        """
        응답 쿠키에 Access, Refresh Token을 부착하는 메소드
        Args:
            response (JSONResponse): content로 유저 정보가 들어간 응답 객체
            data: Access, Refresh Token
        Returns:
            JSONResponse: Access, Refresh Token이 쿠키에 들어간 응답
        """
        access_token = self.jwt_service.create_access_token(data)
        refresh_token = self.jwt_service.create_refresh_token(data)
        return await self.cookie_service.attach_token_into_cookie(
            response, access_token, refresh_token
        )

Auth 서비스의 추상 클래스를 먼저 선언해줬습니다.

JWTService, Cookie 클래스가 init 함수에서 주입되었고, 로그인 후 응답을 생성하는 메소드와 응답 쿠키에 토큰을 달아주는 메소드는 정의되어 있는 상태입니다.

# exception.py
class LoginNotValidIDPWException(HTTPException):
    def __init__(self, detail: str = "입력한 ID 또는 비밀번호가 일치하지 않습니다."):
        super().__init__(status_code=400, detail=detail)


# service.py
class GeneralAuthService(AuthBase):
    """
    아이디(이메일), 비밀번호로 회원가입/로그인, 인증 서비스 클래스
    """

    def __init__(self, email: str, password: str, db: Session, nickname: Optional[str] = None):
        """
        입력받은 이메일, 비밀번호, 닉네임을 초기화한다.
        """
        self.email = email
        self.password = password
        self.nickname = nickname
        self.db = db
        super().__init__()

    async def register(self) -> Dict:
        """
        비밀번호를 해쉬화해서 데이터베이스에 유저 정보를 저장하는 메소드

        Returns:
            dict: 필요한 유저 정보만 담아 반환되는 딕셔너리 자료
        """
        user = User(email=self.email, password=await self._hash_pw(), nickname=self.nickname)
        self.db.add(user)
        self.db.commit()

        return await self._get_login_response(user)

    async def login(self) -> JSONResponse:
        """
        이메일과 비밀번호로 유저 정보를 확인하여 로그인하는 메소드

        Returns:
            JSONResponse: 쿠키에 토큰 정보, 콘텐츠에 유저 정보를 담은 response 객체
        """
        user = self.db.query(User).filter(User.email == self.email).first()
        if (
            user
            and user.password
            and bcrypt.checkpw(self.password.encode(), user.password.encode())
        ):
            return await self._get_login_response(user)
        raise LoginNotValidIDPWException()

    async def _hash_pw(self) -> str:
        """
        비밀번호를 해쉬화 해 반환한다.

        Returns:
            str: 해쉬화 된 비밀번호
        """
        salt_value = bcrypt.gensalt()
        return bcrypt.hashpw(self.password.encode(), salt_value)

저번에 작성했던 PuzzleCreateService와 init 함수 부분이 미묘하게 다릅니다. 퍼즐 서비스에서는 DB 세션을 바로 DI 받았었는데, 여기서는 안 받고있네요.

퍼즐 서비스에서는 SIZE 파라미터를 쿼리 형태로 받아왔습니다. 하지만, 회원가입/로그인할 때 사용할 데이터를 쿼리로 받아올 수는 없겠죠.

저는 JSON Body 형태로 필요 데이터를 받아오기 위해서 의존성 주입 함수를 따로 작성해줬습니다.

# schema.py
class GeneralLoginModel(BaseModel):
    email: str
    password: str


class GeneralRegisterModel(GeneralLoginModel):
    nickname: str

# dependancy.py
def get_general_auth_service_register(
    auth_data: GeneralRegisterModel, db: Session = Depends(get_db)
) -> GeneralAuthService:
    return GeneralAuthService(
        email=auth_data.email, password=auth_data.password, nickname=auth_data.nickname, db=db
    )


def get_general_auth_service_login(
    auth_data: GeneralLoginModel, db: Session = Depends(get_db)
) -> GeneralAuthService:
    return GeneralAuthService(email=auth_data.email, password=auth_data.password, db=db)

이 함수들을 엔드포인트에서 DI 해주면, JSON Body으로 데이터를 받아오는 동시에 DB 세션을 서비스에 주입해줄 수 있습니다.

# controller.py

@router.post(
    "/general-register", status_code=status.HTTP_201_CREATED, description="EMAIL/PW로 가입"
)
async def general_register(
    auth_service: GeneralAuthService = Depends(get_general_auth_service_register),
):
    return await auth_service.register()


@router.post("/general-login", status_code=status.HTTP_200_OK, description="EMAIL/PW로 로그인")
async def general_login(auth_service: GeneralAuthService = Depends(get_general_auth_service_login)):
    return await auth_service.login()

TEST

회원가입/로그인 후에 JWT가 생성되어 쿠키에 담겨 반환되었는지 알아보기 위해 POSTMAN을 사용해서 테스트 해보도록 하겠습니다.

쿠키 access, refresh에 데이터가 잘 들어와있는 모습을 볼 수 있습니다.

2. 소셜 로그인

소셜 로그인 기능 구현에 앞서 소셜 로그인의 큰 틀을 먼저 살펴보겠습니다.

구글 소셜 로그인에 대한 예시를 그림으로 그려보겠습니다.

General Auth와 비교하면 클라이언트에게서 인가 코드를 전달받고 구글에 access token과 유저 정보를 요청하는 로직이 추가되었다고 볼 수 있겠습니다.

# auth.py
class OAuthBase(AuthBase):
    """
    GeneralAuth 서비스를 상속받아 OAuth 서비스의 기초가 되는 추상클래스
    """

    def __init__(self):
        super().__init__()

    @abstractmethod
    async def get_token(self):
        """
        OAuth 유저 정보 획득에 필요한 Token을 획득하는 추상 메소드
        """
        pass

    @abstractmethod
    async def get_userinfo(self):
        """
        Token으로 OAuth 유저 정보를 획득하는 추상 메소드
        """
        pass

기능 확장에 용이함을 주기 위해서 소셜 로그인에서 반드시 필요한 과정인 get_token과 get_userinfo 메소드를 추상메소드로 생성하는 OAuthBase 추상클래스를 작성해줍니다.

# exception.py

class EmailDuplicatedException(HTTPException):
    def __init__(self, detail: str = "중복된 이메일입니다."):
        super().__init__(status_code=400, detail=detail)


class GoogleGetTokenException(HTTPException):
    def __init__(self, detail: str = "구글 토큰 획득 과정에서 오류가 발생했습니다."):
        super().__init__(status_code=400, detail=detail)


class GoogleGetUserInfoException(HTTPException):
    def __init__(self, detail: str = "구글 유저 정보 과정에서 오류가 발생했습니다."):
        super().__init__(status_code=400, detail=detail)


class GoogleRegisterException(HTTPException):
    def __init__(self, detail: str = "구글 소셜 회원가입 과정에서 오류가 발생했습니다."):
        super().__init__(status_code=400, detail=detail)

# service.py

class GoogleOAuthService(OAuthBase):
    def __init__(self, code: str, db: Session = Depends(get_db)):
        """
        구글 소셜 로그인에 필요한 정보를 초기화 한다.
        """
        super().__init__()
        self.provider = "google"
        self.client_id = settings.GOOGLE_CLIENT_ID
        self.redirect_uri = settings.GOOGLE_REDIRECT_URI
        self.client_secret = settings.GOOGLE_CLIENT_SECRET
        self.db = db
        self.code = code
        self.token_request_url = "https://oauth2.googleapis.com/token"
        self.userinfo_endpoint = "https://www.googleapis.com/userinfo/v2/me"

    async def register(self) -> JSONResponse:
        try:
            email = self.oauth_user_info.get("email", None)
            nickname = email.split("@")[0]
            user = User(email=email, nickname=nickname)
            self.db.add(user)
            self.db.flush()

            oauth = OAuth(user_id=user.id, email=user.email, provider=self.provider)
            self.db.add(oauth)

            self.db.commit()
        except Exception:
            raise GoogleRegisterException()

        return await self._get_login_response(user)

    async def login(self) -> JSONResponse:
        """
        소셜 로그인 로직을 수행하는 기초 메소드,
        데이터베이스에 유저 존재 -> 정보 반환
        존재 X -> 저장(회원가입) 후 정보 반환

        Returns:
            JSONResponse: 쿠키에 토큰 정보, 콘텐츠에 유저 정보를 담은 response 객체
        """
        user = await self.is_registered()

        if user is None:
            return await self.register()

        return await self._get_login_response(user)

    async def get_token(self) -> str:
        """
        유저 정보 조회에 필요한 토큰을 받아온다.

        Returns:
            str: 구글 OAuth 서비스에서 받은 토큰
        """
        token_request_payload = {
            "grant_type": "authorization_code",
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "code": self.code,
            "client_secret": self.client_secret,
        }

        async with httpx.AsyncClient() as client:
            response = await client.post(self.token_request_url, data=token_request_payload)
        result = response.json()

        if "access_token" in result:
            return result["access_token"]
        else:
            raise GoogleGetTokenException()

    async def get_userinfo(self, token) -> Dict:
        """
        구글에서 받아온 토큰으로 유저 정보를 요청하고 반환한다.

        Args:
            token (str): 구글에서 받아온 토큰
        Returns:
            dict:
        """
        headers = {"Authorization": f"Bearer {token}"}

        async with httpx.AsyncClient() as client:
            response = await client.get(self.userinfo_endpoint, headers=headers)
        if response.status_code == 200:
            return response.json()
        else:
            raise GoogleGetUserInfoException()

    async def is_registered(self) -> Optional[User]:
        """
        가입된 유저라면 유저 정보를, 아니라면 None을 반환한다.

        Returns:
            User or None: 정보가 있다면 User 객체, 아니면 None 반환
        """
        token = await self.get_token()
        user_info = await self.get_userinfo(token)
        return await self.get_user_from_db(user_info)

    async def get_user_from_db(self, user_info):
        """
        구글에서 받은 유저 정보를 데이터베이스에 검색하고
        존재하면 User 객체, 아니면 None을 반환한다.

        Args:
            user_info (dict): 구글에서 받은 유저 정보
        Returns:
            User or None: 정보가 있다면 User 객체, 아니면 None 반환
        """

        user = self.db.query(User).filter(User.email == user_info.get("email")).first()
        if user:
            oauth_user_info = (
                self.db.query(OAuth)
                .filter(OAuth.provider == self.provider, OAuth.email == user.email)
                .first()
            )
            if oauth_user_info:
                return user
            else:
                raise EmailDuplicatedException()
        else:
            self.oauth_user_info = user_info
            return None

그림의 로직대로 진행하되 code를 쿼리 파라미터로 입력받고, DB 내 유저 정보의 유무에 따라 조회만 할지 저장 할지 결정합니다.

# controller.py

@router.get(
    "/oauth-register/google/callback",
    status_code=status.HTTP_201_CREATED,
    description="구글 소셜 로그인",
)
async def kakao_callback(auth_service: GoogleOAuthService = Depends(GoogleOAuthService)):
    return await auth_service.login()

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>Login Page</title>
    <style>
      body {
        font-family: Arial, sans-serif;
        background-color: #f5f5f5;
        display: flex;
        align-items: center;
        justify-content: center;
        height: 100vh;
        margin: 0;
      }

      .login-container {
        background-color: #fff;
        padding: 20px 30px;
        border-radius: 8px;
        box-shadow: 0 0 10px rgba(0, 0, 0, 0.1);
        text-align: center;
      }

      .google-login-btn img {
        width: 200px;
        border-radius: 3px;
        transition: opacity 0.3s;
      }

      .google-login-btn:hover img {
        opacity: 0.8;
      }
    </style>
  </head>

  <body>
    <div class="login-container">
      <a
        href="https://accounts.google.com/o/oauth2/v2/auth?client_id=639700898145-445d540qksvfnm3tg29mkht55ufkfeuv.apps.googleusercontent.com&redirect_uri=http://localhost:8000/api/v1/auth/oauth-register/google/callback&response_type=code&scope=email profile"
        class="github-login-btn"
      >
        <img
          src="https://encrypted-tbn0.gstatic.com/images?q=tbn:ANd9GcReH1nivRV_9yG4wz04xIz1EEh-J69U_2JRaA&s"
          alt="Social Login"
        />
      </a>
    </div>
  </body>
</html>

엔드포인트와 인가 코드를 받아올 간단한 html을 작성한 뒤 테스트 해보겠습니다.

작성한 html 파일을 오픈해서 구글 로고를 눌러줍시다.

저는 이미 로그인 해놨었기 때문에, 데이터를 GeneralAuth와 동일한 형태로 받아오는 모습입니다. 로그인이 되어 있지 않다면 구글 계정을 선택하는 화면으로 넘어갈 것 입니다.

지난 포스트에서는 십자말풀이 게임판을 만드는 작업까지 진행했습니다. 이번 포스트에서는 생성된 십자말풀이 게임판을 DB에 삽입하고, 이를 검색해보는 작업까지 진행해보도록 하겠습니다.

DB에 게임판 삽입

퍼즐과 정답을 저장할 테이블 구조를 먼저 정의해보겠습니다.

class Puzzle(BaseMin, Base):
    __tablename__ = "puzzle"

    puzzle = Column(JSON, nullable=False)
    name = Column(VARCHAR(50), nullable=False, default=lambda: str(uuid.uuid4()))


class PuzzleAnswer(BaseMin, Base):
    __tablename__ = "puzzleanswer"

    puzzle_id = Column(Integer, ForeignKey("puzzle.id"), nullable=False)
    word_id = Column(Integer, ForeignKey("wordinfo.id"), nullable=False)
    num = Column(Integer, nullable=False)

Puzzle

2차원 리스트로 만들어진 퍼즐은 JSON 형태로 테이블에 넣도록 하겠습니다. name으로 클라이언트가 식별할 수 있는 컬럼을 하나 추가해줬습니다. <

Answer

퍼즐에 할당된 정답은 puzzle_id와 word_id를 외래키로 잡아 어떤 게임판의 정답인지, 정답이 어떤 단어인지를 알 수 있게 만들었습니다. 함께 저장되는 num으로 최초에 게임판 생성할 때와 동일한 데이터를 만들 수 있도록 했습니다.

word_id도 외래키로 저장되어있어 조회 시에 word 테이블과의 조인 작업이 불가피합니다. 추후에 데이터가 많이 쌓인다면 반정규화도 고려해볼만한 사항입니다.

DB에 게임판을 삽입하는 로직은 생성 이후에 곧바로 이루어져야 하는 흐름입니다. 따라서, 게임판 생성 서비스에 함수를 추가하겠습니다.

async def insert_map_answer_into_db(self) -> None:
        """
        생성된 맵과 정답을 DB에 삽입한다.
        """
        map_row = Puzzle(puzzle=self.map)
        self.db.add(map_row)
        self.db.flush()

        insert_data = [
            {"puzzle_id": map_row.id, "word_id": desc["id"], "num": desc["num"]}
            for desc in self.desc
        ]

        self.db.bulk_insert_mappings(PuzzleAnswer, insert_data)
        self.db.commit()

puzzle을 먼저 flush해서 puzzle_id를 추출합니다. 이후에 puzzle_id와 self.desc 속의 num, desc로 여러 row를 한 번에 bulk_insert 합니다.

DB에서 게임판 조회

조회 서비스 구현

게임판 조회 서비스 클래스를 생성하고 조회하는 함수를 추가해보겠습니다.

class PuzzleReadService:
    def __init__(self, puzzle_id: int, db: Session = Depends(get_db)):
        self.db = db
        self.puzzle_id = puzzle_id

    async def read_puzzle_from_db_by_id(self) -> Dict:
        """
        데이터베이스에서 퍼즐 ID로 퍼즐과 정답 정보를 읽어와 반환한다.
        Args:
            puzzle_id (int): 반환할 퍼즐 ID
        Returns:
            Dict: 퍼즐, 정답 정보가 담긴 사전 데이터
        """
        puzzle = self.db.query(Puzzle).filter(Puzzle.id == self.puzzle_id).first()
        if puzzle is None:
            raise PuzzleNotExistException()
        answer = (
            self.db.query(PuzzleAnswer.num, WordInfo.pos, WordInfo.desc, WordInfo.word)
            .filter(PuzzleAnswer.puzzle_id == self.puzzle_id)
            .join(WordInfo, PuzzleAnswer.word_id == WordInfo.id)
            .all()
        )

        answer_json = [
            {"num": num, "desc": {"pos": pos, "desc": desc, "word": word}}
            for num, pos, desc, word in answer
        ]

        return {"map": puzzle.puzzle, "desc": answer_json}

1. puzzle_id을 클라이언트에서 입력받았을 때, 퍼즐만 간단하게 불러옵니다. 예외) puzzle_id인 puzzle이 존재하지 않을 때.
2. PuzzleAnswer.puzzle_id가 puzzle_id이면서, PuzzleAnswer.word_id가 WordInfo.id인 PuzzleAnswer 객체들을 모두 불러옵니다.

예외 처리

from fastapi import HTTPException


class PuzzleNotExistException(HTTPException):
    def __init__(self, detail: str = "존재하지 않는 퍼즐입니다."):
        super().__init__(status_code=404, detail=detail)

예외 처리를 담당하는 모듈을 따로 만들어 추가해주었습니다.

다르게 PuzzleAnswer를 찾아오는 방법

SQLAlchemy ORM에서 제공하는 RelationShip을 사용해서 가져올 수 있습니다.

# import relationship on models.py
from sqlalchemy.orm import relationship


class WordInfo(BaseMin, Base):
    __tablename__ = "wordinfo"

    word = Column(VARCHAR(5), nullable=False, index=True)
    desc = Column(VARCHAR(200), nullable=False)
    len = Column(Integer, nullable=False)
    pos = Column(VARCHAR(7), nullable=False)

    puzzle_answers = relationship("PuzzleAnswer", back_populates="word")


class PuzzleAnswer(BaseMin, Base):
    __tablename__ = "puzzleanswer"

    puzzle_id = Column(Integer, ForeignKey("puzzle.id"), nullable=False)
    word_id = Column(Integer, ForeignKey("wordinfo.id"), nullable=False)
    num = Column(Integer, nullable=False)

    word = relationship("WordInfo", back_populates="puzzle_answers")

models.py에서 relationship을 추가해주고,

from sqlalchemy.orm import joinedload

answer = (
    self.db.query(PuzzleAnswer)
    .option(joinedload(PuzzleAnswer.wordinfo))
    .filter(PuzzleAnswer.puzzle_id == self.puzzle_id)
    .all()
)

answer_josn = [
    {
        "num": answer.num,
        "pos": answer.wordinfo.pos,
        "desc": answer.wordinfo.desc,
        "word": answer.wordinfo.word,
    }
    for answer in puzzle_answers
]

위에서 작성한 쿼리문과 반복문을 수정해주었습니다. joinedload는 eager loading 하도록 해주는 옵션으로 이를 사용해서 불러오지않으면, answer.wordinfo.* 할 때마다 새로운 쿼리를 날리게되므로 N+1문제가 발생할 수 있습니다.

조회 엔드포인트 작성

@router.get("/{puzzle_id}")
async def read_puzzle(
    puzzle_id: int, puzzle_service: PuzzleReadService = Depends(PuzzleReadService)
):
    return await puzzle_service.read_puzzle_from_db_by_id()

요청을 보내보면, 잘 받아오는 것을 알 수 있습니다.

다음번에는 회원가입/로그인 기능에 대해서 포스트해보겠습니다. 읽어주셔서 감사합니다.

지난 번 포스트에서 단어 데이터 추출 후 데이터베이스 삽입까지 했습니다.

이번 포스트에서는 단어 데이터들을 데이터베이스에서 선택해 십자말풀이 퍼즐을 만들어보도록 하겠습니다.

모델 변경점

class WordInfo(BaseMin, Base):
    __tablename__ = "wordinfo"

    word = Column(VARCHAR(5), nullable=False)
    desc = Column(VARCHAR(200), nullable=False)
    len = Column(Integer, nullable=False)
    pos = Column(VARCHAR(7), nullable=False)

시작하기에 앞서 지난 번 모델에서 단어 의 최대 길이가 5로 감소했습니다.

퍼즐 데이터 형식

1. 퍼즐의 크기는 정사각형이다
2. 단어는 위->아래, 왼->오 방향으로만 추가된다.
3. 연결된 단어를 제외하고 단어끼리는 상,하,좌,우로 접하지 않는다.

위 세가지를 기본 전제로 퍼즐을 구성하기로 했습니다.

위 그림은 (7x7) 형태의 제가 생각하는 퍼즐의 최종 모습을 메모장으로 작성해본 샘플입니다.

이러한 데이터를 구성하기 위해서 어떤 데이터 형식을 취해볼까요?

저는 퍼즐의 각 단어 배치를 2차원 배열로 처리하기로 했습니다. (7x7) 형태의 정수 0으로 이루어진 리스트를 생성하고, 단어가 추가될 때마다 숫자가 +1 되면서 마킹하기로 말이죠.

그리고 각 단어에 대해 {번호, 설명}, {번호, 단어} 자료형을 가지는 리스트로 관리해보겠습니다.

퍼즐 생성

Phase1: 첫 단어 배치

각 함수들이 의존성 주입 형태로 배치되어있기 때문에, 가장 아래 함수부터 매개변수를 거슬러올라가며 보는 게 읽기 쉬울 겁니다.

import random
from collections import deque
from fastapi import Depends
from sqlalchemy.orm.session import Session
from typing import List, Tuple, Deque, Dict
from app.models import WordInfo
from app.database import get_db


async def append_letter_into_queue(
    start_point: Tuple[int, int], length: int, word: str, queue: deque, dir: bool
) -> Deque:
    """
    시작점, 길이, 단어, 방향을 토대로 큐에 다음 단어 추가에 대한 힌트를 큐에 삽입 후 반환한다.
    Args:
        start_point (Tuple): 단어 첫 음절이 위치하는 곳
        length (int): 단어의 길이
        word (str): 단어
        queue (deque): 큐
        dir (bool): 단어 방향 가로, 세로 여부
    Returns:
        Deque: 다음 단어의 시작점, 시작 음절, 방향이 추가된 큐
    """
    def add_to_queue(coord: Tuple[int, int], letter: str, dir: bool) -> None:
        if 0 <= coord[0] < 11 and 0 <= coord[1] < 11:
            queue.append(((coord[0], coord[1]), letter, not (dir)))
    if dir:
        add_to_queue((start_point[0], start_point[1] + length - 1), word[-1], dir)
    else:
        add_to_queue((start_point[0] + length - 1, start_point[1]), word[-1], dir)
    if length > 2:
        if length >= 5:
            if dir:
                add_to_queue((start_point[0], start_point[1] + 2), word[2], dir)
            else:
                add_to_queue((start_point[0] + 2, start_point[1]), word[2], dir)
        add_to_queue(start_point, word[0], dir)
    return queue

async def create_map() -> List[str]:
    """
    0으로 초기화 된 맵을 생성해 반환한다.
    Args:
        None
    Returns:
        List[str]
    """
    map = [[0 for _ in range(11)] for _ in range(11)]
    return map

async def find_first_word_info(db: Session = Depends(get_db)) -> WordInfo:
    """
    DB에서 랜덤으로 하나의 단어를 찾아 반환한다.
    Args:
        db (Session): 커넥션
    Returns:
        WordInfo: 랜덤으로 가져온 WordInfo Row 객체
    """
    random_idx = random.randint(1, 494047)
    return db.query(WordInfo).filter(WordInfo.id == random_idx).first()

async def create_puzzle_phase1(
    word: WordInfo = Depends(find_first_word_info), map: List[str] = Depends(create_map)
) -> Dict:
    """
    0으로 초기화 된 맵에 첫 단어를 추가,
    이 후 추가될 단어와 그 위치에 대한 힌트를 큐에 삽입,
    첫 단어 설명 추가 후 반환한다.
    Args:
        word (WordInfo): 단어 정보
        map (List[str]): 0으로 초기화 된 맵
    Returns:
        Dict: {map(List[str]), queue(deque), desc(dict)}
    """
    start_y, start_x = 0, 0
    queue = deque()
    desc = []
    words = []
    for i in range(word.len):
        map[start_y][start_x + i] = 1
    words.append({"num": 1, "word": word.word})
    desc.append({"num": 1, "desc": word.desc})
    queue = await append_letter_into_queue((start_y, start_x), word.len, word.word, queue, True)
    return {"map": map, "queue": queue, "desc": desc, "words": words}

위 함수들을 간단하게 요약하자면, 10x10 퍼즐에 (0,0), 최좌상단에서 가로 방향으로 단어를 하나 배치하고, 단어의 길이에 따라 큐에 다음 단어의 시작점 좌표를 추가해줬다는 겁니다. 단어 배치 시 0으로 초기화 된 맵에 단어 위치에 해당하는 좌표를 1로 바꿨습니다.

위 상황을 그림으로 표현해봤습니다. 오른쪽 그림의 빨간 좌표가 현재 큐에 들어가있는 좌표입니다. 세로로 표기된 것은 편의상 배치한 것이고, 실제 코드 상에서는 세로는 False, 가로는 True 입니다.

Phase2: 첫 단어에서 이어지는 단어들 배치

Phase1의 첫 단어 배치가 끝나면 이제 첫 단어에서 이어지도록 단어를 계속 배치해야합니다.

위 그림을 예시로 들면 "고"와"래"로 시작하는 단어가 되겠네요.

async def find_word_info_start_with(start_word: str, db: Session) -> WordInfo:
    """
    DB에서 {start_word}로 시작하는 단어를 찾아 반환한다.
    Args:
        db (Session): 커넥션
        start_word (str): 시작 어절
    Returns:
        WordInfo: {start_word}로 시작하는 WordInfo Row 객체
    """
    return (
        db.query(WordInfo)
        .filter(WordInfo.word.like(f"{start_word}%"))
        .order_by(func.rand())
        .first()
    )

위 함수는 DB에서 {start_word}로 시작하는 단어를 검색해서 랜덤 추출합니다.

async def create_puzzle_phase2(
    db: Session = Depends(get_db), map_queue_desc: Dict = Depends(create_puzzle_phase1)
) -> Dict:
    """
    단어 하나가 삽입된 상태에서, 시작 어절이 같은 단어를 찾아 추가한다.
    단, 진행방향의 삼면에 이미 다른 단어의 어절이 없어야 한다.
    이 함수는 두 번째 페이즈로, 처음 추가된 단어와 이어지는 경우에만 단어를 추가한다.
    Args:
        db (Session): 커넥션
        map_queue_desc (Dict): 맵, 큐, 설명 및 단어가 들어있는 사전형 자료
    Returns:
        Dict: 맵, 큐, 설명 및 단어가 들어있는 사전형 자료
    """
    map = map_queue_desc.get("map")
    queue = map_queue_desc.get("queue")
    desc = map_queue_desc.get("desc")
    words = map_queue_desc.get("words")
    num = 2
    while queue:
        point, start_word, dir = queue.popleft()
        next_word = await find_word_info_start_with(start_word, db)
        if not next_word:
            continue
        if (next_word) and (
            (dir and point[1] + next_word.len < 11) or (not dir and point[0] + next_word.len < 11)
        ):
            if point[1] >= 1:
                if dir and map[point[0]][point[1] - 1] != 0:
                    continue
            if point[0] >= 1:
                if not dir and map[point[0] - 1][point[1]] != 0:
                    continue
            original_value = map[point[0]][point[1]]
            for i in range(next_word.len):
                if dir:
                    if i and map[point[0]][point[1] + i] != 0:
                        for j in range(i):
                            map[point[0]][point[1] + j] = original_value
                        break
                    if i and (
                        (point[1] + i + 1 < 11 and map[point[0]][point[1] + i + 1] != 0)
                        or (point[0] - 1 >= 0 and map[point[0] - 1][point[1] + i] != 0)
                        or (point[0] + 1 < 11 and map[point[0] + 1][point[1] + i] != 0)
                    ):
                        for j in range(1, i):
                            map[point[0]][point[1] + j] = 0
                        map[point[0]][point[1]] = original_value
                        break
                    map[point[0]][point[1] + i] = num
                else:
                    if i and map[point[0] + i][point[1]] != 0:
                        for j in range(i):
                            map[point[0] + j][point[1]] = original_value
                        break
                    if i and (
                        (point[1] - 1 >= 0 and map[point[0] + i][point[1] - 1] != 0)
                        or (point[1] + 1 < 11 and map[point[0] + i][point[1] + 1] != 0)
                        or (point[0] + i + 1 < 11 and map[point[0] + i + 1][point[1]] != 0)
                    ):
                        for j in range(1, i):
                            map[point[0] + j][point[1]] = 0
                        map[point[0]][point[1]] = original_value
                        break
                    map[point[0] + i][point[1]] = num
            else:
                words.append({"num": num, "word": next_word.word})
                desc.append({"num": num, "desc": next_word.desc})
                num += 1
                queue = await append_letter_into_queue(
                    point, next_word.len, next_word.word, queue, dir
                )
    return {"map": map, "desc": desc, "words": words}

이 함수에서 검색하는 조건은 Docstring에 적혀있는 바와 동일합니다. 어차피 갈아엎을 함수이니 자세히 읽진 않아도 됩니다. 왜 갈아엎어야 할까요? 조건식이 너무 복잡해보여서? 그것도 맞지만 이 로직에는 커다란 문제점이 하나 있습니다. 조건식이 아닌 부분에서 잠시 생각해보시길 바랍니다.

불필요한 DB 호출

create_puzzle_phase2 함수에서 DB호출하는 부분을 한 번 확인해보시길 바랍니다.

1. 큐에서 꺼내 단어를 DB에서 가져온다.
2. 현재 퍼즐에 단어를 넣을 수 있는지 조건을 검사한다.

위의 로직은 DB에서 단어 데이터를 불러온 뒤에 그 단어가 퍼즐에 추가할 수 있는지 조건을 검사하고 있습니다.

하지만 순서를 바꿔 큐에서 꺼낸 좌표에서 가능한 단어의 길이를 먼저 계산한 후에 DB에서 해당 단어를 가져오는 로직이라면 성능적으로 이득을 볼 수 있다고 판단이 됩니다.

조건 계산 로직을 분리하고 변경된 로직에 맞게 코드를 수정해보죠.

async def find_word_info_start_with(start_word: str, limit: int, db: Session) -> WordInfo:
        """
        DB에서 {start_word}로 시작하면서 길이가 limit 이하인 단어를 찾아 반환한다.
        Args:
            start_word (str): 시작 어절
            limit (int): 가능한 최대 길이
        Returns:
            WordInfo: {start_word}로 시작하면서 길이가 limit 이하인 WordInfo Row 객체
        """
        return (
            db.query(WordInfo)
            .filter(WordInfo.word.like(f"{start_word}%"))
            .filter(WordInfo.len <= limit)
            .order_by(func.rand())
            .first()
        )


async def inspect_possible_length(map: List[int], point: Tuple[int, int], dir: bool) -> int:
    """
    추가할 수 있는 단어의 길이를 반환해주는 함수
    Args:
        map (List[int]): 퍼즐 맵 정보
        point (Tuple[int, int]): 새로운 단어의 시작 지점
        dir (bool): 가로,세로 여부
    Returns:
        Int: 추가할 수 있는 단어의 길이
    """
    if point[1] >= 1:
        if dir and map[point[0]][point[1] - 1] != 0:
            return 0
    if point[0] >= 1:
        if not dir and map[point[0] - 1][point[1]] != 0:
            return 0
    if dir:
        for i in range(1, 5):
            if (
                (point[0] - 1 >= 0 and map[point[0] - 1][point[1] + i] != 0)
                or (point[0] + 1 < len(map) and map[point[0] + 1][point[1] + i] != 0)
                or (point[1] + i < len(map[0]) - 1 and map[point[0]][point[1] + i + 1] != 0)
                or (point[1] + i == len(map[0]) - 1)
            ):
                return i
    else:
        for i in range(1, 5):
            if (
                (point[1] - 1 >= 0 and map[point[0] + i][point[1] - 1] != 0)
                or (point[1] + 1 < len(map[0]) and map[point[0] + i][point[1] + 1] != 0)
                or (point[0] + i < len(map) - 1 and map[point[0] + i + 1][point[1]] != 0)
                or (point[0] + i == len(map) - 1)
            ):
                return i
    return 5

async def create_puzzle_phase2(db: Session = Depends(get_db), map_queue_desc: Dict = Depends(create_puzzle_phase1)) -> Dict:
    """
    단어 하나가 삽입된 상태에서, 시작 어절이 같은 단어를 찾아 추가한다.
    단, 진행방향의 삼면에 이미 다른 단어의 어절이 없어야 한다.
    이 함수는 두 번째 페이즈로, 처음 추가된 단어와 이어지는 경우에만 단어를 추가한다.

    Args:
        db (Session): 커넥션
        map_queue_desc (Dict): 맵, 큐, 설명 및 단어가 들어있는 사전형 자료
    Returns:
        Dict: 맵, 큐, 설명 및 단어가 들어있는 사전형 자료
    """
    map = map_queue_desc.get("map")
    queue = map_queue_desc.get("queue")
    desc = map_queue_desc.get("desc")
    words = map_queue_desc.get("words")
    num = 2
    while queue:
        point, start_word, dir = queue.popleft()
        limit = await inspect_possible_length(map, point, dir)

        if limit <= 1:
            continue
        next_word = await find_word_info_start_with(start_word, limit, db)
        if not next_word:
            continue
        for i in range(next_word.len):
            if dir:
                map[point[0]][point[1] + i] = num
            else:
                map[point[0] + i][point[1]] = num

        words.append({"num": num, "word": next_word.word})
        desc.append({"num": num, "desc": next_word.desc}) 
        num += 1

        queue = await append_letter_into_queue(point, next_word.len, next_word.word, queue, dir)

    return {"map": map, "desc": desc, "words": words}

이렇게 바꾸는게 성능적으로 실제로 이득이 있는지 확인해보겠습니다.

from fastapi import APIRouter, status, Depends
from typing import Dict
from app.utils.puzzle import create_puzzle_phase2

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

@router.get("", status_code=status.HTTP_200_OK)
async def create_puzzle(puzzle: Dict = Depends(create_puzzle_phase2)):
    return await puzzle

엔드포인트 하나 만들어주고,

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
import time

from app.api import router


@asynccontextmanager
async def lifespan(app: FastAPI):
    from app import models
    from app.database import engine

    models.Base.metadata.create_all(bind=engine)

    yield

    pass


app = FastAPI(lifespan=lifespan)
app.include_router(router)


@app.middleware("http")
async def log_request_time(request: Request, call_next):
    start_time = time.time()

    response = await call_next(request)

    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)

    return response

main.py에 response time을 측정할 수 있도록 미들웨어를 하나 달아주겠습니다. FastAPI에서 자동으로 생성해주는 Swagger 문서에서 시간 차이를 보겠습니다.

변경 전 로직

단어 12개 배치에 약 4.8초 정도 걸렸네요.

변경 후 로직

단어 20개 배치에 약 4.9초 정도 걸렸습니다.

같은 단어 수인 12개를 배치할 때 약 2.6초 정도 걸려, 45% 정도의 향상률을 보여줍니다.

로직을 이해하기 쉽도록 그림으로 표현해봤습니다.

참고) 함수형 -> 클래스로 변경

클래스의 생성자도 Depends로 종속성 주입이 가능합니다. PuzzleCreateService 생성자에 Optional[int] = None을 선언해 Query 파라미터로 퍼즐의 크기를 입력받을 수 있게 했습니다.

class PuzzleCreateService:
    def __init__(self, size: Optional[int] = None, db: Session = Depends(get_db)):
        """
        퍼즐의 크기를 받아 맵을 생성한다. 기본 값은 7x7
        Args:
            size (int or None): 퍼즐의 크기 + 1
        """
        self.db = db
        self.map_size = size + 1 if size is not None else 8
        self.map = self.create_map()
        self.queue = deque()
        self.desc = []
        self.words = []
        self.num = 2
        self.word_size = 494047
    def create_map(self) -> List[str]:
        """
        0으로 초기화 된 맵을 생성해 반환한다.
        """
        return [[0 for _ in range(self.map_size)] for _ in range(self.map_size)]
    async def append_letter_into_queue(
        self, start_point: Tuple[int, int], length: int, word: str, dir: bool
    ) -> Deque:
        """
        시작점, 길이, 단어, 방향을 토대로 큐에 다음 단어 추가에 대한 힌트를 큐에 삽입 후 반환한다.
        Args:
            start_point (Tuple): 단어 첫 음절이 위치하는 곳
            length (int): 단어의 길이
            word (str): 단어
            dir (bool): 단어 방향 가로, 세로 여부
        Returns:
            Deque: 다음 단어의 시작점, 시작 음절, 방향이 추가된 큐
        """
        def add_to_queue(coord: Tuple[int, int], letter: str, dir: bool) -> None:
            if 0 <= coord[0] < self.map_size and 0 <= coord[1] < self.map_size:
                self.queue.append(((coord[0], coord[1]), letter, not (dir)))
        if dir:
            add_to_queue((start_point[0], start_point[1] + length - 1), word[-1], dir)
        else:
            add_to_queue((start_point[0] + length - 1, start_point[1]), word[-1], dir)
        if length > 2:
            if length >= 5:
                if dir:
                    add_to_queue((start_point[0], start_point[1] + 2), word[2], dir)
                else:
                    add_to_queue((start_point[0] + 2, start_point[1]), word[2], dir)
            add_to_queue(start_point, word[0], dir)
        return self.queue
    async def find_first_word_info(self) -> WordInfo:
        """
        DB에서 랜덤으로 하나의 단어를 찾아 반환한다.
        Returns:
            WordInfo: 랜덤으로 가져온 WordInfo Row 객체
        """
        random_idx = random.randint(1, self.word_size)
        return self.db.query(WordInfo).filter(WordInfo.id == random_idx).first()
    async def find_word_info_start_with(self, start_word: str, limit: int) -> WordInfo:
        """
        DB에서 {start_word}로 시작하면서 길이가 limit 이하인 단어를 찾아 반환한다.
        Args:
            start_word (str): 시작 어절
            limit (int): 가능한 최대 길이
        Returns:
            WordInfo: {start_word}로 시작하면서 길이가 limit 이하인 WordInfo Row 객체
        """
        return (
            self.db.query(WordInfo)
            .filter(WordInfo.word.like(f"{start_word}%"))
            .filter(WordInfo.len <= limit)
            .order_by(func.rand())
            .first()
        )
    async def create_puzzle_phase1(self) -> None:
        """
        0으로 초기화 된 맵에 첫 단어를 추가,
        이 후 추가될 단어와 그 위치에 대한 힌트를 큐에 삽입,
        첫 단어 설명 추가
        """
        word = await self.find_first_word_info()
        start_y, start_x = 0, 0
        for i in range(word.len):
            self.map[start_y][start_x + i] = 1
        self.words.append({"num": 1, "word": word.word})
        self.desc.append({"num": 1, "desc": word.desc})
        self.queue = await self.append_letter_into_queue(
            (start_y, start_x), word.len, word.word, True
        )
    async def inspect_possible_length(self, point: Tuple[int, int], dir: bool) -> int:
        """
        단어의 시작 좌표와 방향으로 가능한 단어의 최대 길이를 반환한다.
        Args:
            point (Tuple[int,int]): 단어가 시작될 좌표
            dir (bool): 단어의 가로, 세로 여부
        Returns:
            int: 가능한 단어의 길이
        """
        y, x = point
        if x >= 1:
            if dir and self.map[y][x - 1] != 0:
                return 0
        if y >= 1:
            if not dir and self.map[y - 1][x] != 0:
                return 0
        if dir:
            for i in range(1, 5):
                if (
                    (y - 1 >= 0 and self.map[y - 1][x + i] != 0)
                    or (y + 1 < self.map_size and self.map[y + 1][x + i] != 0)
                    or (x + i < self.map_size - 1 and self.map[y][x + i + 1] != 0)
                    or (x + i == self.map_size - 1)
                ):
                    return i
        else:
            for i in range(1, 5):
                if (
                    (x - 1 >= 0 and self.map[y + i][x - 1] != 0)
                    or (x + 1 < self.map_size and self.map[y + i][x + 1] != 0)
                    or (y + i < self.map_size - 1 and self.map[y + i + 1][x] != 0)
                    or (y + i == self.map_size - 1)
                ):
                    return i
        return 5
    async def create_puzzle_phase2(self) -> Dict:
        """
        단어 하나가 삽입된 상태에서, 시작 어절이 같은 단어를 찾아 추가한다.
        단, 진행방향의 삼면에 이미 다른 단어의 어절이 없어야 한다.
        이 함수는 두 번째 페이즈로, 처음 추가된 단어와 이어지는 경우에만 단어를 추가한다.
        Args:
            db (Session): 커넥션
            map_queue_desc (Dict): 맵, 큐, 설명 및 단어가 들어있는 사전형 자료
        Returns:
            Dict: 맵, 큐, 설명 및 단어가 들어있는 사전형 자료
        """
        await self.create_puzzle_phase1()
        while self.queue:
            point, start_word, dir = self.queue.popleft()
            limit = await self.inspect_possible_length(point, dir)
            if limit <= 1:
                continue
            next_word = await self.find_word_info_start_with(start_word, limit)
            if not next_word:
                continue
            for i in range(next_word.len):
                if dir:
                    self.map[point[0]][point[1] + i] = self.num
                else:
                    self.map[point[0] + i][point[1]] = self.num
            self.words.append({"num": self.num, "word": next_word.word})
            self.desc.append({"num": self.num, "desc": next_word.desc})
            self.num += 1
            self.queue = await self.append_letter_into_queue(
                point, next_word.len, next_word.word, dir
            )
        return {"map": self.map, "desc": self.desc, "words": self.words}

Phase3: 남은 공간 채우기

남은 공간을 최대한 채우기 위해 퍼즐을 순회하며 아직 0인 좌표에 대해서 단어를 추가할 수 있는지 검증한 뒤 Phase2처럼 가능한 이어서 붙이기로 했습니다.

# 가능한 단어 길이 검사 함수 로직 추가
async def inspect_possible_length(
        self, point: Tuple[int, int], dir: bool, phase: int = 2
    ) -> int:
        """
        단어의 시작 좌표와 방향으로 가능한 단어의 최대 길이를 반환한다.

        Args:
            point (Tuple[int,int]): 단어가 시작될 좌표
            dir (bool): 단어의 가로, 세로 여부
            phase (int): 퍼즐 생성 단계
        Returns:
            int: 가능한 단어의 길이
        """
        y, x = point

        if phase == 3:
            if y > 0 and self.map[y - 1][x] != 0:
                return 0
            if y < self.map_size - 1 and self.map[y + 1][x] != 0:
                return 0
            if x > 0 and self.map[y][x - 1] != 0:
                return 0
            if x < self.map_size - 1 and self.map[y][x + 1] != 0:
                return 0
        if x >= 1:
            if dir and self.map[y][x - 1] != 0:
                return 0
        if y >= 1:
            if not dir and self.map[y - 1][x] != 0:
                return 0

Phase3에서는 단어 시작하는 곳에서도 상하좌우를 확인해야합니다.

async def fill_puzzle_until_queue_empty(self) -> None:
        """
        큐가 빌 때까지 퍼즐에 단어를 추가시킨다.
        """
        while self.queue:
            point, start_word, dir = self.queue.popleft()
            limit = await self.inspect_possible_length(point, dir)

            if limit <= 1:
                continue
            next_word = await self.find_word_info_start_with(start_word, limit)

            if not next_word:
                continue
            for i in range(next_word.len):
                if dir:
                    self.map[point[0]][point[1] + i] = self.num
                else:
                    self.map[point[0] + i][point[1]] = self.num

            self.words.append({"num": self.num, "word": next_word.word})
            self.desc.append({"num": self.num, "desc": next_word.desc})
            self.num += 1

            self.queue = await self.append_letter_into_queue(
                point, next_word.len, next_word.word, dir
            )

    async def create_puzzle_phase2(self) -> None:
        """
        단어 하나가 삽입된 상태에서, 시작 어절이 같은 단어를 찾아 추가한다.
        단, 진행방향의 삼면에 이미 다른 단어의 어절이 없어야 한다.
        이 함수는 두 번째 페이즈로, 처음 추가된 단어와 이어지는 경우에만 단어를 추가한다.
        """
        await self.create_puzzle_phase1()
        await self.fill_puzzle_until_queue_empty()

Phase2에서 큐가 빌 때까지 단어 추가해주던 로직을 추출했고,

async def create_puzzle_phase3(self) -> Dict:
        """
        퍼즐의 비어있는 공간에 단어를 추가한다.
        Returns:
            Returns:
            Dict: 맵, 설명 및 단어가 들어있는 사전형 자료
        """
        await self.create_puzzle_phase2()
        for i in range(self.map_size - 1):
            for j in range(self.map_size - 1):
                if self.map[i][j] == 0:
                    hor = await self.inspect_possible_length((i, j), True, phase=3)
                    ver = await self.inspect_possible_length((i, j), False, phase=3)
                    dir = True if hor > ver else False
                    if not (hor and ver):
                        continue
                    else:
                        word = await self.find_word_info_start_with("", hor if hor > ver else ver)
                        if not word:
                            continue
                        for k in range(word.len):
                            if dir:
                                self.map[i][j + k] = self.num
                            else:
                                self.map[i + k][j] = self.num

                        self.words.append({"num": self.num, "word": word.word})
                        self.desc.append({"num": self.num, "desc": word.desc})
                        self.num += 1
                        self.queue = await self.append_letter_into_queue(
                            (i, j), word.len, word.word, dir
                        )
                        await self.fill_puzzle_until_queue_empty()

        return {"map": self.map, "desc": self.desc, "words": self.words}

추출한 로직을 Phase3에서도 사용하도록 구성했습니다.

저는 이전에 Class형태로 의존성을 주입하도록 코드를 리팩토링 했기 때문에 엔드포인트도 수정이 필요합니다.

from fastapi import APIRouter, status, Depends
from app.utils.puzzle import PuzzleCreateService

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


@router.get("", status_code=status.HTTP_200_OK)
async def create_puzzle(puzzle: PuzzleCreateService = Depends(PuzzleCreateService)):
    return await puzzle.create_puzzle_phase3()

추가적인 성능 개선

Phase3까지 모두 돌아갔을 때 어느정도 시간이 걸리는지 확인해볼까요? 위의 시간 테스트에서 맵 크기를 10x10으로 했기 때문에 동일하게 하겠습니다. 직접 해보실 분들은 Swagger 문서에서 size 파라미터에 10 적고 해보시면 됩니다.

단어가 총 29개나 들어가있네요. 단어 수가 많기도 하지만 게임판을 생성하는데에 약 8.5초나 걸리는건 꽤나 아쉽습니다.

성능을 한 번 더 개선해봅시다.

EXPLAIN SELECT * FROM wordinfo
WHERE word LIKE '마%' 
AND len <= 5
ORDER BY RAND()
LIMIT 1;

"마"로 시작하면서 길이가 5 이하인 단어를 찾을 때, SQLAlchemy ORM을 통해 생성될 SQL ROW Query의 실행 계획을 확인해봅시다.

확인해봐야할 row가 47만개가 넘어가면서, 탐색해야할 범위가 큰 것을 알 수 있습니다. 실제로 저 SQL문을 실행하면 약 0.25초가 걸립니다.

탐색 범위를 좁히기 위해 인덱스를 테이블에 걸어주는 방법이 있습니다. 우리는 word와 len에 조건을 걸어서 탐색하고 있습니다. 그런데 인덱스는 카디널리티가 클 수록 성능이 좋아지는 경향이 있습니다. len은 2~5까지의 범위뿐이라 카디널리티가 매우 작습니다. 따라서, word에 인덱스를 걸어보도록 하죠.

class WordInfo(BaseMin, Base):
    __tablename__ = "wordinfo"

    word = Column(VARCHAR(5), nullable=False, index=True)
    desc = Column(VARCHAR(200), nullable=False)
    len = Column(Integer, nullable=False)
    pos = Column(VARCHAR(7), nullable=False)

모델을 수정하고 alembic으로 마이그레이션하는 방법이 있고,

CREATE INDEX ix_wordinfo_word ON wordinfo(word);

MySQL에서 직접 인덱스를 생성하는 방법도 있습니다.

인덱스 생성 후에 실행 계획을 다시 살펴보겠습니다.

탐색해야 할 row가 현저히 줄어들었습니다. 위에서는 null 값이었던 key에 저희가 생성해준 인덱스도 들어있는 것을 확인할 수 있습니다.

실제로 조회 성능이 0.25초에서 0.015초로 무려 약 94%나 향상됐네요. 다시 한 번 퍼즐을 생성해봅시다.

단어 수가 1개 적은 28개이긴 하지만, 약 3.4초 정도로 나왔습니다. 인덱스를 생성 이전 8.5초에서 생성 후 3.4초로 약 60%의 향상률을 보였습니다.

다음 포스트에서는 결과적으로 생성된 데이터의 형태를 알아보고 데이터베이스에 추가하는 작업을 해보겠습니다. 읽어주셔서 감사합니다.

지난 번에 정제했던 데이터를 데이터베이스에 삽입해보도록 하겠습니다.

프론트엔드까지 구현해볼지는 모르겠지만 백엔드 서비스 형태로 게임을 제공해보기 위해서 FastAPI 프레임워크를 사용하기로 했습니다.

MySQL이 설치가 되어있다는 전제하에 진행해보겠습니다.

# 프로젝트 구조
ROOT FOLDER
ㄴ app
  ㄴ main.py
  ㄴ config.py
  ㄴ database.py
  ㄴ models.py
ㄴ word-data
    ㄴ json files... 로우 데이터 JSON 파일들
ㄴ extract_word.py 데이터 정제 스크립트
ㄴ runserver.py 서버 실행 스크립트

초기 서버 설정

실행환경 구성

# 실행환경 구성
python -m venv venv 가상환경 설정
pip install uvicorn[standard] fastapi sqlalchemy pymysql

# .env
# 각자의 환경에 맞게 .env 파일을 설정해주면 됩니다.
DB_USERNAME={USERNAME}
DB_PASSWORD={PASSWORD}
DB_HOST={HOST}
DB_PORT={PORT}
DB_NAME={NAME}

# config.py
from dotenv import load_dotenv
from functools import lru_cache

import os

load_dotenv()


class Settings:
    # DB Settings
    DB_USERNAME = os.getenv("DB_USERNAME")
    DB_HOST = os.getenv("DB_HOST")
    DB_PASSWORD = os.getenv("DB_PASSWORD")
    DB_NAME = os.getenv("DB_NAME")
    DB_PORT = os.getenv("DB_PORT")


@lru_cache
def get_settings():
    return Settings()


settings = get_settings()

코드 전역에서 사용할 서버 세팅을 관리해주는 파일입니다. lru_cache를 사용해 settings를 사용할 때 추가연산을 하지 않도록 했습니다.

# database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from app.config import settings

engine = create_engine(
    "mysql+pymysql://{username}:{password}@{host}:{port}/{name}".format(
        username=settings.DB_USERNAME,
        password=settings.DB_PASSWORD,
        host=settings.DB_HOST,
        port=settings.DB_PORT,
        name=settings.DB_NAME,
    )
)
SessionLocal = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,
)

Base = declarative_base()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

서버와 MySQL 서버를 연결해주는 세션을 생성하는 파일입니다. API에서 사용할 때 get_db 함수를 Depends 해줄 수 있게 미리 선언했습니다.

# main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    from app import models
    from app.database import engine

    models.Base.metadata.create_all(bind=engine)

    yield

    pass

app = FastAPI(lifespan=lifespan)

# runserver.py
import uvicorn

if __name__ == "__main__":
    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

서버 실행과 관련된 파일입니다. 서버 실행 시에 데이터베이스와 연결되도록 lifespan으로 관리해줍니다.

이전에는 on_event를 사용해서 서버 시작, 종료 시 이벤트를 관리했는데 이는 사장될 문법이므로 지양하는게 좋습니다.

DB 테이블 생성

# models.py
from sqlalchemy import Column, Integer, VARCHAR

from app.database import Base


class BaseMin:
    id = Column(Integer, primary_key=True, index=True)

class WordInfo(BaseMin, Base):
    __tablename__ = "wordinfo"

    # 단어 이름
    word = Column(VARCHAR(7), nullable=False)
    # 단어 설명
    desc = Column(VARCHAR(200), nullable=False)
    # 단어 길이
    len = Column(Integer, nullable=False)
    # 단어 품사
    pos = Column(VARCHAR(7), nullable=False)

BaseMin 클래스를 선언해서 이후 테이블에서 id를 Primary Key로 반복 선언하지 않게 따로 빼줬습니다.

데이터 DB 삽입

# extract_word.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.models import WordInfo

import re
import os
import json
from app.config import settings


DIR = os.getcwd()
DATA_DIR = DIR + "/word-data"
json_list = os.listdir(DATA_DIR)

def process_word(word):
    """
    자음이 포함된 단어 제거, 단어에 한글을 제외한 문자 및 공백 제거, 지정 길이를 벗어나는 단어 제거
    위 세 가지 일을 한 뒤 단어와 그 길이를 반환합니다

    Args:
        word (str): 정제할 단어
    Returns:
        dict: {단어, 단어의 길이}
    """
    word = word.strip()
    consonant_mixed = re.search(r'[ㄱ-ㅎ]', word)

    if consonant_mixed:
        return None

    result = re.sub(r'[^가-힣]', '', word)
    length = len(result)
    if length > 7 or length < 2:
        return None

    return {"word": result, "length": length}

def process_senseinfo(senseinfo):
    """
    단어의 품사가 없거나 '품사 없음'일 시 제거, 방언 및 북한어 제거, 설명에 특이점이 있으면 제거
    위 세 가지 일을 한 뒤 설명과 단어의 품사를 반환합니다

    Args:
        senseinfo (dict): 품사, 설명 데이터가 들어있는 딕셔너리
    Returns:
        dict: {설명, 품사}
    """
    definition = senseinfo["definition"].strip()
    word_type = senseinfo["type"].strip()
    pos = senseinfo.get("pos", None)

    if pos is None or pos == "품사 없음":
        return None

    if word_type == "방언" or word_type == "북한어":
        return None

    if len(definition) > 200 or "&" in definition or "img" in definition or "<FL>" in definition or "규범 표기" in definition or "준말" in definition or "옛말" in definition or "-" in definition:
        return None

    return {"definition": definition, "pos": pos}

if __name__ == "__main__":
    """
    JSON 파일로부터 단어 로우 데이터를 읽어들이고, 정제한 뒤 WordInfo 테이블에 추가합니다.
    """
    engine = create_engine(
    "mysql+pymysql://{username}:{password}@{host}:{port}/{name}".format(
        username=settings.DB_USERNAME,
        password=settings.DB_PASSWORD,
        host=settings.DB_HOST,
        port=settings.DB_PORT,
        name=settings.DB_NAME,
        )
    )

    Session = sessionmaker(bind=engine)
    db = Session()
    word_list = []

    for i, _json in enumerate(json_list):
        print(f"{len(json_list)}/{i+1}번 째 파일, 파일명 : {_json}")
        with open(DATA_DIR+"/"+_json, encoding="utf-8") as f:
            json_data = json.load(f)
            for word_data in json_data["channel"]["item"]:
                processed_word = process_word(word_data["wordinfo"]["word"])
                processed_senseinfo = process_senseinfo(word_data["senseinfo"])
                if processed_word and processed_senseinfo:
                    word_list.append({
                        "word": processed_word["word"],
                        "desc": processed_senseinfo["definition"],
                        "pos": processed_senseinfo["pos"],
                        "len": processed_word["length"]
                    })
                else:
                    continue
            print(f"현재 단어 수 : {len(word_list)}")

    db.bulk_insert_mappings(WordInfo, word_list)
    db.commit()

    db.close()

글 최상단의 프로젝트 구조대로 구성하고 위 스크립트를 실행하면

로그가 위와 같이 나오며 실행됩니다. 50만개가 넘는 row를 한 번에 삽입하는 경우에는 개별적으로 add하는게 아니라 bulk_insert 기능을 활용하는게 성능적으로 훨씬 우수합니다.

MySQL WorkBench에서 확인해본 결과입니다. 잘 들어가있죠? 다음 포스트에서는 십자말풀이 퍼즐 생성을 다뤄보겠습니다. 읽어주셔서 감사합니다.

작업 레포지토리 : https://github.com/fnzksxl/word-puzzle

정제 전 데이터 소스

국립국어원의 우리말샘의 사전 데이터를 로우 데이터로 삼았습니다.

그 중 JSON 형태로 제공되는 파일의 일부를 살펴보겠습니다.

channel 아래 item 배열에 단어 데이터들이 들어있는 형태네요.

{
  "channel": {
    "total": 50000,
    "title": "사전 검색",
    "description": "사전 검색 결과",
    "item": [
      {
        "wordinfo": {
          "conju_info": [
            {
              "conjugation_info": {
                "pronunciation_info": {
                  "pronunciation": "걷치레싱만"
                },
                "conjugation": "겉치레식만"
              }
            }
          ],
          "pronunciation_info": [
            {
              "pronunciation": "걷치레식"
            }
          ],
          "word_unit": "어휘",
          "word": "겉치레-식",
          "original_language_info": [
            {
              "original_language": "겉치레",
              "language_type": "고유어"
            },
            {
              "original_language": "式",
              "language_type": "한자"
            }
          ],
          "word_type": "혼종어"
        },
        "group_order": 1,
        "group_code": 46174,
        "link": "http://opendict.korean.go.kr/dictionary/view?sense_no=643318",
        "target_code": 643318,
        "senseinfo": {
          "definition": "겉으로 보기에만 좋게 꾸미어 드러내는 방식.",
          "sense_no": "001",
          "type": "일반어",
          "example_info": [
            {
              "source": "노컷뉴스 2009년 7월",
              "example": "그러나 주변에서 일상적으로 듣는 {겉치레식} 격려는 아무런 힘을 발휘할 수 없다."
            },
            {
              "source": "서울신문 2013년 3월",
              "example": "“통과의례, {겉치레식의} 인사 청문회 제도에 대한 개선이 필요하다.”라는 지적도 나온다."
            }
          ],
          "definition_original": "겉으로 보기에만 좋게 꾸미어 드러내는 방식.",
          "pos": "명사"
        }
      },

데이터 정제

1. 단순 데이터 추출

# 단어 이름, 설명, 품사 추출하는 스크립트
import json
import os


DIR = os.getcwd()
DATA_DIR = DIR + "/word-data"
json_list = os.listdir(DATA_DIR)

for i, _json in enumerate(json_list):
        print(f"{len(json_list)}/{i+1}번 째 파일, 파일명 : {_json}")
        with open(DATA_DIR+"/"+_json, encoding="utf-8") as f:
            json_data = json.load(f)
            for word_data in json_data["channel"]["item"]:
                word = word_data["wordinfo"]["word"]
                senseinfo = word_data["senseinfo"]
                pos = senseinfo["pos"]
                definition = senseinfo["definition"]

위 데이터 중 단어의 이름인 ["wordinfo"]["word"], 단어의 설명인 ["senseinfo"]["definition"], 단어의 품사인 ["senseinfo"]["pos"] 이렇게 세 개를 가져오기로 했습니다.

2. 데이터 확인

단어 설명에 img 나 FL 태그가 들어있는 경우도 있었고,

어휘에 특수문자가 들어가거나,

자음만 존재하는 등 십자말풀이에서 쓰지 못할 데이터가 존재합니다. 이외에도 설명에 %lt %gt 같은 용어, 너무 긴 단어나 설명이 존재하기도 했습니다.

3. 데이터 정제

위에서 확인해본 데이터를 배제함과 동시에 십자말풀이에 쓸 수 있도록 데이터를 정제해보겠습니다.

단어

1. 2 <= 단어의 길이 <= 7 인 경우만 추출할 것
2. 단어에 자음만 존재하거나 한글을 제외한 다른 문자가 있으면 스킵할 것
3. 단어에 띄워쓰기는 제거할 것

위 조건을 만족하도록 단어를 정제하는 함수를 정규표현식을 이용해 작성해봅시다.

# 원하는 조건으로 단어를 정제하는 함수
def process_word(word):
    word = word.strip()
    consonant_mixed = re.search(r'[ㄱ-ㅎ]', word)

    if consonant_mixed:
        return None

    result = re.sub(r'[^가-힣]', '', word)
    length = len(result)
    if length > 7 or length < 2:
        return None

    return {"word": result, "length": length}

1. re.search(pattern, string) 함수는 패턴과 문자열이 일치하는 부분이 없으면 None, 있으면 그 부분을 반환해줍니다.
2. re.sub(pattern, replacement, string) 함수는 패턴과 문자열이 일치하는 모든 부분을 replacement로 치환해줍니다. r'[^가-힣]'의 패턴의 경우 한글을 제외한 모든 문자를 의미합니다. ^ 기호는 차집합의 의미로 생각하면 이해가 쉽습니다.

다른 메타 문자나 정규표현식의 더 많은 함수는 파이썬에서 제공해주는 문서에서 자세히 확인해볼 수 있습니다.

# EXAMPLE
word_list = ["ㄱ-", "게", "게임", "게이밍 모니터"]
processed_word_list = []
for word in word_list:
    processed_word_list.append(process_word(word))
print(processed_word_list)

# 출력 결과
[None, None, {'word': '게임', 'length': 2}, {'word': '게이밍모니터', 'length': 6}]

설명 및 품사

1. 품사는 반드시 있어야 할 것
2. 방언과 북한어는 제외할 것
3. 설명의 길이 > 200 이거나 태그가 들어있으면 제외할 것
4. 규범 표기, 옛말, 준말 등의 설명이 있으면 제외할 것
5. -가 들어가는 경우는 제외할 것

5번의 경우는 설명에서 이상한 부분이 있었지만 따로 캡쳐하지 못했습니다.

위 조건을 만족하도록 설명과 품사를 전제하는 함수를 만들어 봅시다.

# 원하는 조건으로 설명을 정제하는 함수
def process_senseinfo(senseinfo):
    definition = senseinfo["definition"].strip()
    word_type = senseinfo["type"].strip()
    pos = senseinfo.get("pos", None)

    if pos is None or pos == "품사 없음":
        return None

    pos = pos.strip()

    if word_type == "방언" or word_type == "북한어":
        return None

    if len(definition) > 200 or "&" in definition or "img" in definition or "<FL>" in definition or "규범 표기" in definition or "준말" in definition or "옛말" in definition or "-" in definition:
        return None

    return {"definition": definition, "pos": pos}

# EXAMPLE
senseinfo_list = [{"definition": "&&안녕하세요", "type": "고유어", "pos": "명사"},
                  {"definition": "안녕하세요", "type": "북한어", "pos": "명사"},
                  {"definition": "안녕하세요", "type": "고유어", "pos": "명사"},
                  {"definition": "안녕하세요", "type": "고유어", "pos": "품사 없음"},
                  {"definition": "안녕하세요 <img", "type": "고유어", "pos": "명사"}]
processed_senseinfo_list = []
for senseinfo in senseinfo_list:
    processed_senseinfo_list.append(process_senseinfo(senseinfo))
print(processed_senseinfo_list)

# 출력 결과
[None, None, {'definition': '안녕하세요', 'pos': '명사'}, None, None]

다음 포스트에서는 정제한 단어 데이터들을 데이터베이스에 삽입해보겠습니다. 읽어주셔서 감사합니다.

Gmail 설정하기

장고와 같은 외부 앱에서 Gmail을 사용하려면 앱 비밀번호가 필요하다 비밀번호 설정하기 <= 이 링크를 타고 들어가서 비밀번호를 생성해주도록 하자.

필자가 테스트용으로 만들어본 앱비밀번호 16자리이다. 이 화면에서 벗어나면 다시 확인할 수가 없으니 확인 버튼을 누르기 전에 꼭 미리 저장해놔야 한다.

위 비밀번호는 외부에 반출되면 안되므로 프로젝트의 env 파일에 잘 숨겨두자.

이메일 인증 로직

이메일을 보내기 전에 구현해볼 인증 로직을 먼저 설명해보도록 하겠다.

1. 유저에게서 이메일을 입력받고 주소로 랜덤 코드를 발송한다.
2. 서버에서는 redis에 이메일:랜덤 코드 쌍으로 5분의 유효기간을 주고 저장한다.
3. 유저가 코드 확인 후 검증 API로 확인한 코드를 전송한다.
4. 서버에서 코드의 일치 여부를 확인하고 일치한다면 이 코드의 유효기간을 삭제한다.
5. 코드를 클라이언트 측으로 반환해준다.
6. 회원가입 시 클라이언트는 유저정보와 함께 이 코드를 같이 서버로 보내 이메일 확인이 되었음을 증명한다.
7. 회원가입 성공 시 서버에서는 이메일:랜덤 코드를 redis에서 제거한다.

Django에서 이메일 보내기

장고에서는 이메일을 보내는 클래스가 이미 존재한다. (역시 batteries included 철학답다)

#project_name/settings.py

EMAIL_BACKEND = "django.core.mail.backends.smtp.EmailBackend"
EMAIL_HOST = "smtp.gmail.com"
EMAIL_PORT = 587
EMAIL_HOST_USER = env("EMAIL_ID")
EMAIL_HOST_PASSWORD = env("EMAIL_PW")
EMAIL_USE_TLS = True
DEFAULT_FROM_EMAIL = EMAIL_HOST_USER

프로젝트의 settings.py에 위 내용을 추가해주도록 하자. env 파일에 이메일을 보내는 주체(EMAIL_ID), 위에서 생성한 비밀번호(EMAIL_PW)가 존재한다는 가정 하에 작성된 내용이다.

env 파일을 읽어오는 방법은 사용자에 따라 가지각색이므로 어떻게 불러와도 상관없다.

위 이메일 인증 로직을 따라가면서 구현해보자.

로직 1번

먼저 랜덤 코드를 생성해 이메일로 발송해야한다.

#project_app/utils.py NEW

import strings
import secrets

class SendEmailHelper:
    def make_random_code_for_register():
        digit_and_alpha = string.ascii_letters + string.digits
        return "".join(secrets.choice(digit_and_alpha) for _ in range(6))

sendEmailHelper = SendEmailHelper()

이메일을 보내는 앱에 utils.py를 생성해 랜덤한 6자리의 코드를 생성해보자. 위 코드를 사용하면 알파벳+숫자로 이루어진 코드가 생성된다.

#project_app/views.py

from django.core.mail import EmailMessage
from .utils import sendEmailHelper

class EmailVerifyView(APIView):

    def post(self, request, *args, **kwargs):
        email = request.data.get("email")
        code = sendEmailHelper.make_random_code_for_register()
        message = code
        subject = "EMAIL 제목"
        to = [email]
        mail = EmailMessage(subject=subject, body=message, to=to)
        mail.content_subtype = "html" # html형태로 템플릿을 만들었을 때 필요함
        mail.send()

        return Response({"detail": "Success to send Email"}, status=status.HTTP_202_ACCEPTED)

간단하게 유저로부터 이메일을 입력 받아 주소로 랜덤 코드 6자리를 보내는 코드이다.

message에는 이메일의 내용이, subject에는 제목이, to에는 받는 주소가 들어가야 한다. (to에는 반드시 튜플이나 리스트의 형태의 데이터가 들어가야 함을 주의하자)

포스트맨으로 테스트를 한 번 해보자.

오! 이메일이 잘 들어왔고, 포스트맨을 보면?

응답도 잘 들어와있으나 시간을 한 번 보시라. 4초에 가까운 시간이 걸리는데 이를 일반적인 HTTP 통신 상에서 처리하려고 하면 안 된다.

이를 지난 포스트에서 다루었던 Celery를 이용해 백그라운드로 넘겨서 진행해주자.

위 링크를 통해 기본적인 Celery 세팅은 다 되었다고 판단하고 진행하겠다.

#project_app/tasks.py NEW

from django.core.mail import EmailMessage

from wtnt.celery import app
from .utils import sendEmailHelper


@app.task
def send_email(email):
    code = sendEmailHelper.make_random_code_for_register()
    message = sendEmailHelper.get_template(code)
    subject = "%s" % "[WTNT] 이메일 인증 코드 안내"
    to = [email]
    mail = EmailMessage(subject=subject, body=message, to=to)
    mail.content_subtype = "html"
    mail.send()
    return "Success to send email"

tasks.py에 이메일 보내는 함수를 따로 빼내어 작성했다. 필자는 sendEmailHelper에 이메일의 템플릿을 따로 작성해두어 이메일의 내용에 약간의 구색을 갖춰두었다.

#project_app/views.py

from .tasks import send_email

class EmailVerifyView(APIView):

    def post(self, request, *args, **kwargs):
        email = request.data.get("email")
        try:
            send_email.delay(email)
            return Response({"detail": "Success to send Email"}, status=status.HTTP_202_ACCEPTED)
        except Exception as e:
            return Response({"error": e}, status=status.HTTP_400_BAD_REQUEST)

Celery에 task를 할당하려면 delay 함수를 사용하면 된다. 좀 더 자세히 알아보고 싶다면 공식 문서를 참고해보는게 좋다.

celery -A project worker -l INFO

Celery를 켜주고 같은 API에 요청을 보내보자.

Celery에 요청이 received되고 앞에서 포스트맨을 사용했을 때와 비슷한 시간이 걸려서 Task를 완수했다.

이메일도 예쁘게 도착한 모습.

그리고 포스트맨에서는? 125ms 밖에 걸리지 않았다. 야호!

이제 로직의 1번을 달성했다.

로직 2번

redis에 이메일:랜덤 코드 쌍을 유효기간 5분을 두고 저장한다. 이는 매우 간단하다.

from django_redis import get_redis_connection
from django.core.mail import EmailMessage

from wtnt.celery import app
from .utils import sendEmailHelper

client = get_redis_connection("default")


@app.task
def send_email(email):
    code = sendEmailHelper.make_random_code_for_register()
    client.set(email, code, ex=300)
    message = sendEmailHelper.get_template(code)
    subject = "%s" % "[WTNT] 이메일 인증 코드 안내"
    to = [email]
    mail = EmailMessage(subject=subject, body=message, to=to)
    mail.content_subtype = "html"
    mail.send()
    return "Success to send email"

tasks.py를 위와 같이 변경해주었다.

client.set(email, code, ex=3000)

이 코드가 함수 안에 추가된 것인데, 이는 email 키에 code값을 300초(5분)동안 저장하겠다는 것이다. ex 값을 60으로 바꿔놓고 확인해보면 1분 뒤에 키가 자동으로 삭제되어있을 것이다.

이메일을 보낸 뒤 redis-cli로 확인해보면 이렇게 이메일 키 값이 생겨져있다.

아무런 작업을 하지않고 ex만큼 시간이 지난 뒤 다시 확인해보면 키가 없어진 것을 볼 수 있다.

로직 3,4,5번

유저가 보낸 코드가 일치하는지 확인하고 redis 내부의 값을 변경해줘야한다.

#project_app/views.py

from django_redis import get_redis_connection

client = get_redis_connection()

def patch(self, request, *args, **kwargs):
        code = request.data.get("code")
        email = request.data.get("email")
        answer = client.get(email)
        if code == answer:
            client.set(email, code)
            return Response({"code": code}, status=status.HTTP_200_OK)
        else:
            return Response({"error": "Code Not Matched"}, status=status.HTTP_400_BAD_REQUEST)

위에서 작성한 EmailVerifyView에 patch 메소드를 추가해줬다. 유저로부터 코드를 받아 일치하는지 확인하는 API다.

코드가 일치하면 다시 이메일:랜덤 코드로 set해주는데 이번에는 ex옵션이 없다. 이렇게 하면 유효기간을 없앨 수 있다.

이 후에 클라이언트 쪽으로 유저가 입력한 코드를 다시 보내준다.

로직 6,7번

#project_app/views.py

def post(self, request):
        code = request.data.get("code")
        email = request.data.get("email")
        if code != client.get(email):
            return Response({"error": "Code Not Matched"}, status=status.HTTP_400_BAD_REQUEST)

        # ...

        client.delete(email)

회원가입을 담당하는 뷰에서 code와 email을 추가로 입력받고 redis에서 일치하는지 확인해 유저가 이메일 인증 작업을 완료했는지 검증한다.

회원가입 로직이 완료되면 이메일 키를 삭제해준다.

간단하게 포스트 디테일 페이지로 유저가 접근할 때마다 포스트의 조회수를 1씩 늘려주는 방법이 있다.

def get(self, request, *args, **kwargs):
        team_id = kwargs.get("team_id")
        try:
            team = Team.objects.get(id=team_id)
            team.view += 1
            team.save()
            teamSerializer = TeamCreateSerializer(team)
            response = createSerializerHelper.make_response(teamSerializer.data, request.user.id)

            return Response(response, status=status.HTTP_200_OK)

        except Team.DoesNotExist:
            return Response({"error": "No Content"}, status=status.HTTP_404_NOT_FOUND)

• 문제점 API에 접근할 때마다 조회수를 무조건 올려주게 되어 있으므로 유저가 마음만 먹으면 무한정으로 조회수가 올라간다!

어떻게 해결해볼 수 있을까?

이번에 조회수 기능을 개발하면서 목표는 당일 중복 조회 불가이다. 00시시가 되면 다시 조회수를 집계 해주겠다는 뜻

1. 쿠키에 조회 내역 정보를 담아서 보내준다

예를 들어 id가 1인 유저가 10번, 20번 게시글에 접근했다면 쿠키에 view:1=10|20 같은 방식으로 저장해주는 것이다.

쿠키를 보낼 때, 쿠키의 생명주기를 당일 00시까지로 제한해 발급해주면 다음 날부터는 다시 조회수가 집계된다.

response.set_cookie("view", 조회 내역, expires=내일까지의 시간)

이제 쿠키의 유무와 내용만 먼저 확인하면 조회수를 집계할 수 있다. 이 방법에는 문제가 없을까?

쿠키는 클라이언트가 관리할 수 있다. 즉, 쿠키를 의도적으로 삭제할 수 있다는 것.

물론, 지금 개발하고 있는 서비스는 유튜브처럼 조회수가 비즈니스적으로 중요하지 않아 유저가 쿠키를 삭제해서 조회수를 올린다고 큰 문제가 되지는 않지만 쉽게 눈치챌 수 있는 기믹이다.

2. 조회 내역 정보를 DB에 담아두고 서버에서 관리한다

조회하는 유저와 게시글 정보 쌍을 DB에 저장해 조회수 중복을 방지할 수 있다.

user_id와 post_id를 저장하는 테이블이 있다면 가능한 얘기. 하지만 이 방법은 조회의 주체가 비회원이라면 관리가 불가능하다. 비회원의 id를 0으로 간주해 저장하는 방법도 있겠으나 다른 사람이 조회했을 때도 집계가 되지 않을 수 있다.

조회하는 유저의 IP정보를 저장하면 어떻게 될까? 서비스의 특성상 이 서비스는 주로 학교의 동아리 회원들이 학교에서 사용할 가능성이 높다. 즉, 같은 IP로 접근할 가능성이 꽤나 있다는 것.

따라서 user_id와 IP를 조합한 정보를 저장해보도록 하자. 게시글을 조회할 때마다 RDBMS에 I/O를 더 가져가기가 싫어서 이 정보를 redis에 저장해보기로 했다.

redis는 집합 자료형을 제공하므로 SADD 명령을 실행했을 때 반환 값을 통해 중복 여부를 체크하기에 용이하다.

또한 Celery-beat로 스케쥴링해주면 특정 키를 정해진 시간에 삭제해 하루가 지나면 다시 조회수를 카운팅해줄 수도 있다.

Celery란 무엇일까? 링크에 Celery의 개요와 장단점이 잘 설명되어 있다.

로직

조회 관련 정보는 views:team_id 집합에 저장된다.

1. 유저가 게시글에 접근하면 redis에 유저id_IP 조합의 밸류값이 있는지 확인한다.
2. 존재한다면 -> PASS 오늘 첫 방문이라면 -> views:team_id 집합에 유저id_IP조합을 저장후 조회 수를 +1 해준다.
3. 매일 자정마다 views:로 시작하는 키를 모두 제거한다.

2번 방법을 통해 한 번 구현해보도록 하자. 우선 Celery를 설치하고, Django 프로젝트에 이식하자. (redis는 로컬에 설치되어 있거나 docker로 실행되어 있다고 가정한다.)

# Celery 관련 라이브러리 설치
pip install celery django-celery-beat django-celery-results

#project_name/settings.py

INSTALLED_APPS = [
    # 기존 APP들 +
    "celery",
    "django_celery_beat",
    "django_celery_results",
]

CELERY_BROKER_URL = "redis://127.0.0.1:6379/0"
CELERY_RESULT_BACKEND = "django-db"
CELERY_TIMEZONE = "Asia/Seoul"

INSTALLED_APPS에 방금 설치한 내용을 업데이트, Celery 설정을 추가해준다. 아래의 설정 파일을 따라해 Celery가 장고의 설정을 가져오는 경우에 RESULT_BACKEND를 저렇게 설정해 줄 수 있다.

# django_celery_results 관련 DB를 migrate 해준다.
python manage.py migrate

#project_name/celery.py NEW

import os

from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "#project_name.settings")
app = Celery("#project_name")

app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
    print(f"Request: {self.request!r}")

참고: Windows 환경이라면? 링크를 참고해서 설정을 추가해야한다. 필자도 Windows 환경이었는데 도움을 많이 받았다.

프로젝트 폴더 내에 celery.py 스크립트를 새로 생성한 뒤 위 내용을 적어주자.

위 내용은 장고 프로젝트의 세팅에서 celery세팅을 가져오고 이와 관련된 설정은 CELERY_로 시작한다는 뜻이다. namespace는 입맛대로 커스텀하면 된다.

app.autodiscover_tasks() 함수를 사용하면 각각의 앱 폴더에서 tasks.py 내부에 celery 데코레이터가 붙어있는 task를 모두 찾아서 적용할 수 있다.

#project_name/__init__.py

from .celery import app as celery_app

__all__ = ("celery_app",)

위 설정을 추가하면 @shared_task 데코레이터를 사용해도 task를 app에 불러올 수 있다. (tasks.py에서 celery app을 import 안 해도 된다는 뜻) 이번에는 @app.task를 사용해줄 것이므로 해도 되고 안해도 된다.

#project_app/views.py

from django_redis import get_redis_connection

client = get_redis_connection()

def get(self, request, *args, **kwargs):
    team_id = kwargs.get("team_id")
    if not request.user.id:
        user_id = 0
    else:
        user_id = request.user.id
    try:
        redis_ans = client.sadd(f"views:{team_id}", f"{user_id}_{request.META.get('REMOTE_ADDR')}")
        team = Team.objects.get(id=team_id)
        if redis_ans:
            team.view += 1
            team.save()
        teamSerializer = TeamCreateSerializer(team)
        response = createSerializerHelper.make_response(teamSerializer.data, request.user.id)

        return Response(response, status=status.HTTP_200_OK)

    except Team.DoesNotExist:
        return Response({"error": "No Content"}, status=status.HTTP_404_NOT_FOUND)

비회원의 조회에 대해서 카운트해주기 위해서 비회원의 유저id는 0으로 고정했다.

유저가 API에 접근했을 때 views:team_id에 유저id_IP쌍을 SADD 해준다. 이 때, 집합 추가에 성공한다면 이 명령은 1을 반환하고 실패하면 0을 반환해준다.

따라서 이 값을 분기점으로 조회수를 추가해준다.

#project_app/tasks.py NEW

from django_redis import get_redis_connection

from #project_name.celery import app

client = get_redis_connection()


@app.task
def delete_view_history():
    cursor = "0"
    print("--- 팀 조회 히스토리 삭제 시작 ---")
    while cursor != 0:
        cursor, keys = client.scan(cursor=cursor, match="views:*")
        if keys:
            client.delete(*keys)
    print("--- 팀 조회 히스토리 삭제 완료 ---")

celery에서 사용할 수 있는 task를 추가해주자. views:로 시작되는 키를 반복적으로 찾아내어 밸류값을 삭제하는 함수이다.

여기서 잠깐!

필자는 장고의 캐시백엔드로 redis를 이미 사용하고 있다. 그런데 왜 django_redis에서 redis_connection을 따로 빼내어서 연결 설정을 해줬을까?

이유는 간단하다. django_redis의 cache.py를 보면 SADD에 관한 내용이 없다.. set, get, delete 등 뿐이고 set이나 sorted_set에 대한 내용을 지원해주지 않는다.

task는 추가해줬으나 어떻게 자정마다 실행하게 할 수 있을까? 이를 가능케하는 것이 Celery-beat이다.

#project_name/celery.py

from celery.schedules import crontab

app.conf.beat_schedule = {
    "delete-view-history-per-day": {"task": "project_app.tasks.delete_view_history", "schedule": crontab(minute=0, hour=0)}
}

celery.py로 돌아가 위 내용을 추가해주자. crontab을 통해 특정 시간을 지정해 task를 실행시킬 수 있게 해준다.

실행

celery -A project_name worker -l INFO
celery -A project_name beat
python manage.py runserver

각각의 터미널에서 명령을 입력해 실행해주자.

celery worker user/tasks.send_email은 이메일 보내는 task를 따로 추가한 것이니 출력에 없어도 된다.

celery beat

잘 실행되었다면 위와 같이 출력되어 있을 것이다.

Postman으로 조회를 한 번 해본 뒤에 redis-cli에서 확인해보면 아래와 같이 views:team_id 키가 생성되어있다.

이게 자정에 삭제되는지 안 되는지 확인하기까지 시간이 많이 남았다면 celery.py의 crontab 설정을 바꿔볼 수 있다.

app.conf.beat_schedule = {
    "delete-view-history-per-day": {"task": "project_app.tasks.delete_view_history", "schedule": crontab()}
}

이렇게 바꿔주면 매 분마다 task가 실행된다.

잘 삭제가 된 것을 확인할 수 있다.

위 내용이 실행된 결과를 아까 따로 설치한 django-celery-results를 통해서도 알 수 있는데

두 번째 열에 해당하는 내용이 방금 우리가 해본 내용이다. celery를 실행한 쉘에서 None값이 반환되었으므로 django_celery_results_taskresult 테이블에는 null이 저장되는 것!

처음에는 클라이언트가 작성한 글을 저장하려고 할 때 장고에서 CharField로 생성한 필드에 저장하려고 했으나, CharField는 max_length를 반드시 명시해줘야 하기 때문에 글이 한도 길이를 지정하고 싶지 않았고, 글의 길이가 짧을 경우 패딩 문자가 저장되는 용량이 아까웠습니다.

그래서 어떻게 할 것인가?

클라이언트 측에서 보내준 글 내용을 Binary 값으로 바꾸어 데이터베이스의 Blob 필드에 저장하고자 합니다!

다르게 생각해본 방법은 파일로 만들어서 보관해볼까도 싶었지만, 서비스의 특성 상 읽어오는 작업이 많고 작성할 때의 작업이 너무 오래걸릴 것 같아 폐기했습니다.

간단할 줄 알았는데..

장고에서 모델을 설정할 때 CharField로 지정한 부분을 BinaryField로 바꾸고 저장하면 바로 끝나겠다 싶었습니다.

# Model 필드 BinaryField로 수정
explain = models.BinaryField()

수정한 뒤에 Postman으로 데이터를 보내 테스트를 해봤으나, serialzier의 is_valid()에서 계속 False가 떴습니다.

따로 알아본 결과 DRF의 Serializer에서는 BinaryField를 지원해주지 않는다! 라는 사실을 알게되었고, 어떻게 할까 생각하던 중 커스텀필드를 작성하자는 생각을 하게 되었습니다.

    def to_internal_value(self, data):
        """
        Transform the *incoming* primitive data into a native value.
        """
        raise NotImplementedError(
            '{cls}.to_internal_value() must be implemented for field '
            '{field_name}. If you do not need to support write operations '
            'you probably want to subclass `ReadOnlyField` instead.'.format(
                cls=self.__class__.__name__,
                field_name=self.field_name,
            )
        )

    def to_representation(self, value):
        """
        Transform the *outgoing* native value into primitive data.
        """
        raise NotImplementedError(
            '{cls}.to_representation() must be implemented for field {field_name}.'.format(
                cls=self.__class__.__name__,
                field_name=self.field_name,
            )
        )

위 내용은 rest_framwork.fields의 Field 클래스 내부 함수입니다. 반드시 두 함수는 오버라이딩 되어 재정의 되어야 필드로 쓸 수 있죠!

fields.py에서 미리 정의되어 있는 다른 필드들을 봐도 모두 오버라이딩 되어있습니다!

해결방법

to_internal_value 함수는 외부에서 데이터베이스로, to_representation 함수는 데이터베이스에서 외부로 값을 보낼 때 호출되는 함수들입니다.

우리는 클라이언트에게서 str값을 받아 binary형태로 바꾸어 저장할 것이므로 to_internal_value 함수에서는 encode, to_representation 함수에서는 decode를 해주면 되겠죠?

# 제가 새로 생성한 바이너리 필드입니다!
class MyBinaryField(serializers.Field):
    def to_representation(self, value):
        return value.decode("utf-8")

    def to_internal_value(self, data):
        return data.encode("utf-8")

이제 시리얼라이저에서 이 필드로 정의해주면 str값을 blob형태로 데이터베이스에 저장하고 blob 값을 str형태로 받아올 수 있습니다.

class TeamCreateSerializer(serializers.ModelSerializer):
    leader_id = serializers.IntegerField()
    explain = MyBinaryField()

    class Meta:
        model = Team
        fields = ["id", "leader_id", "name", "explain", "genre", "like", "version", "image"]

해결완료!

프론트와 로그인 기능 관련해서 API 연동 중에 도메인이 다름에 따라서 쿠키가 전달이 잘 안 됐다.

Chrome 기본적으로 크롬에서는 쿠키의 samesite 설정을 "Lax"로 해놓기 때문에 현 상황에서는 samesite를 None으로 지정해줄 필요가 있다!

첫 번째 코드

response.set_cookie(
                    "access", response.headers.get("access", None), httponly=True, samesite=None, secure=True
                )

빠르게 구글링을 통해 set_cookie 부분을 작성했다. 그런데 프론트 측에서 확인해본 결과 cookie의 samesite 속성이 제대로 지정되지 않았다. 크롬에서 기본으로 세팅되는 Lax도 안 되어 있었다는 뜻!

왜 그랬을까요?

이유를 알아보기 위해서 라이브러리를 파고 들어가봅시다.

def set_cookie(
        self,
        key,
        value="",
        max_age=None,
        expires=None,
        path="/",
        domain=None,
        secure=False,
        httponly=False,
        samesite=None,
    ):
        """
        Set a cookie.

        ``expires`` can be:
        - a string in the correct format,
        - a naive ``datetime.datetime`` object in UTC,
        - an aware ``datetime.datetime`` object in any time zone.
        If it is a ``datetime.datetime`` object then calculate ``max_age``.

        ``max_age`` can be:
        - int/float specifying seconds,
        - ``datetime.timedelta`` object.
        """
        self.cookies[key] = value
        if expires is not None:
            if isinstance(expires, datetime.datetime):
                if timezone.is_naive(expires):
                    expires = timezone.make_aware(expires, datetime.timezone.utc)
                delta = expires - datetime.datetime.now(tz=datetime.timezone.utc)
                # Add one second so the date matches exactly (a fraction of
                # time gets lost between converting to a timedelta and
                # then the date string).
                delta += datetime.timedelta(seconds=1)
                # Just set max_age - the max_age logic will set expires.
                expires = None
                if max_age is not None:
                    raise ValueError("'expires' and 'max_age' can't be used together.")
                max_age = max(0, delta.days * 86400 + delta.seconds)
            else:
                self.cookies[key]["expires"] = expires
        else:
            self.cookies[key]["expires"] = ""
        if max_age is not None:
            if isinstance(max_age, datetime.timedelta):
                max_age = max_age.total_seconds()
            self.cookies[key]["max-age"] = int(max_age)
            # IE requires expires, so set it if hasn't been already.
            if not expires:
                self.cookies[key]["expires"] = http_date(time.time() + max_age)
        if path is not None:
            self.cookies[key]["path"] = path
        if domain is not None:
            self.cookies[key]["domain"] = domain
        if secure:
            self.cookies[key]["secure"] = True
        if httponly:
            self.cookies[key]["httponly"] = True
        if samesite:
            if samesite.lower() not in ("lax", "none", "strict"):
                raise ValueError('samesite must be "lax", "none", or "strict".')
            self.cookies[key]["samesite"] = samesite

위 코드는 장고의 HttpResponseBase 클래스의 set_cookie 함수 전문입니다.

인자를 받는 부분에서부터 samesite=None이 default 값이네요. 벌써 쎄하죠?

아래쪽으로 내려와서 samesite에 관련된 부분을 다시 한 번 봅시다.

if samesite:
            if samesite.lower() not in ("lax", "none", "strict"):
                raise ValueError('samesite must be "lax", "none", or "strict".')
            self.cookies[key]["samesite"] = samesite

아하, samesite 속성을 지정해줄 때 None 값으로 넣어서 if문 자체가 통과가 안 됐었네요..

내용을 참고해보니 none을 문자열로 넣어야 했습니다!

참고 (Django Docs)

여기서도 set_cookie에서 samesite를 어떻게 지정해야하는지 나오네요. 역시 공식 독스..

수정 후 코드

response.set_cookie(
                    "access", response.headers.get("access", None), httponly=True, samesite="none", secure=True
                )

간단한 수정으로 문제 해결 완료!

역시 가장 확실한 방법은 공식 docs를 확인하거나 직접 라이브러리를 까보는 것.. 알량한 구글링이 아니라 내가 직접 체득하는게 최고다!

FastAPI를 사용하다가 장고도 한 번 써보고 싶은 마음에 무작정 돌입하게 됐는데, 건드는게 생각보다 쉽지 않아 여기저기 공부하면서 하고 있네요.. 장고가 처음이다보니 부족한 지식으로 잘못 기술되거나 보셨을 때 개선의 여지가 있다면 코멘트 부탁드리겠습니다! (__)

이번 프로젝트에서 적용한 인증/인가 방식을 한 번 소개해볼까 합니다. 글에서는 단순히 로직만 소개하는 것이 아니라 나름대로 구현할 때 살펴봤던 점도 함께 기술해보겠습니다.

흐름도

전체적인 흐름도입니다. 위에서는 FrontEnd만 표기되어 있지만, 사실 App과 통신할 때 Authorization 헤더를 사용하고 웹에서 프론트와 통신할 때는 쿠키를 사용했습니다. 글 읽으시다가 참고하기 좋으시라고 미리 올려두겠습니다.

이번에 앱과 웹 동시에 통신하는 백엔드를 처음 개발해보는데 이게 통상적인 방법인지는 모르겠습니다만 커스텀 헤더를 추가해서 앱에서 온 요청인지 웹에서 온 요청인지 구분하고 있습니다. (api 따로 추가 X)

커스텀 미들웨어

App에서 통신할 때는 굳이 쿠키를 이용하지 않고 JWT를 주고 받는 방식, Web에서 통신할 때는 쿠키에 JWT를 넣어 주고 받는 방식을 구현하고자 했는데요. 이를 구현하기 위해서 미들웨어를 따로 추가해줬습니다.

from django.utils.deprecation import MiddlewareMixin
from rest_framework import status


class AttachJWTFromHeaderToCookieMiddleware(MiddlewareMixin):
    def __init__(self, get_response):
        super().__init__(get_response)
        self.API = ["github/login", "callback/github"]
        self.REFRESH = "token/refresh"

    def process_response(self, request, response):
        path = request.path_info
        is_valid = any(api in path for api in self.API)
        is_refresh = True if self.REFRESH in path else False
        if (
            is_valid
            or is_refresh
            and (response.status_code == status.HTTP_200_OK or response.status_code == status.HTTP_201_CREATED)
        ):
            if request.META.get("HTTP_X_FROM", None) == "web":
                response.set_cookie("access", response.headers.get("access", None), httponly=True)
                if is_valid:
                    del response.headers["access"]

                response.content = response.render().rendered_content

        return response


class AttachJWTFromCookieToHeaderMiddleware(MiddlewareMixin):
    def __init__(self, get_response):
        super().__init__(get_response)
        self.NOT_API = ["github/login", "callback/github"]

    def process_request(self, request):
        path = request.path_info
        is_valid = any(api in path for api in self.NOT_API)

        if not is_valid:
            if request.META.get("HTTP_X_FROM", None) == "web":
                request.META["HTTP_AUTHORIZATION"] = f"Bearer {request.COOKIES.get('access', None)}"

AttachJWTFromHeaderToCookieMiddleware

말 그대로 헤더에서 쿠키로 JWT를 붙여주는 미들웨어입니다. 서버에서 클라이언트 측으로 JWT를 보낼 때는 현재 회원가입/로그인, 토큰 리프레싱 API 밖에 없기 때문에 요청 Path에 위 태스크의 API 주소가 들어있다면 실행되도록 커스텀했습니다.

웹에서 온 요청일 때만 리스폰스의 헤더에서 쿠키로 JWT를 옮겨줍니다.

AttachJWTFromCookieToHeaderMiddleware

위의 미들웨어와 정반대의 일을 하는 미들웨어입니다. 클라이언트 쪽에서 JWT가 들어오는 요청은 회원가입/로그인 요청을 제외한 모든 요청입니다.

웹에서 온 요청일 때만 리퀘스트의 쿠키에서 헤더로 JWT를 옮겨줍니다.

왜? 굳이 쿠키에서 헤더로 다시 보냈나요?

라는 생각을 하신 분들이 계실 수도 있는데요.

DRF에서 JWT를 사용하면서 사용자를 식별하기 위해

rest_framework_simplejwt.authentication.JWTAuthentication

를 사용합니다. 저도 이 인증방식을 기본으로 채택해놨구요. 이 JWTAuthentication을 조금 살펴봅시다.

def authenticate(self, request: Request) -> Optional[Tuple[AuthUser, Token]]:
        header = self.get_header(request)

        if header is None:
            return None

        raw_token = self.get_raw_token(header)
        if raw_token is None:
            return None

        validated_token = self.get_validated_token(raw_token)

        return self.get_user(validated_token), validated_token

def get_header(self, request: Request) -> bytes:
        """
        Extracts the header containing the JSON web token from the given
        request.
        """
        header = request.META.get(api_settings.AUTH_HEADER_NAME)
        if isinstance(header, str):
            # Work around django test client oddness
            header = header.encode(HTTP_HEADER_ENCODING)

JWTAuthentication의 일부 함수들인데요. 음.. 쿠키로 인증하는 내용은 없고 읽어보니 기본 인증 헤더(Authorization)로부터 JWT를 가져오네요. 물론 JWTAuthentication을 상속받아 쿠키에서도 토큰을 가져오도록 커스텀 할 수도 있겠습니다만은, 저는 미들웨어에서 그냥 헤더로 넣어버리는 방법을 선택해봤습니다.

JWTAuthentication를 통과하고나면 request.user에 유저 객체 또는 AnonymousUser가 들어갑니다. 이를 통해 APIView에서 permission_classes로 권한 설정을 쉽게 할 수 있었어요!

    def get_validated_token(self, raw_token: bytes) -> Token:
        """
        Validates an encoded JSON web token and returns a validated token
        wrapper object.
        """
        messages = []
        for AuthToken in api_settings.AUTH_TOKEN_CLASSES:
            try:
                return AuthToken(raw_token)
            except TokenError as e:
                messages.append(
                    {
                        "token_class": AuthToken.__name__,
                        "token_type": AuthToken.token_type,
                        "message": e.args[0],
                    }
                )

        raise InvalidToken(
            {
                "detail": _("Given token not valid for any token type"),
                "messages": messages,
            }
        )

authenticate 함수 내에서 실행되는 토큰 확인 함수입니다. 여기서 정상적인 토큰인지, 토큰이 만료되었는지 확인하게 됩니다. simple_jwt 라이브러리에서 정의된 AuthToken 클래스에서 확인합니다. 여기서 위 흐름도의 401을 반환하는 경우의 수를 충족하겠죠? get_user 함수를 통해 유저 객체를 반환해줍니다.

드디어 View로..

위에서 미리 설명한대로 APIView 속에서 유저의 권한을 쉽게 확인할 수 있습니다. 아래에 View를 두 개 올려둘게요.

class UserManageView(APIView):
    serializer_class = ApproveUserSerializer
    permission_classes = [IsAdminUser]

    def get(self, request):
        queryset = User.objects.filter(is_approved=False)
        if queryset:
            serializer = self.serializer_class(queryset, many=True)
            return Response(serializer.data)
        else:
            return Response({"error": "No Content"}, status=status.HTTP_404_NOT_FOUND)

    def patch(self, request, *args, **kwargs):
        user_id = kwargs.get("user_id")
        user = User.objects.get(id=user_id)

        serializer = ApproveUserSerializer(user, data={"is_approved": True}, partial=True)
        if serializer.is_valid():
            serializer.save()
            return Response({"success": True}, status=status.HTTP_202_ACCEPTED)

        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

class GithubOAuthCallBackView(APIView):
    permission_classes = [AllowAny]

    def get(self, request: Request):
        if code := request.GET.get("code"):
            response = self.send_code_to_github_login_view(code)
            if response.status_code == 200:
                return Response(response.json(), status=status.HTTP_200_OK)
            return Response(
                {"error": "Failed to process with GithubLoginView"},
                status=response.status_code,
            )

    def send_code_to_github_login_view(self, code: str):
        url = "http://localhost:8000/api/auth/github/login"
        payload = {"code": code}
        headers = {"Content-Type": "application/json"}
        response = requests.post(url, json=payload, headers=headers)

        return response

두 뷰가 하는 일에 대해서는 우선 제쳐두고, permission_classes 부분을 보도록 합시다.

각각 AllowAny, IsAdminUser가 있죠?

APIView에서 메소드들에 접근하기 전에

def initial(self, request, *args, **kwargs):
        """
        Runs anything that needs to occur prior to calling the method handler.
        """
        self.format_kwarg = self.get_format_suffix(**kwargs)

        # Perform content negotiation and store the accepted info on the request
        neg = self.perform_content_negotiation(request)
        request.accepted_renderer, request.accepted_media_type = neg

        # Determine the API version, if versioning is in use.
        version, scheme = self.determine_version(request, *args, **kwargs)
        request.version, request.versioning_scheme = version, scheme

        # Ensure that the incoming request is permitted
        self.perform_authentication(request)
        self.check_permissions(request)
        self.check_throttles(request)

이 곳을 먼저 지나갑니다. 저희가 이번에 볼 곳은 저 check_permissions인데요. 저희가 설정해놓은 permission_classes를 검증해보면서 권한을 확인해봅니다.

각각의 permission_classes 내부에 오버라이딩 되어 있는 has_permission 메소드를 통해 True 또는 False를 반환하게 되어 있어요.

class IsAdminUser(BasePermission):
    """
    Allows access only to admin users.
    """

    def has_permission(self, request, view):
        return bool(request.user and request.user.is_staff)

위에 설정되어 있는 IsAdminUser의 모습입니다. JWTAuthentication을 통해 식별된 유저가 request.user에 들어있을 거에요. 식별된 유저의 is_staff 컬럼이 True, False인지에 따라서 권한이 정해지는 방식입니다!

이 형식을 따라서 우리도 쉽게 권한 설정을 커스터마이징 할 수 있습니다.

# 이건 제가 사용하고 있는 커스텀 권한입니다!
from rest_framework.permissions import BasePermission


class IsApprovedUser(BasePermission):
    def has_permission(self, request, view):
        return bool(request.user and request.user.is_approved)

has_permission 함수에서 False를 반환하면, 403 상태 코드를 반환하도록 설계되어 있습니다!

토큰 리프레싱

401 상태 코드를 처음 리턴 받았다면, 클라이언트 쪽에서는 토큰 리프레싱 해주는 API로 다시 통신을 시도하도록 했습니다.

제가 사용한 simple_jwt 라이브러리에서는 제공해주는 뷰를 통해 이를 쉽게 할 수 있는데요.

class WtntTokenRefreshView(TokenRefreshView):
    def post(self, request: Request, *args, **kwargs):
        _, access_token = request.META.get("HTTP_AUTHORIZATION").split(" ")
        user_id = AccessToken(access_token, verify=False).payload.get("user_id")

        refresh_token = cache.get(user_id)
        if not refresh_token:
            return Response({"error": "Expired Refresh Token"}, status=status.HTTP_401_UNAUTHORIZED)
        serializer = self.get_serializer(data={"refresh": refresh_token})

        try:
            serializer.is_valid(raise_exception=True)
        except TokenError as e:
            raise InvalidToken(e.args[0])

        token = serializer.validated_data
        response = Response({"success": True}, status=status.HTTP_200_OK)

        response.headers["access"] = token["access"]

        return response

원래 이 뷰에서는, 기본으로 지정되어 있는 TokenRefreshSerializer에 RefreshToken을 데이터로 넣어주기만 하면 토큰을 리프레싱 해줍니다. 하지만 이번 프로젝트를 진행하면서 RefreshToken은 클라이언트 측에 전해주지 않고 서버 측에서 관리하기로 했는데요. 따라서, AccessToken의 페이로드에 들어있는 유저 정보를 통해 RefreshToken을 식별해서 리프레싱해줘야 합니다.

simple-jwt에서는 AccessToken을 디코드할 때 만료기한이 지났을 경우 바로 401 상태 코드를 반환하게 되어 있습니다.

유저가 보낸 AccessToken은 이미 만료되어 있기 때문에 이 뷰에 접근을 요청했잖아요!!

네, 그래서 verify=False 옵션을 반드시 명시해서 만료된 AccessToken에서도 유저 정보를 식별할 수 있어야합니다.

RefreshToken은 어디에서 받아 오나요?

AccessToken의 만료기한을 30분으로 정해놨는데, RefreshToken을 현재 사용중인 MySQL의 테이블에 넣어놓고 관리하면 자주 I/O가 생길 것 같았고 RefreshToken의 만료기한이 지났을 때 따로 삭제해줘야 했습니다.(스케쥴러 등을 통해)

그래서 Redis에 토큰을 저장하고, expire로 만료기한을 지정하는 방법을 사용해보기로 했습니다.

AccessToken의 payload에서 가져온 유저의 식별정보로 RefreshToken을 DB에서 받아옵니다.

RefreshToken이 존재한다면 시리얼라이저를 통해 새로운 AccessToken을 발급해주고, 존재하지 않는다면 두 번째 401 상태 코드를 반환합니다.

이렇게 인증/인가 로직을 구현해봤습니다. 제 부족한 점이나 궁금하신 점이 있으시다면 편하게 의견 남겨주세요!!

1. 백엔드..?

학부생 수준에서 AI를 조금 더 깊이 알아봤다고 할 정도의 얕은 지식과 더불어 백엔드 웹프레임워크에 대한 지식은 전무하다싶은 상황에서 처음으로 진행해본 프로젝트였다. BE 부분을 맡아서 해야한다는 사실이 내게는 트러블 그 자체였다. 다행히 FastAPI는 공식문서가 정리가 잘 되어있는 편이었고 파이썬 웹백엔드에 대한 강의를 따로 수강하면서 프로젝트를 진행했더니 어떻게든 완성한 것 같다.

2. AI 완성도

자료 양이 방대해서 그런지 예상했던 것보다 더 번역이 잘 됐다 ㅋㅋ 데이터셋에 이런 문구가 존재했구나..? 싶었던 단어와 어미가 바뀌는 걸 보고 신기했다. 내가 해도 이정돈데 정말 전문적으로 공부한 사람들이 각 잡고 만들면 어떤 결과가 나올지 궁금하긴 하다.

Ex) 그런건 여기 천지삐까리다! -> 그런건 여기에 많이 있다!

3. 결과물 스샷

로그인창

로그인 후 번역기 창

번역 및 검색 기록

방명록

4. 아쉬웠던 점

• STT 기술도 활용해서 한국어 발화 데이터를 통해 사투리를 음성으로 입력 받아서 번역하는 기능까지 해보고 싶었는데 그 부분은 음성 데이터에 대한 지식 부족과 컴퓨팅 자원 부족으로 시도해보지 못 했다. 언젠가 시간과 자원이 갖춰진다면 시도해보고 싶다.

• 유저의 검색 기록 기반으로 데이터를 정리해서 추가적으로 모델을 계속 학습하는 파이프라인을 구축해보고 싶었으나 유저가 승인한 정보를 학습했을 때 AI에 긍정적으로 영향을 미칠지 확신하지 못해서 일단은 보류했다. (그저 검색 기록만으로 데이터를 구축하는 건 어려운 일이 아니다. 정보를 어떻게 정제할지 가이드라인을 정하지 못한 것..)

• 백엔드의 완성도가 꽤 부족한 것 같다. Github의 V1 부분을 보면 코드가 좀 많이 더럽다. 신경 쓰지 못한 부분과 몰랐던 부분이 많아서 가독성이 엄청 떨어진다. (그래서 V2로 리팩토링을 한 번 진행해봤다. 고수분들이 보시기엔 그래도 매~우 많이 부족할 것..)

5. 결과!

(다른 사람들 이름도 있어서 한 번에 다 가렸다..)

교수님들이 좋게 봐주셔서 다행히 단과대 및 학부 경진대회에서 모두 입상할 수 있었다!

환경 설정 변경

테스트는 서비스에 활용되는 DB가 아닌 TEST용 DB를 사용하는 것이 좋다.

따라서, 테스트와 서비스 상황을 구분해서 환경 설정을 따로 해주도록 하자.

config.py

환경설정을 좀 더 깔끔하게 해보고 싶어서 pydantic의 BaseSettings로 바꿔봤다. 이를 사용하기 위해서는 라이브러리를 하나 더 설치해줘야 한다.

pip install pydantic-settings

from pydantic_settings import BaseSettings
from typing import ClassVar
from functools import lru_cache


class BaseConfig(BaseSettings):
    # DB
    DB_USERNAME: str
    DB_HOST: str
    DB_PASSWORD: str
    DB_PORT: int

    # CRED
    SECRET_KEY: str
    ALGORITHM: str

    # PAPAGO
    CLIENT_ID: str
    CLIENT_SECRET: str

    # Test Config
    TESTING: ClassVar[bool] = False

    class Config:
        extra = "ignore"
        env_file = ".env"
        env_file_encoding = "utf-8"


class testSettings(BaseConfig):
    TESTING: ClassVar[bool] = True
    DB_TEST_NAME: str


class Settings(BaseConfig):
    TESTING: ClassVar[bool] = False
    DB_NAME: str


class isMain(BaseSettings):
    IS_MAIN: bool

    class Config:
        extra = "ignore"
        env_file = ".env"
        env_file_encoding = "utf-8"


@lru_cache
def get_settings():
    if is_main.IS_MAIN:
        return Settings()
    else:
        return testSettings()


is_main = isMain()
settings = get_settings()

• 테스트 환경에서 사용할 testSettings, 서비스 환경에서 사용할 Settings로 나눴다.

• .env 파일에 IS_MAIN이라는 환경변수를 추가해서 IS_MAIN이 true인 경우에만 Settings를 사용한다.

• class Config에서 extra = "ignore"를 설정해주지 않으면 에러가 나니 꼭 추가해주도록 하자.

• 만약 테스트와 서비스 환경이 데이터베이스 이름뿐 아니라 다른 것도 상이하다면 묶어둔 BaseConfig에서 분리해서 작성해줘야 한다.

Pytest

pip install pytest-asyncio pytest-cov

Pytest 라이브러리는 말 그대로 Python 코드를 테스트해보기 위한 라이브러리이다. 우리는 비동기식으로 테스트 코드를 실행하고 html로 정리된 테스트 결과를 보기 위해서 위의 두 라이브러리를 설치하도록 하겠다.

Pytest를 설치하고 나면 pytest 명령어를 콘솔에서 사용할 수 있는데, 명령어를 실행했을 때 test_*.py를 수집해서 안의 코드들을 테스트하게 된다.

Test

FASTAPI에서 TestClient를 제공해주지만, 필자가 알고 있는 지식(?)으로는 비동기 방식으로 지원해주지 않기 때문에 httpx로 AsyncClient를 생성해서 비동기 방식으로 테스트를 진행하겠다.

Test 코드

conftest.py

import bcrypt
import pytest_asyncio
import json
from httpx import AsyncClient

from app import main, models
from app.database import engine, get_db
from app.config import settings


@pytest_asyncio.fixture(scope="session")
def app():
    if not settings.TESTING:
        raise SystemError("TESTING environment must be set true")

    return main.app

@pytest_asyncio.fixture
async def session():
    db = next(get_db())
    try:
        yield db
    finally:
        db.close()


@pytest_asyncio.fixture
async def client(app):
    async with AsyncClient(app=app, base_url="http://test/v2") as ac:
        models.Base.metadata.drop_all(bind=engine)
        models.Base.metadata.create_all(bind=engine)

        yield ac

• FIXTURE란 Pytest에서 반복적으로 생성되어야 하는 아이템을 매개변수로 불러와 생성할 수 있게 해주는 일종의 모듈이라고 볼 수 있다.

• FIXTURE에서 FIXTURE를 불러와서 다른 FIXTURE를 만들 수도 있다.

• 위 세 개의 FIXTURE로 만들어놓은 API와 데이터베이스를 연결하고, client를 통해서 API에 접근할 수 있게 작성하였다.

test_user.py

conftest.py에 유저를 DB에 등록하는 FIXTURE를 생성하자.

@pytest_asyncio.fixture
def user(session) -> models.User:
    salt_value = bcrypt.gensalt()
    pw = bcrypt.hashpw("testpw".encode(), salt_value)
    row = models.User(password=pw, email="test@sample.com")
    session.add(row)
    session.commit()
    return row

import pytest
import json


# 유저 생성 테스트
@pytest.mark.asyncio
async def test_add_user(client):
    body = {"email": "sample@sample.com", "password": "samplepw"}
    r = await client.post("/users/register", data=json.dumps(body))
    data = r.json()

    assert r.status_code == 201
    assert data.get("email") == body["email"]

# 유저 생성 에러 테스트 (너무 긴 이메일)
@pytest.mark.asyncio
async def test_add_user_failed_by_long_email(client):
    body = {"email": "longlonglonglonglonglonglonglonglongmail@sample.com", "password": "samplepw"}
    r = await client.post("/users/register", data=json.dumps(body))

    assert r.status_code == 422

# 유저 이메일 중복 확인 에러 테스트 (중복 이메일)
@pytest.mark.asyncio
async def test_email_duplicated_exist(client, user):
    email = "test@sample.com"
    r = await client.get(f"/users/duplicated?email={email}")
    data = r.json()

    assert r.status_code == 200
    assert data.get("duplicated")
    assert email == user.email

# 유저 이메일 중복 확인 테스트
@pytest.mark.asyncio
async def test_email_duplicated_no_exist(client):
    email = "noexist@sample.com"
    r = await client.get(f"/users/duplicated?email={email}")
    data = r.json()

    assert r.status_code == 200
    assert not data.get("duplicated")

# 유저 로그인 테스트
@pytest.mark.asyncio
async def test_login_user(client, user):
    body = {"email": user.email, "password": "testpw"}
    r = await client.post("/users/login", data=json.dumps(body))
    data = r.json()

    assert r.status_code == 200
    assert data.get("user_id") == user.id

# 유저 로그인 에러 테스트 (잘못된 패스워드)
@pytest.mark.asyncio
async def test_login_user_invalid_pw(client, user):
    body = {"email": user.email, "password": "wrongpw"}
    r = await client.post("/users/login", data=json.dumps(body))
    data = r.json()

    assert r.status_code == 403
    assert data.get("detail") == "Incorrect Password"

# 유저 로그인 에러 테스트 (잘못된 이메일)
@pytest.mark.asyncio
async def test_login_user_invalid_email(client):
    body = {"email": "noemail", "password": "testpw"}
    r = await client.post("/users/login", data=json.dumps(body))
    data = r.json()

    assert r.status_code == 403
    assert data.get("detail") == "No Email Found"

함수 두 개를 가져와서 찬찬히 살펴보도록 하자.

# 유저 이메일 중복 확인 테스트
@pytest.mark.asyncio
async def test_email_duplicated_no_exist(client):
    email = "noexist@sample.com"
    r = await client.get(f"/users/duplicated?email={email}")
    data = r.json()

    assert r.status_code == 200
    assert not data.get("duplicated")

# 유저 이메일 중복 확인 에러 테스트 (중복 이메일)
@pytest.mark.asyncio
async def test_email_duplicated_exist(client, user):
    email = "test@sample.com"
    r = await client.get(f"/users/duplicated?email={email}")
    data = r.json()

    assert r.status_code == 200
    assert data.get("duplicated")
    assert email == user.email

1. 유저 이메일 중복 확인 테스트 (에러X) 이 코드는 이메일이 중복인지 아닌지 확인하는 엔드포인트의 테스트 코드다. Fixture를 client만 받아왔으므로 User 테이블에는 어떠한 데이터도 들어있지 않을 것이므로, 어떠한 email을 검사하더라도 통과할 것이다. 따라서, status_code는 성공이므로 200, 반환값은 duplicated가 아니어야 한다. 이를 assert 함수를 통해서 검증하는 것.

2. 유저 이메일 중복 확인 테스트 (중복 이메일) 여기서는 user Fixture도 받아오게 되는데, 이러면 User 테이블에 Fixture에서 생성한 유저 데이터가 삽입되어 있는 상태이다. Fixture에서 생성한 유저의 이메일이 "test@sample.com"이므로 이와 같은 이메일로 엔드포인트에 접근해서 일부러 실패하는 것!

test_translate.py

우리는 번역 작업을 로그인 시에만 가능하게 설정해놨으므로, 토큰 값도 같이 보내줘야한다. 따라서 conftest.py에 token을 반환해주는 FIXTURE를 추가하도록 하자.

@pytest_asyncio.fixture
async def token(client, user) -> str:
    body = {"email": user.email, "password": "testpw"}
    r = await client.post("/users/login", data=json.dumps(body))
    return r.headers.get("access_token")

import pytest
import json


@pytest.mark.asyncio
async def test_translate(client, token):
    body = {"dialect": "밥 뭇나?"}
    headers = {"Authorization": f"Bearer {token}"}
    r = await client.post("/AI", data=json.dumps(body), headers=headers)
    data = r.json()

    assert r.status_code == 200
    assert data.get("dialect") == body["dialect"]

test_item.py

생성된 TsItem을 삭제하는 엔드포인트를 테스트해보기 위해서 데이터베이스에 Item을 삽입하는 FIXTURE와 다른 토큰을 생성하는 user2, token2 FIXTURE를 만들어주자.

@pytest_asyncio.fixture
def item(session, user) -> models.TsItem:
    row = models.TsItem(
        dialect="dialect",
        standard="standard",
        english="english",
        chinese="chinese",
        japanese="japanese",
        owner_id=user.id,
    )
    session.add(row)
    session.commit()
    return row

@pytest_asyncio.fixture
def user2(session) -> models.User:
    salt_value = bcrypt.gensalt()
    pw = bcrypt.hashpw("testpw".encode(), salt_value)
    row = models.User(password=pw, email="test2@sample.com")
    session.add(row)
    session.commit()
    return row

@pytest_asyncio.fixture
async def token2(client, user2) -> str:
    body = {"email": user2.email, "password": "testpw"}
    r = await client.post("/users/login", data=json.dumps(body))
    return r.headers.get("access_token")

import pytest
import json


@pytest.mark.asyncio
async def test_add_item(client, token):
    body = {
        "dialect": "dialect",
        "standard": "standard",
        "english": "english",
        "chinese": "chinese",
        "japanese": "japanese",
    }
    headers = {"Authorization": f"Bearer {token}"}

    r = await client.post("/items", data=json.dumps(body), headers=headers)
    data = r.json()

    assert r.status_code == 201
    assert data.get("dialect") == body["dialect"]
    assert data.get("standard") == body["standard"]
    assert data.get("english") == body["english"]
    assert data.get("chinese") == body["chinese"]
    assert data.get("japanese") == body["japanese"]


@pytest.mark.asyncio
async def test_add_item_failed_by_long_text(client, token):
    body = {
        "dialect": "a" * 256,
        "standard": "standard",
        "english": "english",
        "chinese": "chinese",
        "japanese": "japanese",
    }
    headers = {"Authorization": f"Bearer {token}"}

    r = await client.post("/items", data=json.dumps(body), headers=headers)

    assert r.status_code == 422


@pytest.mark.asyncio
async def test_get_items(client, item):
    r = await client.get("/items")
    data = r.json()

    assert r.status_code == 200
    assert data[0].get("id") == item.id


@pytest.mark.asyncio
async def test_delete_item(client, item, token):
    headers = {"Authorization": f"Bearer {token}"}

    r = await client.delete(f"/items/{item.id}", headers=headers)

    assert r.status_code == 202


@pytest.mark.asyncio
async def test_delete_item_failed_by_invaild_id(client, item, token):
    headers = {"Authorization": f"Bearer {token}"}

    r = await client.delete(f"/items/{item.id+1}", headers=headers)
    data = r.json()

    assert r.status_code == 400
    assert data.get("detail") == "No Item Found"


@pytest.mark.asyncio
async def test_delete_item_failed_by_wrong_user(client, item, token2):
    headers = {"Authorization": f"Bearer {token2}"}

    r = await client.delete(f"/items/{item.id}", headers=headers)
    data = r.json()

    assert r.status_code == 403
    assert data.get("detail") == "It's not owner"

test_guestbook.py

방명록을 생성하는 Fixture도 추가해주자.

@pytest_asyncio.fixture
def guestbook(session, user) -> models.GuestBook:
    row = models.GuestBook(message="test", message_owner="test_owner", owner_id=user.id)
    session.add(row)
    session.commit()
    return row

import pytest
import json


@pytest.mark.asyncio
async def test_add_guestbook(client, token):
    body = {"message": "test", "message_owner": "testowner"}
    headers = {"Authorization": f"Bearer {token}"}

    r = await client.post("/guestbooks", data=json.dumps(body), headers=headers)
    data = r.json()

    assert r.status_code == 201
    assert data.get("message") == body["message"]


@pytest.mark.asyncio
async def test_add_guestbook_failed_by_long_owner(client, token):
    body = {"message": "test", "message_owner": "a" * 25}
    headers = {"Authorization": f"Bearer {token}"}

    r = await client.post("/guestbooks", data=json.dumps(body), headers=headers)

    assert r.status_code == 422


@pytest.mark.asyncio
async def test_get_guestbooks(client, guestbook):
    r = await client.get("/guestbooks")
    data = r.json()

    assert r.status_code == 200
    assert data[0].get("id") == guestbook.id


@pytest.mark.asyncio
async def test_update_guestbook(client, token, guestbook):
    body = {"message": "modified_message"}
    headers = {"Authorization": f"Bearer {token}"}

    r = await client.put(f"/guestbooks/{guestbook.id}", data=json.dumps(body), headers=headers)
    data = r.json()

    assert r.status_code == 202
    assert data.get("message") == body["message"]


@pytest.mark.asyncio
async def test_update_guestbook_failed_by_invalid_id(client, token, guestbook):
    body = {"message": "modified_message"}
    headers = {"Authorization": f"Bearer {token}"}

    r = await client.put(f"/guestbooks/{guestbook.id+1}", data=json.dumps(body), headers=headers)
    data = r.json()

    assert r.status_code == 400
    assert data.get("detail") == "Guestbook Not Found"


@pytest.mark.asyncio
async def test_update_guestbook_failed_by_wrong_user(client, token2, guestbook):
    body = {"message": "modified_message"}
    headers = {"Authorization": f"Bearer {token2}"}

    r = await client.put(f"/guestbooks/{guestbook.id}", data=json.dumps(body), headers=headers)
    data = r.json()

    assert r.status_code == 403
    assert data.get("detail") == "It's not owner"

Test 결과

Test 시작에 앞서 -

최상위폴더
ㄴ .coveragerc
ㄴ pytet.ini

테스트에서 제외하고 싶은 파일이 있다면?

# .coveragerc

[run]
omit =
    app/config.py

이곳에 스크립트 파일을 추가해주면 테스트에서 제외된다!

테스트 설정

# pytest.ini

[pytest]
addopts = --cov=app --cov-report term --cov-report html

1. --cov=app 테스트를 진행할 범위를 지정하는 곳이다.

2. --cov-report term pytest를 실행하는 터미널에서 결과를 출력하게 해주는 옵션이다.

3. --cov-report html 테스트를 완료한 뒤 결과보고서를 html 형식으로 저장하겠다는 옵션이다. 이 html 보고서는 htmlcov 폴더에 저장되어 있다.

Test 실행

가상환경을 activate하고 pytest -sv를 입력해주자. 여기서 -sv 옵션은 테스트 진행 중 더 상세한 정보를 표기해준다.

Terminal

• 터미널에서 테스트 결과를 띄워주는 모습이다. 커버리지를 최대한 높여보고자 테스트 코드를 작성했기 때문에 꽤나 퍼센테이지가 높은 것을 볼 수 있다 ㅎㅎ.

HtmlCov

htmlcov 폴더의 index.html로 들어가보면 이런 화면이 나온다. 확실히 터미널보다 한 눈에 들어오는 것을 볼 수 있다.

• 그리고 터미널과 다르게 Html로 커버리지 레포트를 받으면 좋은 점이 첫 index 화면에서 스크립트 이름을 클릭하면 어떤 코드가 커버되었고 어떤 코드가 안 되었는지 시각적으로 쉽게 확인할 수 있다.

여기까지 해서 캡스톤 디자인 프로젝트 테스트 코드까지 모두 작성해보았다. 다음 포스트에서 이번 프로젝트를 회고해보면서 시리즈를 마치도록 하겠다.

Database Table 생성

APP/
ㄴ models.py --> 작업할 공간

models.py

class GuestBook(BaseMin, Base):
    __tablename__ = "guestbook"

    message = Column(String(255), nullable=False)
    message_owner = Column(String(20), nullable=False)
    owner_id = Column(Integer, ForeignKey("user.id"))

    book_owner = relationship("User", back_populates="guestbooks")

1. message: 방명록에 남길 메세지
2. message_owner: 방명록에 남길 이름
3. owner_id: (외래키) 추가한 회원의 식별 ID

기능

controller.py

from fastapi import Depends, APIRouter, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm.session import Session
from typing import List

from app.database import get_db
from .schema import GuestBookAdd, GuestBookReturn, GuestBookUpdate
from .service import addGuestBook, findAllGuestBook, updateGuestBook

router = APIRouter()
security = HTTPBearer()

# 방명록 작성
@router.post("", response_model=GuestBookReturn, status_code=status.HTTP_201_CREATED)
async def guestbook_add(
    data: GuestBookAdd,
    cred: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db),
):
    return await addGuestBook(data, cred, db)

# 모든 방명록 조회
@router.get("", response_model=List[GuestBookReturn], status_code=status.HTTP_200_OK)
async def guestbook(db: Session = Depends(get_db)):
    return await findAllGuestBook(db)

# 방명록 수정
@router.put("/{id}", response_model=GuestBookReturn, status_code=status.HTTP_202_ACCEPTED)
async def guestbook_update(
    data: GuestBookUpdate,
    id: int,
    cred: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db),
):
    return await updateGuestBook(data, id, cred, db)

1. 방명록 작성

@router.post("", response_model=GuestBookReturn, status_code=status.HTTP_201_CREATED)
async def guestbook_add(
    data: GuestBookAdd,
    cred: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db),
):
    return await addGuestBook(data, cred, db)

1. GuestBookAdd 형태에 맞추어 Body 정보를 받는다.
2. 로그인 할 때 받았던 토큰을 cred 인자로 받는다.
3. addGuestBook에서 받아온 정보를 GuestBookReturn 형태에 맞추어 반환한다.
4. 문제없이 작동했다면 반환할 때 201 상태 코드를 반환한다.

2. 모든 방명록 조회

@router.get("", response_model=List[GuestBookReturn], status_code=status.HTTP_200_OK)
async def guestbook(db: Session = Depends(get_db)):
    return await findAllGuestBook(db)

1. findAllGuestBook에서 받아온 정보를 GuestBookReturn으로 이루어진 List 형태에 맞춰 반환한다.
2. 문제없이 작동했다면 반환할 때 200 상태 코드를 반환한다.

3. 방명록 수정

@router.put("/{id}", response_model=GuestBookReturn, status_code=status.HTTP_202_ACCEPTED)
async def guestbook_update(
    data: GuestBookUpdate,
    id: int,
    cred: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db),
):
    return await updateGuestBook(data, id, cred, db)

1. GuestBookUpdate 형태에 맞추어 Body 정보를 받는다.
2. 수정할 GuestBook 객체의 ID를 Path 파라미터로 받는다.
3. 로그인 할 때 받았던 토큰을 cred 인자로 받는다.
4. updateGuestBook에서 받아온 정보를 GuestBookReturn 형태에 맞추어 반환한다.
5. 문제없이 작동했다면 반환할 때 202 상태 코드를 반환한다.

service.py

from ..user import utils
from .utils import add_guestbook, find_all_guestbook, update_guestbook


async def addGuestBook(data, cred, db):
    decoded_dict = await utils.verify_user(cred)
    return await add_guestbook(data, decoded_dict.get("id"), db)


async def findAllGuestBook(db):
    return await find_all_guestbook(db)


async def updateGuestBook(data, id, cred, db):
    decoded_dict = await utils.verify_user(cred)
    return await update_guestbook(data.message, id, decoded_dict.get("id"), db)

1. 방명록 생성

async def addGuestBook(data, cred, db):
    decoded_dict = await utils.verify_user(cred)
    return await add_guestbook(data, decoded_dict.get("id"), db)

1. verify_user 함수에 받아온 토큰 값을 넣어 해독된 딕셔너리를 받아온다.
2. add_guestbook으로 메세지와 작성자를 DB에 저장한 뒤 반환한다.

2. 모든 방명록 조회

async def findAllGuestBook(db):
    return await find_all_guestbook(db)

1. find_all_guestbook에서 받은 정보를 반환한다.

3. 방명록 수정

async def updateGuestBook(data, id, cred, db):
    decoded_dict = await utils.verify_user(cred)
    return await update_guestbook(data.message, id, decoded_dict.get("id"), db)

1. verify_user 함수에 받아온 토큰 값을 넣어 해독된 딕셔너리를 받아온다.
2. update_guestbook에서 방명록을 수정한 뒤 정보를 반환한다.

utils.py

from fastapi import HTTPException, status
from sqlalchemy import desc

from app.models import GuestBook


async def add_guestbook(data, id, db):
    try:
        row = GuestBook(message=data.message, owner_id=id, message_owner=data.message_owner)
        db.add(row)
        db.commit()

        return row
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"{e} occured while adding guestbook",
        )


async def find_all_guestbook(db):
    return db.query(GuestBook).order_by(desc(GuestBook.created_at)).all()


async def update_guestbook(message, id, owner_id, db):
    row = db.query(GuestBook).filter_by(id=id).first()
    if row is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Guestbook Not Found",
        )
    if row.owner_id == owner_id:
        row.message = message
        db.commit()
    else:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="It's not owner")

    return row

1. 방명록 생성

async def add_guestbook(data, id, db):
    try:
        row = GuestBook(message=data.message, owner_id=id, message_owner=data.message_owner)
        db.add(row)
        db.commit()

        return row
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"{e} occured while adding guestbook",
        )

1. 넘어온 데이터로 models.GuestBook 객체를 생성한다.
2. DB에 추가한다.
3. Commit

2. 모든 방명록 조회

async def find_all_guestbook(db):
    return db.query(GuestBook).order_by(desc(GuestBook.created_at)).all()

sqlalchemy에서 쿼리문을 작성할 때 .order_by, .options, .offset, .limit 등으로 여러 옵션을 줘서 작성할 수 있다. desc() 옵션을 추가할 시 내부 인자를 기준으로 내림차순 정렬해서 반환해준다.

1. GuestBook 테이블의 모든 row를 내림차순으로 정렬해서 반환한다.

3. 방명록 수정

async def update_guestbook(message, id, owner_id, db):
    row = db.query(GuestBook).filter_by(id=id).first()
    if row is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Guestbook Not Found",
        )
    if row.owner_id == owner_id:
        row.message = message
        db.commit()
    else:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="It's not owner")

1. Path 파라미터로 들어온 id로 TsItem 객체를 검색한다.
2. row가 None이라면, 즉시 400 상태 코드를 반환한다.
3. models.TsItem 객체의 owner_id(회원 식별 ID)와 토큰 해독 후 나온 ID가 일치하지 않는다면 즉시 403을 반환한다.
4. 일치한다면 방명록 내용을 수정하고 Commit

schema.py

from pydantic import BaseModel


class GuestBookAdd(BaseModel):
    message: str
    message_owner: str


class GuestBookReturn(GuestBookAdd):
    id: int
    owner_id: int


class GuestBookUpdate(BaseModel):
    message: str

생각해볼 점

지난 번과 마찬가지로 정의된 relationship을 가지고 회원 별 방명록 확인도 구현해보는 것도 재밌을 것이다.

[참고]

필자는 어차피 방명록인데 모두가 봐야하지 않나?는 생각에 전체 방명록만 불러오게 구현했다. 테이블 작성할 때 relationship이 적혀있는건 오늘 벨로그 글을 쓰게 되면서 알게 된 사실이다..

이번에도 Swagger로 보면 이렇게 잘 생성되어 있을 것이다!

이 때까지 글을 읽어준 독자님들께 감사하다고 말씀을 드리고 싶다. 이번 게시글까지 해서 현재 작성되어있는 API는 모두 업로드했다. 두 편 정도로 나누어 Test 코드 작성과 회고를 끝으로 이 시리즈를 마무리해보고자 한다.

Database Table 생성

APP/
ㄴ models.py --> 작업할 공간

models.py

# 이전 상태에서 추가로 import 해줘야 할 라이브러리
from sqlalchemy.sql.schema import ForeignKey


class TsItem(BaseMin, Base):
    __tablename__ = "tsitem"

    dialect = Column(String(255), nullable=False)
    standard = Column(String(255), nullable=False)
    english = Column(String(255), nullable=False)
    chinese = Column(String(255), nullable=False)
    japanese = Column(String(255), nullable=False)
    owner_id = Column(Integer, ForeignKey("user.id"))

    owner = relationship("User", back_populates="items")

1. dialect: 사투리
2. standard: 표준어
3. english: 영어
4. chinese: 중국어
5. japanse: 일본어
6. owner_id: (외래키) 추가한 회원의 식별 ID

기능

controller.py

from fastapi import Depends, APIRouter, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm.session import Session
from typing import List

from app.database import get_db
from .schema import TsItemAdd, TsItem, TsItemDelete
from .service import addTsItem, findTsItems, deleteTsItem

router = APIRouter()
security = HTTPBearer()


# TsItem 테이블의 모든 row 로드
@router.get("", response_model=List[TsItem])
async def get_item_list(db: Session = Depends(get_db)):
    return await findTsItems(db)


# TsItem 테이블에 Item 추가
@router.post("", response_model=TsItem, status_code=status.HTTP_201_CREATED)
async def add_item(
    data: TsItemAdd,
    cred: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db),
):
    return await addTsItem(data, cred, db)


# TsItem 테이블에서 Item 제거
@router.delete("/{id}", response_model=TsItemDelete, status_code=status.HTTP_202_ACCEPTED)
async def delete_item(
    id: int,
    cred: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db),
):
    return await deleteTsItem(id, cred, db)

1. TsItem 조회

@router.get("", response_model=List[TsItem])
async def get_item_list(db: Session = Depends(get_db)):
    return await findTsItems(db)

1. findtsItems 에서 받아온 정보를 TsItem으로 이루어진 List 형태에 맞춰 반환한다.
2. 문제없이 작동했다면 반환할 때 200 상태 코드를 반환한다.

2. TsItem 생성

@router.post("", response_model=TsItem, status_code=status.HTTP_201_CREATED)
async def add_item(
    data: TsItemAdd,
    cred: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db),
):
    return await addTsItem(data, cred, db)

1. TsItemAdd 형태에 맞추어 Body 정보를 받는다.
2. 로그인 할 때 받았던 토큰을 cred 인자로 받는다.
3. addTsItem에서 받아온 정보를 TsItem 형식에 맞게 반환한다.
4. 문제없이 작동했다면 반환할 때 201 상태 코드를 반환한다.

3. TsItem 삭제

@router.delete("/{id}", response_model=TsItemDelete, status_code=status.HTTP_202_ACCEPTED)
async def delete_item(
    id: int,
    cred: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db),
):
    return await deleteTsItem(id, cred, db)

1. Path 파라미터로 id 값을 받는다.
2. 로그인 할 때 받았던 토큰을 cred 인자로 받는다.
3. deleteTsItem에서 받아온 정보를 TsItem 형식에 맞게 반환한다.
4. 문제없이 작동했다면 반환할 때 202 상태 코드를 반환한다.

service.py

from ..user import utils
from .utils import add_tsitem, find_tsitems, delete_tsitem


async def addTsItem(data, cred, db):
    decoded_dict = await utils.verify_user(cred)
    row = await add_tsitem(data, decoded_dict.get("id"), db)

    return row


async def findTsItems(db):
    return await find_tsitems(db)


async def deleteTsItem(id, cred, db):
    decoded_dict = await utils.verify_user(cred)
    return await delete_tsitem(id, decoded_dict.get("id"), db)

1. TsItem 추가

async def addTsItem(data, cred, db):
    decoded_dict = await utils.verify_user(cred)
    row = await add_tsitem(data, decoded_dict.get("id"), db)

    return row

user/utils.py 의 verify_user에서 반환되는 딕셔너리 객체에는 JWT 토큰을 인코딩할 때 사용되었던 정보들이 들어있다. ex) email, id

1. verify_user 함수에 받아온 토큰 값을 넣어 해독된 딕셔너리를 받아온다.
2. add_tsitem으로 사투리,표준어,...,일본어를 DB에 저장한다.
3. 반환된 정보를 반환한다

2. TsItem들 반환

async def findTsItems(db):
    return await find_tsitems(db)

1. find_tsitems 함수에서 반환된 정보를 반환한다.

3. TsItem 삭제

async def deleteTsItem(id, cred, db):
    decoded_dict = await utils.verify_user(cred)
    return await delete_tsitem(id, decoded_dict.get("id"), db)

1. verify_user 함수에 받아온 토큰 값을 넣어 해독된 딕셔너리를 받아온다.
2. delete_tsitem 함수로 TsItem 테이블에서 row를 삭제한 뒤 정보를 반환한다.

utils.py

from fastapi import HTTPException, status

from app.models import TsItem


async def add_tsitem(data, id, db):
    try:
        row = TsItem(**data.dict(), owner_id=id)
        db.add(row)
        db.commit()

        return row
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"{e} occured while adding tsitem",
        )


async def find_tsitems(db):
    row = db.query(TsItem).all()

    return row


async def delete_tsitem(id, user_id, db):
    row = db.query(TsItem).filter_by(id=id).first()
    if row is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="No Item Found",
        )
    if row.owner_id == user_id:
        db.delete(row)
        db.commit()
    else:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="It's not owner")

    return row

1. TsItem 추가

async def add_tsitem(data, id, db):
    try:
        row = TsItem(**data.dict(), owner_id=id)
        db.add(row)
        db.commit()

        return row
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"{e} occured while adding tsitem",
        )

1. 넘어온 딕셔너리 객체를 풀어서 models.TsItem 객체를 생성한다. 여기서 owner_id는 외래키이다.
2. DB의 TsItem 테이블에 추가한다.
3. Commit

2. TsItem들 반환

async def find_tsitems(db):
    row = db.query(TsItem).all()

    return row

all() 메소드로 query를 사용하면 리스트 형태로 반환된다.

1. TsItem 테이블의 모든 row를 반환한다.

3. TsItem 삭제

async def delete_tsitem(id, user_id, db):
    row = db.query(TsItem).filter_by(id=id).first()
    if row is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="No Item Found",
        )
    if row.owner_id == user_id:
        db.delete(row)
        db.commit()
    else:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="It's not owner")

    return row

sqlalchemy에서 row를 찾지못하면 None 값을 반환한다.

1. Path 파라미터로 들어온 id로 TsItem 객체를 검색한다.
2. row가 None이라면, 즉시 400 상태 코드를 반환한다.
3. models.TsItem 객체의 owner_id(회원 식별 ID)와 토큰 해독 후 나온 ID가 일치하지 않는다면 403 상태 코드를 반환한다.
4. 일치한다면 객체를 테이블에서 삭제하고 Commit

schema.py

from pydantic import BaseModel


class TsItemAdd(BaseModel):
    dialect: str
    standard: str
    english: str
    chinese: str
    japanese: str


class TsItem(TsItemAdd):
    id: int
    owner_id: int

    class Config:
        from_attributes = True


class TsItemDelete(BaseModel):
    id: int

생각해볼 점

생성한 Table에는 owner 라는 relationship으로 회원과 TsItem이 일대다 관계를 맺고 있지만, 위에 작성된 엔드포인트 중에서 relationship을 사용해서 query문을 작성한 적이 없다. 이는 모든 TsItem을 불러오는 엔드포인트만 존재하기 때문인데, 이 글을 읽고 있는 독자(있나?)가 있다면 이 부분을 생각해보면서 회원 별로 생성한 TsItem을 불러오는 엔드포인트를 작성해보면 좋을 것 같다.

[참고]

필자가 모든 TsItem을 불러오는 엔드포인트로 작성한 이유는, 캡스톤디자인 경연대회 당일날에 다른 사람들은 어떤 문장을 번역해봤을지 늦게 온 사람들도 확인해볼 수 있도록 하기 위해서였다.

위의 내용들을 잘 따라왔다면 사진처럼 Swagger에서 확인이 될 것이다. 이번에도 필자는 리팩토링 이 후 사진이라 V2로 태깅이 되어있다.

다음 게시글에서는 방명록 관련해서 작성하겠다.

지난 포스트에서는 회원가입 및 로그인 부분을 진행했다. 이번에는 사투리를 번역하는 엔드포인트를 작성해보자.

기능 개발에 앞서 -

몇 가지 변경점이 있다.

1. user/utils.py 에 함수 추가

async def verify_user(cred):
    token = cred.credentials
    try:
        jwt_dict = jwt.decode(token, settings.SECRET_KEY, settings.ALGORITHM)
        if jwt_dict:
            return jwt_dict
    except ExpiredSignatureError:
        raise HTTPException(401, "Expired")

1. 토큰 정보를 추출한다.
2. JWT 토큰을 서버의 SECRET_KEY와 ALGORITHM으로 해독한다.
3. 제대로 해독되었다면 해독한 딕셔너리 결과값을 반환한다.
4. 만료된 토큰이라면 401 상태코드를 즉시 반환한다.

2. AI/test.py 변경

from transformers import pipeline
import requests

from app.AI import config
from app.config import settings

nlg_pipeline = pipeline(
    "text2text-generation", model=config.model_path, tokenizer=config.model_name
)


def generate_text(text, num_return_sequences=1, max_length=60):
    target_style_name = "표준어"
    text = f"{target_style_name} 말투로 변환:{text}"
    out = nlg_pipeline(text, num_return_sequences=num_return_sequences, max_length=max_length)
    return [x["generated_text"] for x in out]

# Papago API로 번역하는 함수
def translate_with_papago(text, source_lang="ko", target_lang="en"):
    url = "https://openapi.naver.com/v1/papago/n2mt"

    headers = {
        "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
        "X-Naver-Client-Id": settings.CLIENT_ID,
        "X-Naver-Client-Secret": settings.CLIENT_SECRET,
    }

    data = {"source": source_lang, "target": target_lang, "text": text}

    response = requests.post(url, headers=headers, data=data)
    translation_result = response.json()
    translated_text = translation_result["message"]["result"]["translatedText"]

    return translated_text

1. 변경 전 코드에서는 deep-translator 라이브러리를 사용했었지만, 번역 품질이 매우 좋지 않았다. 따라서 PAPAGO API를 사용해서 표준어로 변환한 문장을 각각 영어,일본어,중국어로 다시 번역하는 방법으로 변경했다.
2. test.py를 직접 실행했을 때 수행되는 코드 또한 삭제되어있는데, 테스트 코드 작성 후 테스트 해볼 때 포함 안 되는게 거슬려서 삭제했다. 삭제하지 않아도 무방하다.

papago api 사용법은 https://developers.naver.com/docs/papago/README.md 을 참고하자.

또한 .env 파일에 CLIENT_ID와 CLIENT_SECRET도 추가해줘야 한다!

성능 차이

1. Papago

시간을 측정할 때는 영어로 번역하는 부분만 시간을 측정했다.

2. deep-translator

같은 문장 (너 오늘 많이 먹었네)를 스크립트로 돌려보겠다.

from deep_translator import GoogleTranslator
from time import time

start = time()
target_text = "너 오늘 많이 먹었네."

print(f"번역 전 : {target_text}")
print(f"번역 후 :{GoogleTranslator(source='ko', traget='en').translate(target_text)}")
print(f"Deep-Translator 경과시간 : {time() - start}")

deep-translator를 사용하는 것보다 papago를 사용하는게 시간 측면에서 유리했다. 안 그래도 NLP모델을 CPU 환경에서 사용중인데 최대한 부담을 줄여주는게 좋아보였다.

번역

APP/
ㄴ API/
  ㄴ translate/ --> 작업할 공간
    ㄴ service.py 
    ㄴ controller.py
    ㄴ schema.py

v1/__init__.py

from fastapi import APIRouter

from .user import controller as user
from .translate import controller as translate

router = APIRouter()

router.include_router(user.router, prefix="/users", tags=["User"])
router.include_router(translate.router, prefix="/AI", tags=["AI"])

먼저 __init__.py에 우리가 작성할 router를 포함시켜주자.

controller.py

from fastapi import Depends, APIRouter
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from .schema import Translated, ToTranslate
from .service import itemTranslate

router = APIRouter()
security = HTTPBearer()

# 사투리 -> 표준어 및 다국어로 변환 후 반환
@router.post("", response_model=Translated)
async def translate_item(
    data: ToTranslate,
    cred: HTTPAuthorizationCredentials = Depends(security),
):
    return await itemTranslate(data, cred)

1. ToTranslate 형태에 맞게 Body를 받는다.
2. 로그인 할 때 받았던 토큰을 cred 인자로 받는다.
3. itemTranslate에서 반환된 정보를 Translated 형식에 맞게 반환한다.
4. 문제없이 작동했다면 반환할 때 200 상태 코드를 반환한다.

service.py

from app.AI import test
from ..user import utils


async def itemTranslate(data, cred):
    if await utils.verify_user(cred):
        translated_text = test.generate_text(data.dialect)[0]  # Translation
        english_text = test.translate_with_papago(translated_text)
        chinese_text = test.translate_with_papago(translated_text, target_lang="zh-CN")
        japanese_text = test.translate_with_papago(translated_text, target_lang="ja")

        return {
            "dialect": data.dialect,
            "standard": translated_text,
            "english": english_text,
            "chinese": chinese_text,
            "japanese": japanese_text,
        }

1. JWT 토큰을 검증한다.
2. translate_with_papago 함수로 표준어를 영어, 중국어, 일본어로 번역한다.
3. 딕셔너리 형태로 사투리, 표준어, 영어, 중국어, 일본어를 반환한다.

schema.py

from pydantic import BaseModel


class ToTranslate(BaseModel):
    dialect: str


class Translated(ToTranslate):
    standard: str
    english: str
    chinese: str
    japanese: str

위의 내용들을 잘 따라왔다면 Swagger에서 확인이 될 것이다. 이번에도 필자는 리팩토링 이후 사진이라 V2로 태깅이 되어있다.

유의할 점은 Swagger에서 이 엔드포인트를 사용하려고 할 때 옆의 자물쇠 버튼을 클릭해서 Login하고 헤더에 들어있는 토큰 값을 넣어줘야하니 잘 기억해놓도록 하자.

폴더 구조

DOMAIN ㄴ service.py --> 비즈니스 로직이 작성되는 스크립트 ㄴ controller.py --> 엔드포인트들이 작성되는 스크립트 ㄴ utils.py --> 비즈니스 로직에서 사용되는 함수들이 작성되는 스크립트 ㄴ schema.py --> 입출력 형태를 검증하는 class들이 작성되는 스크립트

위의 형태로 각 도메인들을 채워나갈 것이다.

FastAPI 간단 설명

from fastapi import FastAPI
from pydantic import BaseModel

class ItemsReturn(BaseModel):
    start: int
    limit: int

class AddItem(BaseModel):
    name: str
    price: int

# FastAPI 앱 설정
app = FastAPI()

# app에 'get' 메소드로 http://{host}:{port}/hello에 접근하겠다.
@app.get("/hello")
async def print_hello():
    return {"message": "Hello, World!"}

# app에 'get' 메소드로 http://{host}:{port}/{id}에 접근하겠다.
@app.get("/{id}")
async def print_id(id: int):
    return {"id": id}

# app에 'post' 메소드로 http://{host}:{port}/items에 접근하겠다.
# 반환되는 상태 코드를 201로 하겠다.
@app.post("/items",status_code=201)
# 받아오는 Body 형태를 AddItem에 맞추겠다.
async def add_item(item: AddItem):
    return {"name": item.name, "price": item.price}

# app에 'get' 메소드로
# http://{host}:{port}/items/?start={start}&limit={limit}에 접근하겠다.
# 반환 형태를 ItemsReturn에 맞추겠다.
# 반환되는 상태 코드를 200으로 하겠다.
@app.get("/items/", response_model=ItemsReturn, status_code=200)
async def print_item(start: int, limit: int):
    return {"start": start, "limit": limit}

이를 main.py로 저장하고

콘솔창에서 아래 구문을 실행시켜보자.

uvicorn main:app --host 0.0.0.0

정상적으로 작성했다면, FastAPI 서버가 실행되었을 것이다. http://localhost:8000/docs 로 들어가면 자동으로 Swagger UI가 생성되어있는 것을 볼 수 있다.

우리 프로젝트에서는?

API
ㄴ __init__.py
ㄴ v1
    ㄴ __init__.py

이 __init__.py에다가 APIRouter로 router를 만들어 main에서 선언한 app에 추가하는 방식으로 엔드포인트를 선언하여 유지보수를 더 쉽게 할 수 있다.

api/__init__.py

from fastapi import APIRouter

from . import v1

router = APIRouter()
router.include_router(v1.router, prefix="/v1")

v1/__init__.py

from fastapi import APIRouter

from .user import controller as user

router = APIRouter()

router.include_router(user.router, prefix="/users", tags=["User"])

prefix --> router에 포함되어 있는 엔드포인트 앞에 붙여줄 전치사 tags --> Swagger에서 tag 아래로 정렬해준다

회원가입 및 로그인

왜 회원가입이..?

경상도 사투리를 번역해서 보여주는 웹사이트에 회원가입 및 로그인 기능이 반드시! 필요하냐라고 물어보면 자신있게 예쓰라고 대답하긴 힘들지만, 추가하고자 하는 기능 중에 하나(아직 구현하지 않았다)가 사용자의 피드백을 통해서 추가 학습까지 가능한 파이프라인을 구축하는 것이기 때문에 익명으로 받아오는 정보를 사용하기보다는 회원가입으로 추적 가능한 데이터를 사용하는게 더 나을 것 같아서 구현해보았다.

Database Table 생성

APP/
ㄴ models.py --> 작업할 공간

models.py

# 이전 상태에서 추가로 import 해줘야 할 라이브러리들.
from sqlalchemy import String, Boolean
from sqlalchemy.orm import relationship

class User(BaseMin, Base):
    __tablename__ = "user"

    email = Column(String(30), nullable=False, unique=True)
    password = Column(String(255), nullable=False)
    is_provider = Column(Boolean, default=True)
    items = relationship("TsItem", back_populates="owner")
    guestbooks = relationship("GuestBook", back_populates="book_owner")

    def as_dict(self):
        return {column.name: getattr(self, column.name) for column in self.__table__.columns}

User 테이블은 BaseMin도 상속받았기 때문에, id, created_at, updated_at은 자동으로 생성된다.

nullable=False --> 반드시 존재해야 하는 값 unique=True --> 테이블에서 하나만 존재해 하는 값 default=True --> 별도로 지정해주지 않았을 때 기본으로 저장되는 값

1. email: 로그인 및 회원가입 시 식별하기 위한 이메일
2. password: 해쉬화되어 DB에 저장된다
3. is_provider: 위에서 언급한 파이프라인 구축을 위한 피드백 제공 여부
4. items: 후에 작성할 TsItem에 대해서 일대다 관계를 형성한다.
5. guestbooks: 후에 작성할 guestbooks에 대해서 일대다 관계를 형성한다.
6. as_dict -> models.User 객체를 딕셔너리 형태로 반환하는 함수

기능

APP/
ㄴ API/
  ㄴ USER/ --> 작업할 공간
    ㄴ service.py 
    ㄴ controller.py
    ㄴ utils.py
    ㄴ schema.py

controller.py

from fastapi import APIRouter, status, Depends
from sqlalchemy.orm.session import Session

from app.database import get_db
from .schema import UserAdd, UserAddReturn, DuplicatedEmail
from .service import userAdd, userLogin, emailDuplicated

router = APIRouter()

# 회원가입
@router.post("/register", response_model=UserAddReturn, status_code=status.HTTP_201_CREATED)
async def add_user(data: UserAdd, db: Session = Depends(get_db)):
    return await userAdd(data, db)

# 로그인
@router.post("/login")
async def issue_token(data: UserAdd, db: Session = Depends(get_db)):
    return await userLogin(data, db)

# 이메일 중복 여부 확인
@router.get("/duplicated", status_code=status.HTTP_200_OK)
async def is_duplicated(data: DuplicatedEmail = Depends(), db: Session = Depends(get_db)):
    return await emailDuplicated(data, db)

Depends()

엔드포인트에서 매개변수로 Depends로 함수를 전달해주면, FastAPI가 자동으로 함수를 실행시켜 엔드포인트에 전달해준다.

database.py의 get_db를 살펴보도록 하자.

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

database.py의 윗줄에서 생성한 SessionLocal의 인스턴스를 생성해서 넘겨준다. 우리는 이 세션을 통해서 MySQL과 통신할 것이다!

1. 회원가입

### 회원가입 ###
@router.post("/register", response_model=UserAddReturn, status_code=status.HTTP_201_CREATED)
async def add_user(data: UserAdd, db: Session = Depends(get_db):
    return await userAdd(data, db)

1. UserAdd 형태에 맞게 Body를 받는다.
2. userAdd 함수에서 반환된 값을 UserAddReturn 형태에 맞추어 반환한다.
3. 문제없이 작동했다면 반환할 때 201 상태 코드를 반환한다.

2. 로그인

@router.post("/login")
async def issue_token(data: UserAdd, db: Session = Depends(get_db)):
    return await userLogin(data, db)

1. UserAdd 형태에 맞게 Body를 받는다.
2. userLogin에서 반환된 값을 반환한다.
3. 문제없이 작동했다면 반환할 때 200 상태 코드를 반환한다.

3. 닉네임 중복 확인

@router.get("/duplicated", status_code=status.HTTP_200_OK)
async def is_duplicated(data: DuplicatedEmail = Depends(), db: Session = Depends(get_db)):
    return await emailDuplicated(data, db)

1. DuplicatedEmail 형태에 맞게 Query 파라미터를 받는다. ( Pydantic Class를 Depends로 받으면 Query 파라미터로 받을 수 있다.)
2. emailDuplicated에서 반환된 값을 반환한다.
3. 문제없이 작동했다면 반환할 때 200 상태 코드를 반환한다.

service.py

pip install bcrypt

import bcrypt
from fastapi.responses import JSONResponse
from .utils import (
    add_user,
    find_user_by_email,
    create_access_token,
    is_password_correct,
    is_duplicated,
)


async def userAdd(data, db):
    salt_value = bcrypt.gensalt()
    pw = bcrypt.hashpw(data.password.encode(), salt_value)

    row = await add_user(data.email, pw, db)

    return row.as_dict()


async def userLogin(data, db):
    user = await find_user_by_email(data.email, db)
    if await is_password_correct(data.password, user.password):
        token, user_id = await create_access_token(user)
        return JSONResponse(content={"user_id": user_id}, headers={"access_token": token})


async def emailDuplicated(data, db):
    return {"duplicated": await is_duplicated(data.email, db)}

1. 회원가입

async def userAdd(data, db):
    salt_value = bcrypt.gensalt()
    pw = bcrypt.hashpw(data.password.encode(), salt_value)

    row = await add_user(data.email, pw, db)

    return row.as_dict()

1. bcrypt 라이브러리를 활용해 솔트값을 생성한다.
2. body에 담겨온 password를 솔트값으로 해쉬화한다.
3. add_user 함수로 email, 해쉬화된 비밀번호를 DB에 저장한다.
4. 반환된 models.User 객체를 딕셔너리로 바꾸어 반환한다.

2. 로그인

async def userLogin(data, db):
    user = await find_user_by_email(data.email, db)
    if await is_password_correct(data.password, user.password):
        token, user_id = await create_access_token(user)
        return JSONResponse(content={"user_id": user_id}, headers={"access_token": token})

1. find_user_by_email 함수에서 body에 담겨온 email로 models.User 객체를 반환한다.
2. is_password_correct 함수로 비밀번호가 일치하는지 확인한다.
3. create_access_token 함수로 JWT 토큰과 user_id를 반환한다.
4. 토큰은 Header에 user_id는 콘텐트에 담아 반환한다.

JWT 토큰에 관련된 설명은 https://velog.io/@chuu1019/%EC%95%8C%EA%B3%A0-%EC%93%B0%EC%9E%90-JWTJson-Web-Token 이곳에 잘 되어 있으니 확인해보길 바란다.

3. 닉네임 중복 확인

async def emailDuplicated(data, db):
    return {"duplicated": await is_duplicated(data.email, db)}

1. is_duplicated 함수에서 body에 담겨온 email로 중복 여부를 판단해 boolean값을 딕셔너리에 담아 반환한다.

utils.py

import bcrypt

from datetime import datetime, timedelta
from fastapi import HTTPException, status
from jose import jwt

from app.config import settings
from app.models import User
from .schema import UserPayload


async def add_user(email, pw, db):
    try:
        row = User(**{"email": email, "password": pw})
        db.add(row)
        db.commit()

        return row
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"{e} occured while registering",
        )


async def find_user_by_email(email, db):
    row = db.query(User).filter_by(email=email).first()
    if row:
        return row
    else:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="No Email Found",
        )


async def is_password_correct(data, user):
    if bcrypt.checkpw(data.encode(), user.encode()):
        return True
    else:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Incorrect Password",
        )


async def create_access_token(user):
    user_schema = user.as_dict()
    expire = datetime.utcnow() + timedelta(days=1)
    user_info = UserPayload(**user_schema, exp=expire)

    return (
        jwt.encode(user_info.dict(), settings.SECRET_KEY, algorithm=settings.ALGORITHM),
        user_schema["id"],
    )


async def is_duplicated(email, db):
    if db.query(User).filter_by(email=email).first():
        return True
    else:
        return False

1. DB에 유저 추가

async def add_user(email, pw, db):
    try:
        row = User(**{"email": email, "password": pw})
        db.add(row)
        db.commit()

        return row
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=f"{e} occured while registering",
        )

1. email과 해쉬화된 비밀번호로 models.User 객체를 생성한다.
2. DB의 User 테이블에 추가한다.
3. Commit

async def find_user_by_email(email, db):
    row = db.query(User).filter_by(email=email).first()
    if row:
        return row
    else:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="No Email Found",
        )

1. email로 User 테이블에서 models.User 객체를 검색한다.
2. 존재한다면 (row is not None) 객체 반환
3. 존재하지 않다면 403 상태 코드를 즉시 반환한다.

2. 비밀번호 확인

async def is_password_correct(data, user):
    if bcrypt.checkpw(data.encode(), user.encode()):
        return True
    else:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Incorrect Password",
        )

1. Body의 비밀번호와 models.User 객체의 비밀번호가 일치하는지 확인한다.
2. 일치한다면 True 반환
3. 일치하지 않다면 403 상태 코드를 즉시 반환한다.

3. JWT 토큰 생성

async def create_access_token(user):
    user_schema = user.as_dict()
    expire = datetime.utcnow() + timedelta(days=1)
    user_info = UserPayload(**user_schema, exp=expire)

    return (
        jwt.encode(user_info.dict(), settings.SECRET_KEY, algorithm=settings.ALGORITHM),
        user_schema["id"],
    )

1. models.User 객체를 딕셔너리화 한다.
2. 토큰의 만료기간을 정의한다.
3. 위 두 개를 합쳐 토큰화 할 정보를 생성한다.
4. .env 파일에 정의된 SECRET_KEY와 ALGORITHM으로 토큰을 생성한 뒤 user_id와 함께 반환한다.

지난번 .env 파일에는 SECRET_KEY와 ALGORITHM이 정의되어 있지 않으므로 추가해줘야 할 것이다.

4. 닉네임 중복 확인

async def is_duplicated(email, db):
    if db.query(User).filter_by(email=email).first():
        return True
    else:
        return False

1. Query 파라미터로 받아온 email을 가진 models.User 객체가 존재한다면 True 반환
2. 그렇지 않다면 False 반환

schema.py

from datetime import datetime
from pydantic import BaseModel


class DuplicatedEmail(BaseModel):
    email: str


class UserAdd(DuplicatedEmail):
    password: str


class UserAddReturn(DuplicatedEmail):
    id: int


class UserPayload(UserAddReturn):
    exp: datetime

위의 내용들을 잘 따라왔다면 사진처럼 Swagger에서 확인이 될 것이다. 필자는 리팩토링을 거치고 난 뒤의 모습이라 UserV2 태그의 v2로 보이지만, 게시글들을 잘 따라왔다면 User 태그의 /v1 prefix로 보일 것이다. 테스트 코드는 API들을 작성한 뒤 마지막에 한 번에 쭉쭉 업로드하겠다.

Pre-Commit

Pre-Commit이란?

깃허브와 같은 코드 저장소에 커밋을 수행하기 전에 포맷팅이나 린팅이 잘 되어 있는지 확인해주는 도구라고 볼 수 있다. 리팩토링을 진행하면서 코드 가독성을 높이기 위해 도입해보았다.

pip install pre-commit
pre-commit install

먼저 pre-commit을 설치해준 뒤

.vscode/
 ㄴ settings.json
pyproject.toml
.flake8
.pre-commit-config.yaml

최상위 깃에 위 파일들을 생성한다.

settings.json

# settings.json
{
"python.linting.flake8Enabled": true,
  "python.linting.flake8Args": [
      "--max-line-length=100"
  ],
  "python.formatting.provider": "black",
  "python.formatting.blackArgs": ["--line-length", "100"]
}

.flake8

# .flake8
[flake8]
exclude =
    .git,
    .gitignore,
    *.pot,
    *.py[co],
    __pycache__,
    venv,
    .env

ignore =
    E121,
    E126,
    E127,
    E128,
    E203,
    E225,
    E226,
    E231,
    E241,
    E251,
    E261,
    E265,
    E302,
    E303,
    E305,
    E402,
    E501,
    E741,
    W291,
    W292,
    W293,
    W391,
    W503,
    W504,
    F403,
    B007,
    B950,

max-line-length = 100

.pyproject.toml

# .pyproject.toml

[tool.black]
line-length = 100
target-version = ['py311']
exclude = '''
  \.git
  | \.hg
  | \.mypy_cache
  | \.tox
  | \.venv
  | _build
  | buck-out
  | build
  | dist
'''

[tool.flake8]
max-line-length = 100

.pre-commit-config.yaml

# .pre-commit-config.yaml

repos:
  - repo: https://github.com/pre-commit/pre-commit-hooks
    rev: v4.4.0
    hooks:
      - id: trailing-whitespace
      - id: check-yaml
      - id: check-json

  - repo: https://github.com/psf/black
    rev: 23.9.1
    hooks:
      - id: black
        name: black
        description: "Black: The uncompromising Python code formatter"
        entry: black
        language: python
        minimum_pre_commit_version: 2.9.2
        require_serial: true
        types_or: [python, pyi]

  - repo: https://github.com/pycqa/flake8
    rev: 6.1.0
    hooks:
      - id: flake8

이렇게 설정 파일들을 각각 작성해주고 난 뒤 vscode에서 저장하면 black 포맷터가 자동으로 파일을 포맷팅해준다.

이 후 pre-commit을 설치한 가상환경을 activate하고 git commit 했을 때 pre-commit이 제대로 작동하는 것을 확인할 수 있다.

[예시]

지난 번 게시글까지 해서 프로젝트에 사용될 AI모델을 학습하였다. 이제부터는 AI모델을 파이썬 백엔드에 붙여서 간단하게 API화 해보도록 하자.

FastAPI

이번에 내가 선택한 웹프레임워크는 FastAPI이다! Django를 사용해서 이번 프로젝트를 진행하기에는 너무 가벼운 서비스였고, Flask보다 FastAPI가 차후에 더 널리 쓰일 웹프레임워크라 생각해서 이를 선택하기로 했다.

사실 백엔드로 진행해보는 첫 번째 프로젝트이다..

예상 폴더 구조

MAIN
ㄴ APP/
  ㄴ AI/
  ㄴ API/
    ㄴ V1/
      ㄴ UTILS/
        ㄴ utils.py
      apis...
  ㄴ config.py
  ㄴ database.py
  ㄴ main.py
  ㄴ schemas.py
ㄴ SAVED_MODEL/
  ㄴ model_weights..
ㄴ dataset/
  ㄴ data.tsv
ㄴ .env
ㄴ server.py

프로젝트 생성

FastAPI를 사용하기 위해서 두 라이브러리를 설치해주자 uvicorn은 ASGI 서버를 구동하기 위한 라이브러리며, 비동기를 지원하는 FastAPI를 실행하기 위해 설치해준다.

pip install fastapi uvicorn sqlalchemy

1. server.py

# server.py
# 서버 구동 시 실행해줄 파이썬 스크립트

import uvicorn

if __name__ == "__main__":
    # 서버 구동 이후 디버깅을 위해 reload=True 옵션을 추가한다    
    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

2. app/main.py

# app/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app import api

app = FastAPI()

# 프론트를 다른 PC에서 구동할 예정이어서 CORS 에러 방지를 위해
# 미들웨어를 추가해줬다.
app.add_middleware(
    CORSMiddleware,
    allow_origins={"*"},
    allow_credentials=True,
    allow_methods={"OPTIONS", "GET", "POST"},
    allow_headers={"*"},
)

# 서버 최초 구동 시에 데이터베이스와 서버를 연결하는 작업
@app.on_event("startup")
def on_startup():
  from app import models
  from app.database import engine

  models.Base.metadata.create_all(bind=engine)

# 라우터를 포함하는 작업으로 엔드포인트들을 연결해준다
app.include_router(api.router)

[수정]

FastAPI에서 on_event 사용을 지양하고 있습니다. https://velog.io/@taegong_s/series/%EC%8B%AD%EC%9E%90%EB%A7%90%ED%92%80%EC%9D%B4-%EA%B2%8C%EC%9E%84 이 시리즈에서 서버 세팅 글에 lifespan으로 같은 일을 하는 부분을 포스팅 해놨으니 참고하시길 바랍니다. 제가 폴더 구조 설계부터 많은 발전이 있었는데 최근 프로젝트 게시글을 보시는게 좋을 듯하여 따로 수정글을 남깁니다.

3. .env 환경변수

.env 파일을 최상위 폴더에 생성하고 gitingnore에 추가해서 따로 관리하자

DB_USERNAME={USERNAME}
DB_HOST={HOST}
DB_PASSWORD={PASSWORD}
DB_NAME={NAME}
DB_PORT={PORT}

4. app/config.py

# app/config.py
import os
from dotenv import load_dotenv
from functools import lru_cache

load_dotenv()

class Settings():
  DB_USERNAME = os.environ.get("DB_USERNAME")
  DB_HOST = os.environ.get("DB_HOST")
  DB_PASSWORD = os.environ.get("DB_PASSWORD")
  DB_NAME = os.environ.get("DB_NAME")
  DB_PORT = int(os.environ.get("DB_PORT"))

@lru_cache
def get_settings():
    return Settings()

settings = get_settings()

lru_cache 데코레이터를 사용하면, 이 후에 get_settings 함수의 결과를 캐싱해서 빠르게 반환할 수 있도록 도와준다

5. app/database.py

SQLAlchemy ORM을 사용해서 파이썬 코드를 통해 MySQL에 접근할 수 있도록 세팅하는 스크립트다.

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from app.config import settings

# MySQL과 연결할 Engine 생성
engine = create_engine(
  "mysql+pymysql://{username}:{password}@{host}:{port}/{name}".format(
    username=settings.DB_USERNAME,
    password=settings.DB_PASSWORD,
    host=settings.DB_HOST,
    port=settings.DB_PORT,
    name=settings.DB_NAME,
  )
)

# SQLAlchemy가 데이터베이스와 상호작용할 때 사용하는 Session생성
SessionLocal = sessionmaker(
    bind=engine,
    autocommit=False,
    autoflush=False,
)

# 이 후 데이터베이스의 테이블을 정의할 때 이 Base를 상속받는다.
Base = declarative_base()

# 엔드포인트에서 Session 객체를 생성할 때 종속성 주입으로 이 함수의 리턴 값을 전달한다.
def get_db():
  db = SessionLocal()
  try:
    yield db
  finally:
    db.close()

6. app/models.py

MySQL의 테이블을 생성하는 코드를 작성하는 스크립트다.

# app/models.py
from sqlalchemy import Column, Integer, DateTime, func
from app.database import Base

# 테이블의 기본이 되는 인덱스, 생성시각, 변경시각을 사전 정의해두자
class BaseMin:
  id = Column(Integer, primary_key=True, index=True)
  created_at = Column(DateTime, nullable=False, default=func.utc_timestamp())
  updated_at = Column(DateTime, nullable=False, default=func.utc_timestamp(),
                      onupdate=func.utc_timestamp())

### 프로젝트를 진행하면서 필요한 테이블들을 추가하자

7. app/schemas.py

Pydantic 라이브러리를 이용하면 Response_model이나 엔드포인트에서 필요한 인자의 타입을 검증할 수 있다. 이에 필요한 class들을 정의하는 스크립트다.

# app/schemas.py
from pydantic import BaseModel

# 프로젝트에서 사용하지는 않지만 예시로 하나 작성해두었다.
### 프로젝트를 진행하면서 필요한 클래스들을 추가하자
class Example(BaseModel):
    id: int
    name: str

이 후 개발을 위한 환경설정을 해주었다. 앞으로 필요한 기능을 하나씩 추가해보면서 프로젝트를 완성해보자!

[수정]

리팩토링을 거치면서 위에서 작성한 폴더 구조도 다르고, 주요 코드들도 바뀌었다. 이후 게시글들과 https://github.com/fnzksxl/capston-design 의 v2 API들을 참고해주길 바란다.

모델 파라미터 설정

# config.py
from transformers import AutoModelForSeq2SeqLM,AutoTokenizer

# 모델 학습 후에 가중치 저장할 폴더
model_path='./saved_model'
# 모델 초기 가중치 로드할 곳
model_name = "gogamza/kobart-base-v2"
# 데이터셋 파일
data_root='./dataset'

# 내가 학습한 모델 가중치의 유무에 따라 분기 처리
if os.path.exists(f'{model_path}/pytorch_model.bin'):
  print("Use Customized Model")
  model = AutoModelForSeq2SeqLM.from_pretrained(model_path)
else:
  print("Use Pretrained Model")
  model = AutoModelForSeq2SeqLM.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)

# Training Arguments
args={
  num_train_epochs=3
  per_device_train_batch_size=32
  per_device_eval_batch_size=32

  # 저장 폴더에 가중치 덮어쓰기 하는 옵션
  overwrite_output_dir=True

  # 몇 번의 학습 스텝이 지난 후에 평가를 수행할지 결정
  # 하나의 배치를 처리했을 때가 하나의 스텝이라 볼 수 있음
  eval_steps=3000 

  # 몇 번의 학습 스텝이 지난 후에 체크포인트를 진행할 지 결정
  # 중간에 학습이 끊기더라도 다시 진행하기 용이함
  save_steps=3000

  # LR(Learning Rate) 스케쥴러에서 사용할 스텝 수
  # Warmup을 통해 LR을 조절해 파라미터 업데이트의 폭을 조절함
  warmup_steps=300

  # 평가 수행 주기를 Epoch와 Steps중 Steps로 선택
  evaluation_strategy="steps"

  # 메모리 관리를 위해 예측값에 대한 손실만 계산하게 하는 옵션
  prediction_loss_only=True

  # 학습 중 저장되는 체크포인트의 최대 개수
  save_total_limit=3
}

모델 학습 코드

# train.py
from transformers import Seq2SeqTrainingArguments,Seq2SeqTrainer,\
                         DataCollatorForSeq2Seq
import dataset
import config

train_dataset,test_dataset = dataset.make_dataset(
    config.data_root, config.tokenizer
)

# 각 배치를 모델 학습에 알맞은 형태로 바꿔주는 역할
data_collator = DataCollatorForSeq2Seq(
    tokenizer=config.tokenizer, model=config.model
)

training_args = Seq2SeqTrainingArguments(
    **config.args
    output_dir=config.model_path,
    )

trainer = Seq2SeqTrainer(
    model=config.model,
    args=training_args,
    data_collator=data_collator,
    train_dataset=train_dataset,
    eval_dataset=test_dataset,
)

# 모델 학습 진행
try:
  trainer.train()
except Exception as e:
  print(f"Failed to train model caused by {e}")

try:
  trainer.save_model(config.model_path)
  print("Model saved successfully.")
except Exception as e:
  print(f"Failed to save model caused by {e}")

예전에 작성 했던 코드들을 다시 되돌아보며 velog에 작성하고 있는데, 왜 이랬지? 싶은 부분들은 수정해서 업로드하고 있지만 기능적으로 빠진 부분들은 당장 추가하기에 어렵다.

또한 모델 학습 중 Evaluation 되는 수치들에 대한 정보가 없어 이를 참고용으로 올리지 못하는 것도 아쉽다.

다음에 AI로 학습할 일이 있다면 학습 결과는 꼭 따로 보관해야겠다..

(현재 폴더 구조)

AI Folder
ㄴ config.py
ㄴ train.py
ㄴ saved_model/
   ㄴ (saved_model_weights_files)
ㄴ dataset/
   ㄴ data.tsv

위 코드까지 문제없이 돌아갔다면, saved_model 폴더에 가중치가 잘 저장되었을 것이다!

테스트

표준어로 바뀐 사투리를 영어로도 바꾸기 위해 라이브러리를 하나 설치해주자!

pip install deep-translater

# test.py
from transformers import pipeline
from deep_translator import GoogleTranslator

nlg_pipeline=pipeline('text2text-generation',model=config.model_path,tokenizer=config.model_name)

def generate_text(pipe, text, num_return_sequences, max_length):
  target_style_name = "표준어"
  text = f"{target_style_name} 말투로 변환:{text}"
  out = pipe(text, num_return_sequences=num_return_sequences, max_length=max_length)
  return [x['generated_text'] for x in out]

num_return_sequences의 값에 따라서 반환되는 텍스트의 개수가 바뀐다 만약 3으로 지정했다면 길이가 3인 리스트에 담겨서 값이 반환될 것임!

def get_koen_text(target_text_ko):
  return GoogleTranslator(source='ko',target='en').translate(target_text_ko)

print("Write 'q' to exit")
while True:
  src_text=input("Dialect to translate : ")
  if src_text == 'q':
    break
  target_text_ko=generate_text(nlg_pipeline,src_text,num_return_sequences=1,max_length=64)[0]
  target_text_en = get_koen_text(target_text_ko)
  print(f"Translated Standard : {target_text_ko}")
  print(f"Translated Dialect : {target_text_en}")

여기까지 잘 따라와졌다면, 여러분은 경상도 사투리를 표준어로 번역해서 영어로 보여줄 수 있는 프로그램을 작성한 것이다! 아직 시도해보지는 않았지만 AI HUB에 있는 다른 방언 데이터셋을 이용해서 비슷한 방법으로 학습한다면 다른 사투리도 충분히 표준어로 바꿀 수 있다고 생각한다.