이용자 앱 구현

시리즈, 입문

2024. 11. 13. PM 2:44:13

#fastapi #sqlalchemy #async #sqlmodel #pytest #sqlite #의존성 주입 #Depends #orm

User 모델

FastAPI는 데이터 영역을 Pydantic에 의존해서 다뤄요. 그러니 이용자 정보를 다루는 모델을 pydantic으로 만듭니다. pudding_todo 디렉터리 안에 있는 apps 디렉터리 안에 account 디렉터리를 만들고 그 안에 models.py 파일을 만들고 다음 코드를 작성하세요.

# pudding_todo/apps/account/models.py

from pydantic import BaseModel
from starlette.authentication import BaseUser

class User(BaseUser, BaseModel):
    username: str
    password: str

    @property
    def is_authenticated(self) -> bool:
        return True

    @property
    def display_name(self) -> str:
        return self.username

    @property
    def identity(self) -> str:
        return self.username

BaseModel은 Pydantic으로 데이터를 다룰 모델을 만들 때 사용하는 기반 클래스예요. 이 클래스를 상속받아야 Pydantic에서 제공하는 편리한 기능을 사용할 수 있지요.

BaseUser 클래스는 Starlette에서 사용하는 사용자 정보의 자료 구조예요. Starlette은 HTTP Request 정보를 starlette.requests.Request 클래스를 인스턴스 객체로(대개 request 라는 이름) 제공하는데, 웹 애플리케이션 서버에 접속하는 클라이언트의 인증 정보를 토대로 사용자 정보를 request.user 로 제공하죠. Type hint 등 코드 에디터의 지원을 받고자 상속받는 것일 뿐이니, 굳이 BaseUser 클래스를 상속받지 않아도 동작엔 전혀 문제 없어요.

User Service 구현

User Service는 사용자 인증 등을 처리하는 계층입니다. 웹 클라이언트가 FastAPI 웹 애플리케이션 서버에 접근하면, 클라이언트가 접근한 URL을 보고 Router에 등록된 Endpoint를 찾아요. Endpoint는 클라이언트의 요청(Request)을 처리한 후 응답(Response)하는데, 처리 과정에는 여러 관심사를 사용하기도 해요.

예를 들어, 계정을 다루는 단위가 있고 할 일(Todo)을 다루는 단위가 있으며, 특정 이용자의 상태를 관리하면서 해당 이용자의 할 일을 관리한다면 계정과 할 일을 다루는 두 단위가 필요한 거죠. 이런 단위를 서비스 계층으로 관심사를 나눈 거예요.

그럼 User Service를 만들고, 이 서비스에 인증을 처리하는 authenticate()를 구현해볼게요.

# pudding_todo/apps/account/services.py

from starlette.authentication import UnauthenticatedUser
from .models import User


class UserService:
    _data = {
        "puddingcamp": "PuddingCamp2024",
    }

    async def authenticate(self, username: str, password: str) -> User | UnauthenticatedUser:
        if self._data.get(username) != password:
            return UnauthenticatedUser()
        return User(username=username, password=password)

아직은 영속 데이터를 다루진 않으니 서비스에 사용자 인증 정보가 있도록 임시 구현합니다. authenticated() 메서드는 사용자 계정명과 비밀번호를 인자로 받아서 인증을 처리해요. 인증을 성공하면 앞서 만든 User 모델의 인스턴스 객체를 반환하고, 실패하면 UnauthenticatedUser 클래스의 인스턴스 객체를 반환해요. 두 클래스는 Starlette에 있는 BaseUser 클래스를 상속하고요.

간단한 구현이죠? 어떻게 동작할지 코드만 봐도 예상 가능하지만, 테스트 코드를 작성해서 동작을 검증합니다.

# tests/apps/account/test_user_service.py

import pytest

from starlette.authentication import UnauthenticatedUser
from pudding_todo.apps.account.models import User, BaseUser
from pudding_todo.apps.account.services import UserService


@pytest.mark.parametrize(
    "username, password, expected, is_authenticated",
    [
        ("test", "password", UnauthenticatedUser, False),
        ("puddingcamp", "wrong", UnauthenticatedUser, False),
        ("puddingcamp", "PuddingCamp2024", User, True),
    ],
)
async def test_login(username, password, expected, is_authenticated):
    user_service = UserService()
    user = await user_service.authenticate(username, password)
    assert isinstance(user, BaseUser)
    assert isinstance(user, expected)
    assert user.is_authenticated is is_authenticated

총 세 경우에 대해 테스트를 수행해요. 계정명과 비밀번호가 틀렸을 때, 비밀번호만 틀렸을 때, 유효한 계정명과 비밀번호일 때 UserService가 어떤 객체를 반환하는지 검사하죠. 근데 이 파일은 어디에 작성하냐구요? 최상위 경로에 tests 디렉터리를 만들고, 그곳에 apps 디렉터리를, apps 디렉터리 안에는 account 디렉터리를 만들고, accounts 디렉터리에 test_service.py 파일을 만들어 테스트 코드를 작성하면 돼요.

이 코드는 c30f439 커밋을 참고하세요.

Pytest 설정과 테스트 수행

실제 테스트를 통과하는지 확인하기 위해 Pytest를 설치하고 설정할게요.

poetry add pytest pytest-asyncio pytest-xdist --group dev

먼저 Pytest를 설치하고요.

pyproject.toml 파일에 Pytest 설정을 추가합니다. 이 설정은 bd6359d 커핏을 참고하세요.

이제 최상위 경로에서 pytest .를 실행하면 테스트를 수행해요. 물론 테스트는 잘 통과하고요.

Login API

Login 종단점 함수

인증하는 서비스를 구현했으니, 이 서비스를 사용하여 이용자의 인증 요청을 받아줄 종단점을 구현할 차례예요. 참고로 종단점이라는 표현은 Endpoint를 번역한 것인데, Router에 등록되어 클라이언트의 요청에 대응하는 계층을 FastAPI에서는 Endpoint라고 칭해요. 저는 진입점이나 종단점이라는 표현을 주로 사용하죠.

account 디렉터리에 endpoints.py 파일을 새로 만들어 다음 코드를 작성합니다. models.py 파일이 있는 곳에 만드는 거예요

# pudding_todo/apps/account/endpoints.py

from fastapi import APIRouter, HTTPException, status

from .models import User
from .services import UserService

router = APIRouter()


@router.post("/login", name="login")
async def login(username, password) -> User:
    service = UserService()

    user = await service.authenticate(username, password)
    if not user.is_authenticated:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

    return user

최상위 URL 경로에 동작하는 toppage Endpoint 와 비슷하죠? 차이점이라면 HTTP Post method로 접근하므로 @router.post()로 login 함수를 장식한 것입니다. 접근에 연결할 URL 경로는 /login으로 지정했고요.

login 함수는 FastAPI의 Endpoint이기에 앞서 Python 함수잖아요. 이 함수에 계정명과 비밀번호를 인자로 전달하면 인증 결과를 알 수 있어요. Endpoint로 보면 FastAPI가 계정명과 비밀번호를 login 함수가 전달받아야 하고요. 그런데 FastAPI는 데이터를 Pydantic으로 다룬다고 앞서 설명드렸잖아요. HTTP Request로 전달되는 정보에서 Endpoint 함수에게 필요한 정보를 데이터로 만들어주는 동작을 Pydantic으로 간편하게 구현할 수 있어요.

from fastapi import APIRouter, HTTPException, status

from .models import User
from .schemas import LoginSchema
from .services import UserService

router = APIRouter()


@router.post("/login", name="login")
async def login(payload: LoginSchema) -> User:
    service = UserService()

    user = await service.authenticate(payload.username, payload.password)
    if not user.is_authenticated:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

    return user

이렇게 말이죠. LoginSchema라고 자료형 각주(Type annotation)를 달아서 이 Python 함수가 무엇을 인자로 필요로 하는지 의도를 드러내고 있어요. LoginSchema가 어떤 정보를 다룰지는 예상이 되는데, 실제로는 어떻게 작성했을까요? endpoints.py 파일이 있는 경로에 schemas.py 파일을 새로 열어 다음 코드를 작성하세요.

# pudding_todo/apps/account/schemas.py

from pydantic import BaseModel, str, Field


class LoginSchema(BaseModel):
    username: str = Field(min_length=4)
    password: str = Field(min_length=8)

User 모델과 비슷해요. Field 함수로 LoginSchema 객체의 속성의 값에 어떤 제약이 필요한지 정의한 점은 다르구요.

Router에 login Endpoint 등록

종단점 함수를 구현했으니, 이 함수를 FastAPI 애플리케이션의 라우터에 등록할 차례예요. endpoints.py 파일이 있는 경로에 router.py 파일을 새로 만들고, endpoints.py 에 있는 router 객체를 등록하세요.

# pudding_todo/apps/account/router.py

from fastapi import APIRouter

from .endpoints import router as account_router
router = APIRouter()

router.include_router(account_router)

이제 account 앱의 router를 FastAPI 애플리케이션 서버에 등록합니다.

from fastapi import FastAPI

from pudding_todo.apps.account.router import router as account_router
from pudding_todo.apps.common.router import router as common_router


app = FastAPI()

app.include_router(common_router)
app.include_router(account_router)

common 앱의 라우터를 등록하는 과정과 같아요.

Login API 테스팅

앞서 test_user_service.py 파일에 UserService의 인증 API에 대한 테스트 코드를 작성했잖아요. 종단점에 대한 테스트 코드도 흐름은 비슷해요. 바로 만들어보죠. test_user_service.py 파일이 있는 경로에 test_auth_endpoints.py 파일을 만들고 다음 코드를 작성합니다.

# tests/apps/account/test_auth_endpoints.py

from typing import Generator

import pytest
from fastapi import status
from fastapi.testclient import TestClient

from pudding_todo.app import app


@pytest.fixture()
def client() -> Generator[TestClient, None, None]:
    with TestClient(app) as test_client:
        yield test_client


@pytest.mark.parametrize(
    "username, password, expected_status_code",
    [
        ("puddingcamp", "wrongwrong", status.HTTP_401_UNAUTHORIZED),
        ("puddingcamp", "wrong", status.HTTP_422_UNPROCESSABLE_ENTITY),
        ("puddingcamp", None, status.HTTP_422_UNPROCESSABLE_ENTITY),
        (None, "wrongwrong", status.HTTP_422_UNPROCESSABLE_ENTITY),
    ],
)
def test_failed_login(username: str, password: str, expected_status_code: int, client: TestClient):
    payload = {"username": username, "password": password}
    res = client.post(app.router.url_path_for("login"), json=payload)
    assert res.status_code == expected_status_code


def test_login_successfully(client: TestClient):
    payload = {"username": "puddingcamp", "password": "PuddingCamp2024"}
    res = client.post(app.router.url_path_for("login"), json=payload)
    assert res.status_code == status.HTTP_200_OK

FastAPI는 HTTP 테스팅에 사용하는 TestClient를 제공해요. FastAPI 애플리케이션 인스턴스 객체를 사용해 실제 웹 서버를 실행하지 않고 API에 대해 테스트를 수행할 수 있죠.

@pytest.fixture()
def client() -> Generator[TestClient, None, None]:
    with TestClient(app) as test_client:
        yield test_client

이 코드는 TestClient의 인스턴스 객체를 Pytest의 각 테스트에 의존성 주입해주도록 fixture 설정을 하는 거예요.

FastAPI 애플리케이션 서버의 인스턴스 객체는 라우터를 다루는 객체를 router 속성으로 제공해요. 경로(Router)에는 이름을 지어줄 수 있는데, 우리는 앞서 login 종단점을 라우터에 등록할 때 login이라는 이름을 붙여줬어요. URL 경로는 변경되어도 해당 경로의 이름이 고유하게 유지된다면 우리는 이름으로 URL 경로를 url_path_for() 메서드로 편하고 일관되게 가져올 수 있습니다.

app.router.url_path_for("login")

url_path_for() 로 가져온 URL 문자열을 사용해 login API를 HTTP Post 메서드로 호출합니다. TestClient의 인스턴스 객체인 client는 각 HTTP method에 대응하는 API를 method명과 동일하게 제공해요. HTTP Post는 client.post(), HTTP Get은 client.get() 이죠.

login API는 HTTP Post method로 HTTP 요청을 받는데, 계정명과 비밀번호를 전달받아야해요. 이 정보는 LoginSchema 로 정리해 받죠. FastAPI 종단점은 별도로 설정하지 않으면 컨텐츠형을 JSON 으로 간주해요. 그래서 계정명과 비밀번호를 JSON으로 전달해야 하죠.

def test_login_successfully(client: TestClient):
    payload = {"username": "puddingcamp", "password": "PuddingCamp2024"}
    res = client.post(app.router.url_path_for("login"), json=payload)
    assert res.status_code == status.HTTP_200_OK

Python 사전형(dict) 객체를 전달하면 TestClient가 알아서 JSON 값으로 직렬화(serialization)해줘요. 다만, json 키워드 인자로 전달해야 하는데, data 키워드 인자로 전달하면 FormData로 전달되어서 HTTP Status code인 422 응답을 받게 됩니다.

계정명과 비밀번호가 유효하면 HTTP Status code로 200을 받으니, 테스트 코드도 같은 검증 방식을 따르도록 구현했어요.

SQLAlchemy 설정

패키지 설치

여태까지는 UserService가 인증 정보를 갖고 있는 임시 구현으로 사용자 정보를 다뤘어요. 이제 RDBMS에 영속 데이터로 데이터를 관리할게요.

Python으로 구현된 걸출한 ORM이 몇 가지 있는데,

Django

외 환경에서 가장 많이 애용되는 SQLAlchemy를 우리 프로젝트에 사용할게요. SQLAlchemy는 비동기로 다룰려면

greenlet

이 필요하니 이것도 설치해야 해요.

poetry add sqlalchemy greenlet

RDBMS는 가벼운 RDBMS인 SQLite를 사용할 건데, SQLite도 비동기로 다룰 것이므로 비동기로 동작하는 드라이버가 필요해요. 그래서 aiosqlite 패키지를 설치합니다.

poetry add aiosqlite

SQLAlchemy 설정

이제 SQLAlchemy를 설정해볼까요?

SQLAlchemy로 데이터베이스에 접근해 데이터를 다루려면 접속할 데이터베이스에 맞는 데이터베이스 엔진 객체를 생성하고, 데이터베이스 엔진 객체를 기반으로 데이터베이스 통신(Transaction) 단위인 세션 객체를 만들어 사용해요. 다시 말해 엔진은 실제 데이터베이스에 연결을 하지 않고, 연결에 필요한 상태를 관리하고요. 실제 데이터베이스에 접속하는 건 세션 객체에서 이뤄지는 거죠.

먼저 엔진부터 만들겠습니다. app.py 파일이 있는 경로에 db.py 파일을 만들고, 다음 코드를 작성하세요.

# pudding_todo/db.py

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.pool import NullPool

engine = create_async_engine(
    "sqlite+aiosqlite:///./pudding_todo.db",
    future=True,
    echo=True,
    poolclass=NullPool,
)

비동기 엔진을 만들어야 하므로 create_async_engine() 함수를 사용하고요. 접속할 데이터베이스 종류(dialect)와 드라이버 정보를 포함하여 URI 형태의 DSN(Data Source Name)을 인자로 전달합니다. SQLite 제약 상 연결 풀(Connection Pool)을 사용하지 않을 것이므로 연결 풀 클래스는 NullPool을 전달하고요.

이번엔 데이터베이스에 접근할 때마다 사용할 세션 객체를 만들게요. 접근할 때마다, 라는 표현에 유의하세요. 웹 애플리케이션 서버가 데이터베이스에 접근하는 시간은 종단점이 시작되고 종료될 때까지입니다. 이 종단점은 Python의 함수지요. 그 말은 종단점 함수가 호출될 때 데이터베이스 세션을 시작하고, 종단점 함수가 종료될 때 세션을 닫는다는 의미입니다.

설명은 복잡한데 코드는 간단해요.

# pudding_todo/db.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker


engine = create_async_engine( ... )


async def use_session():
    session_class = async_sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=False, autoflush=False,
    )
    async with session_class() as session:
        yield session

세션 메이커로 세션에 사용할 세션 클래스를 만들고, 이 클래스의 인스턴스 객체를 생성해 사용합니다. 그런데 세션 객체는 컨텍스트 매니저(Python Context Manager)로 세션 접속과 종료를 편하게 하도록 해요. 그래서 async with session_class() as session:과 같이 세션 객체를 만들었어요. with 컨텍스트에서 벗어나면 세션 종료(session.close())도 이뤄집니다.

use_session() 함수(coroutine)는 비동기 세션(AsyncSession) 객체를 반환하는, 비동기 발생자(AsyncGenerator)예요. anext() 함수를 사용해야 하죠.

session = await anext(use_session())

좀 번거로워 보이네요. 하지만 괜찮아요. FastAPI가 대신 처리해서 세션 객체만 사용하도록 해주거든요.

from fastapi import Depends

async def some_endpoint(session = Depends(use_session)):
    await session.commit()

데이터베이스 세션 의존성 주입

인증 처리에서 데이터베이스에 접속해 데이터를 다루는 건 UserService이니 세션은 UserService에 주입해주겠습니다.

# pudding_todo/apps/account/services.py

from fastapi import Depends
from pudding_todo.db import use_session

class UserService:
    def __init__(self, session = Depends(use_session)) -> None:
        self.session = session

    async def authenticate(self, username: str, password: str) -> User | UnauthenticatedUser:
        if self._data.get(username) != password:
            return UnauthenticatedUser()
        return User(username=username, password=password)

이 코드엔 좀 찝찝한 점이 있습니다. 바로 생성자의 인자로 세션을 받는데, FastAPI가 제공하는 Depends() 함수가 반환하는 값을 기본값 인자로 받는 거죠. Depends() 함수는 Any 를 반환한다고 자료형 각주되어 있어서 코드 에디터가 자료형 추론을 제대로 하지 못하고요. 세션을 인자로 받지 않으면 UserService의 인스턴스 객체를 생성하지 못하게 하는 의도가 의존 관계에 드러나지 않아요.

이 문제를 FastAPI는 Annotaed로 대응할 수 있도록 하고 있어요. Annotated를 사용해서 use_session()의 의존성을 자료형 각주로 표현해보죠.

# pudding_todo/db.py

from typing import Annotated

from fastapi import Depends

async def use_session():
    session_class = async_sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=False, autoflush=False,
    )
    async with session_class() as session:
        yield session


DbSessionDep = Annotated[AsyncSession, Depends(use_session)]

Annotated로 만든 DbSessionDep 객체를 자료형 각주에 사용하면 됩니다.

# pudding_todo/apps/account/services.py

from fastapi import Depends
from pudding_todo.db import DbSessionDep

class UserService:
    def __init__(self, session: DbSessionDep) -> None:
        self.session = session

    async def authenticate(self, username: str, password: str) -> User | UnauthenticatedUser:
        if self._data.get(username) != password:
            return UnauthenticatedUser()
        return User(username=username, password=password)

login 종단점 함수가 UserService를 사용하니까, 즉, 의존하니까 UserService도 같은 방식으로 의존성을 주입하겠습니다. account 앱의 models.py 파일이 있는 곳에 deps.py 파일을 만들고 다음 코드를 작성하세요.

from typing import Annotated

from fastapi import Depends

from .services import UserService

UserServiceDep = Annotated[UserService, Depends(UserService)]

그 다음엔 UserServiceDep을 login 종단점 함수에 각주하여 의존성을 FastAPI가 주입하도록 변경할게요.

from fastapi import APIRouter, HTTPException, status

from .models import User
from .schemas import LoginSchema
from .deps import UserServiceDep

router = APIRouter()


@router.post("/login", name="login")
async def login(payload: LoginSchema, service: UserServiceDep) -> User:
    user = await service.authenticate(payload.username, payload.password)
    if not user.is_authenticated:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    
    return user

한결 보기 깔끔해졌습니다.

SQLModel로 Pydantic 모델을 SQLAlchemy ORM으로 사용

개요

SQLAlchemy의 ORM 모델과 Pydantic 모델은 서로 다른 자료구조예요. FastAPI는 Pydantic으로 데이터를 다루므로, SQLAlchemy로 정의한 모델의 데이터는 Pydantic 모델로 변환해야 해요. SQLAlchemy로 데이터베이스에 저장하려면 마찬가지로 Pydantic 객체의 데이터를 SQLAlchemy 모델로 변환해야하죠. 꽤 번거롭습니다.

이걸 하나로 통합한 도구가 바로 SQLModel 이예요. FastAPI를 작성한 개발자가 만든 도구인데, Pydantic 모델을 작성하듯이 모델을 만들면 SQLAlchemy ORM 모델의 객체로도 사용할 수 있어요.

설치해볼까요?

poetry add sqlmodel

SQLModel로 Pydantic 기반 ORM 모델 만들기

SQLModel을 이용해 기존 User 모델을 변경하는 건 간단해요.

from typing import Optional

from starlette.authentication import BaseUser
from sqlmodel import Field, SQLModel, String


class User(BaseUser, SQLModel, table=True):
    id: int = Field(
        primary_key=True,
        nullable=False,
        sa_column_kwargs={"autoincrement": True},
    )
    username: str = Field(unique=True)
    password: str = Field(nullable=True, default=None, exclude=True)

Pydantic의 BaseModel 대신 SQLModel의 SQLModel을 상속받고요. 데이터베이스 테이블에 매핑된다는 걸 table=True 인자로 설정합니다. 나머지는 Pydantic의 모델과 거의 동일하죠? SQLAlchemy 모델에 필요한 정보는 sa_ 접두사가 붙은 인자를 사용하면 돼요.

sa_type : SQLAlchemy의 필드 자료형
sa_column_args : SQLAlchemy의 필드 객체의 위치 인자
sa_column_kwargs : SQLAlchemy의 필드 객체의 키워드 인자
sa_column : SQLAlchemy의 Column 객체. 이 인자를 지정하는 경우 sq_type 등 일부 Field() 함수의 인자는 사용하지 못함.

id 필드도 추가했는데요. 기본키(Primary Key) 용도예요.

데이터베이스 기반 테스팅 확인을 위한 사전 작업

데이터베이스 설정과 연동을 했으니 실제로 잘 동작하는지 테스팅을 수행하여 확인해볼게요.

테스팅에서도 데이터베이스 연결을 하며, 사실상 매번 사용하므로 모든 테스트에서 의존성 주입받을 수 있게 conftest.py 에 데이터베이스 관련 설정을 할게요. tests 디렉터리에 conftest.py 파일을 만들고 다음 코드를 작성합니다.

# tests/conftest.py

from typing import Generator

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.pool import NullPool
from sqlmodel import SQLModel

from pudding_todo.app import create_app


@pytest.fixture()
async def db_session():
    engine = create_async_engine(
        "sqlite+aiosqlite:///file:testing.db?mode=memory&cache=shared&uri=true",
        future=True,
        echo=False,
        poolclass=NullPool,
    )

    async with engine.connect() as conn:
        await conn.begin()
        await conn.run_sync(SQLModel.metadata.drop_all)
        await conn.run_sync(SQLModel.metadata.create_all)
   
        session_class = async_sessionmaker(
            conn, class_=AsyncSession, expire_on_commit=False, autoflush=False,
        )
        async with session_class() as session:
            yield session
            
        await conn.run_sync(SQLModel.metadata.drop_all)
        await conn.rollback()

    await engine.dispose()


@pytest.fixture(autouse=True)
def fastapi_app(db_session: AsyncSession) -> Generator[FastAPI, None, None]:
    app = create_app()
    app.dependency_overrides[use_session] = lambda: db_session
    yield app


@pytest.fixture()
def client(fastapi_app: FastAPI) -> Generator[TestClient, None, None]:
    with TestClient(fastapi_app) as test_client:
        yield test_client

작성하는 김에 TestClient fixture도 conftest로 옮기고, 테스트에서 사용할 전용 FastAPI 애플리케이션 인스턴스 객체도 만들어 사용하도록 했어요. app.dependency_overrides[use_session] = lambda: db_session 이 코드는 잠시 무시하세요. 😅 그럼 FastAPI 애플리케이션을 생성하는 create_app() 함수도 작성하겠습니다.

# pudding_todo/app.py

from contextlib import asynccontextmanager

from fastapi import FastAPI

from pudding_todo.apps.account.router import router as account_router
from pudding_todo.apps.common.router import router as common_router

from .authentication import fastapi_users, auth_backend
from .db import engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield

    await engine.dispose()


def create_app() -> FastAPI:
    app = FastAPI(lifespan=lifespan)
    app.include_router(common_router)
    app.include_router(account_router)

    return app

app = create_app()

기존 코드가 create_app() 함수 안으로 들어갔고, lifespan() 함수가 새로 등장했네요. lifespan() 함수는 FastAPI 애플리케이션의 생애주기에 관여하는데, FastAPI 애플리케이션 서버가 구동된 직후에 동작할 코드를 yield문 전에 작성하고, 구동을 종료한 직후에 동작할 코드를 yield문 후에 작성해요.

다시 conftest.py로 돌아와서. 테스팅 상황에서는 데이터베이스 동작을 빠르게 하도록 SQLite를 메모리 모드로 설정했어요. 이러면 SQLite 데이터베이스가 파일로 만들어지지 않고 메모리에 만들어져서 조금이라도 더 빠르고, 테스팅이 끝난 후 테스팅에서 생성된 SQLite 파일을 지우지 않아도 되어 편리해요.

테스팅에서는 매번 데이터베이스를 생성하고 제거하므로 SQLAlchemy ORM 모델에 대응하는 테이블도 만들어야 해요. 이 작업은 테스팅에서만 이뤄지죠.

자, 그럼 테스팅에서 사용할 사용자 계정을 하나 만들어볼까요? conftest.py 에 다음 코드를 추가할게요.

# tests/conftest.py

from pudding_todo.apps.account.models import User


@pytest.fixture()
async def valid_user(db_session):
    password = "PuddingCamp2024"
    user = User(username="puddingcamp", password=hashed_password)
    db_session.add(user)
    await db_session.commit()
    yield user

이렇게 해서 이 conftest.py 파일이 있는 이하 경로에 있는 테스트에 valid_user를 의존성으로 명기하면, 즉 인자로 선언하면 그때마다 puddingcamp 계정명을 갖는 User 객체를 생성하고 데이터베이스에 저장한 후 이 객체를 인자로 전달받아 사용할 수 있어요.

이 코드는 4cb73d1 커밋에 있어요.

데이터베이스 기반 테스팅 확인

드디어 테스팅을 위한 사전 작업을 마쳤어요. 이번엔 기존 테스트 코드를 살짝 고칠게요.

# tests/apps/account/test_auth_endpoints.py

def test_login_successfully(client: TestClient, valid_user: User):
    payload = {"username": valid_user.username, "password": "PuddingCamp2024"}
    res = client.post(app.router.url_path_for("login"), json=payload)
    assert res.status_code == status.HTTP_200_OK

기존 test_login_successfully() 테스트 함수와 거의 동일해요. conftest.py 에 Pytest Fixture로 정의한 valid_user를 인자로 지정해 User 객체를 주입받는 조치가 추가됐어요. 그리고 payload의 username을 이 valid_user 객체의 username을 사용했습니다. 이 테스트를 수행하면 별 문제없이 통과해요. 재밌죠? 데이터베이스가 없이 작성한 테스트인데, 데이터베이스 연동 추가한 구현에 대한 테스트 코드가 실질적 변경없이 그대로 수행되잖아요. 🙂

만약 test_login_successfully() 함수에 valid_user Fixture를 주입하지 않으면 이 Fixture는 실행되지 않아서 puddingcamp 이용자가 데이터베이스에 저장되지 않아요. @pytest.mark.usefixtures 장식자를 사용하는 방법도 있는데, 이번 편에서는 다루지 않을게요.

푸딩캠프 뉴스레터를 구독하면 학습과 성장, 기술에 관해 요약된 컨텐츠를 매주 편하게 받아보실 수 있습니다.

이전 컨텐츠

다음 컨텐츠