할 일(Todo) 생성

시리즈, 입문

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

#fastapi #sqlmodel #sqlalchemy #UTC #timezone #relationship

할 일(ToDo) 모델링

(1) 기본 모델링

할 일 관리 앱 기획을 기반으로 할 일(ToDo)을 다음과 같이 모델링했어요.

# pudding_todo/apps/todo/models.py

from datetime import datetime, UTC
from typing import Optional

from pydantic import AwareDatetime
from sqlmodel import Field, SQLModel, Text, func

class Todo(SQLModel, table=True):
    id: int = Field(
        primary_key=True,
        nullable=False,
        sa_column_kwargs={"autoincrement": True},
    )
    name: str
    description: Optional[str] = Field(
	    nullable=True,
	    default=None,
	    sa_type=Text,
	)
   
    created_at: AwareDatetime = Field(
        default_factory=lambda: datetime.now(UTC),
        nullable=False,
        sa_column_kwargs={
            "server_default": func.now(),
        },
    )
    updated_at: AwareDatetime = Field(
        default_factory=lambda: datetime.now(UTC),
        nullable=False,
        sa_column_kwargs={
            "server_default": func.now(),
            "server_onupdate": func.now(),
        },
    )
    duedate_at: Optional[AwareDatetime] = Field(
        default=None,
        nullable=True,
    )
    completed_at: Optional[AwareDatetime] = Field(
        default=None,
        nullable=True,
    )
    cancelled_at: Optional[AwareDatetime] = Field(
        default=None,
        nullable=True,
    )

id 필드를 보면 Field() 함수에 sa_column_kwargs 인자를 사용하고 있어요. SQLAlchemy의 Column 객체에 사용할 키워드 인자를 지정하는 거죠. SQLModel의 Field() 함수는 Pydantic의 Field() 함수와 거의 동일한데, Pydantic 필드를 SQLAlchemy ORM 필드로도 다룰 수 있는 객체를 만든다는 점에서 달라요. Pydantic에 있는 FieldInfo 클래스를 상속받아 확장했지요. 그래서 반드시 SQLModel에 있는 Field() 함수를 사용해야 해요.

sa_ 접두사가 붙은 키워드 인자는 SQLAlchemy 관련 인자로 총 네 가지가 있어요. SQLAlchemy의 필드 자료형을 지정하는 데 사용하는 sa_type인자는 빈번하게 사용하죠.

일시 정보를 다루는 필드들을 보면 datetime 클래스가 아닌 AwareDateTime을 사용하는 점이 눈길을 끕니다. 국제화 시대에 발맞춰 시간대(Timezone) 정보가 있는 datetime 객체로 다루려고요. 그래서 자료형 각주는 Pydantic에서 제공하는 AwareDateTime으로 지정했지요.

그런데 SQLite 데이터베이스에서는 일시 정보가 없는 Naive DateTime 값을 다루기 때문에 Pydantic 필드 유효성 검사(validator)에서 문제가 발생합니다. 데이터베이스로부터 받은 Naive DateTime을 SQLAlchemy에서 Aware DateTime으로 다루도록 해야 해요.

(2) SQLAlchemy-Utc

이 상황에 대응하는 가장 쉬운 방법은 SQLAlchemy-Utc 패키지를 사용하는 거예요. 그러면 PostgreSQL에서 일시 정보를 다루는 것처럼 SQLite에서도 Aware DateTime을 다루거든요.

poetry add sqlalchemy-utc

그런 다음 Field()에 sa_type 인자로 SQLAlchemy-Utc의 UtcDateTime 필드 자료형을 지정해줍니다.

created_at: AwareDatetime = Field(
	default_factory=lambda: datetime.now(UTC),
	nullable=False,
	sa_type=UtcDateTime,
	sa_column_kwargs={
		"server_default": func.now(),
	},
)
updated_at: AwareDatetime = Field(
	default_factory=lambda: datetime.now(UTC),
	nullable=False,
	sa_type=UtcDateTime,
	sa_column_kwargs={
		"server_default": func.now(),
		"server_onupdate": func.now(),
	},
)
duedate_at: Optional[AwareDatetime] = Field(
	default=None,
	nullable=True,
	sa_type=UtcDateTime,
)
completed_at: Optional[AwareDatetime] = Field(
	default=None,
	nullable=True,
	sa_type=UtcDateTime,
)
cancelled_at: Optional[AwareDatetime] = Field(
	default=None,
	nullable=True,
	sa_type=UtcDateTime,
)

여기까지 진행한 코드 커밋 : 389f6f0

할 일 생성

단순하게나마 할 일 모델을 구현했으니 이 모델로 데이터 저장이 잘 되는지 확인해볼게요.

(1) TodoCreateSchema

할 일 데이터를 생성할 때 사용자가 입력하는 정보는 두 가지지요. 할 일 이름과 설명. 설명은 적어도 되고 안 적어도 돼요. 이러한 데이터를 받는 데 사용할 스키마를 만들게요.

# pudding_todo/apps/todo/schemas.py

from typing import Optional
from pydantic import BaseModel, Field


class TodoCreateSchema(BaseModel):
    name: str
    description: Optional[str] = Field(default=None)

데이터베이스 ORM 모델이 아니라는 걸 명확히 하려고 Pydantic의 BaseModel 클래스를 상속하고 Field() 함수를 사용했어요.

(2) 할 일을 데이터베이스에 저장

TodoService에 create() API 구현

TodoCreateSchema를 적재 데이터(Payload) 삼아 할 일 데이터를 데이터베이스에 저장하는 create() API를 TodoService 에 구현할게요.

# pudding_todo/apps/todo/services.py

from pudding_todo.db import DbSessionDep
from sqlalchemy.ext.asyncio import AsyncSession

from .models import Todo
from .schemas import TodoCreateSchema

class TodoService:
    session: AsyncSession

    def __init__(self, session: DbSessionDep):
        self.session = session

    async def create(self, payload: TodoCreateSchema) -> Todo:
        todo = Todo(**payload.model_dump(mode="python"))
        self.session.add(todo)
        await self.session.commit()
        return todo

Pydantic으로 만든 Schema(또는 Pydantic 모델)로 객체를 만들 때 데이터 유효성을 검사하며 객체를 생성하려면 model_validate() 클래스 메서드를 사용해요. Todo 모델은 SQLAlchemy ORM 모델이기도 하지만, Pydantic 모델이기도 하잖아요. 기왕이면 Todo 인스턴스 객체를 만들 때에도 데이터 유효성을 검사하면 좋겠지만, Todo 모델의 새 객체를 만들어 데이터베이스에 저장할 때엔 유효성 검사를 하지 않는 게 나아요.

우리가 할 일을 저장할 때 입력하는 실제 정보는 할 일 이름(name)과 설명(description)이죠. 그런데 ORM 모델로써 Todo엔 기본키인 id 같은 정보가 필요해요. 이 기본키의 값은 애플리케이션 계층, 즉 Python에서 생성하는 정보가 아니라 데이터베이스 생성하고 관리하는 값이죠. 다시 말해 Todo 인스턴스 객체를 만드는 시점에는 id 필드값이 없어요. 하지만 Pydantic Schema로써 Todo의 id 필드는 필수 항목이예요. Optional이 아니며, 그렇다고 값이 지정되지 않으면 기본값이 부여되는 default 필드 인자를 사용해서도 안 되지요.

todo = Todo(
	name=payload.name,
	description=payload.description,
)

따라서 이와 같이 유효성 검사 없이 인스턴스 객체를 만들어야 해요. 값의 유효성 검증은 TodoCreateSchema가 이미 했다고 전제하는 거죠.

model_dump() 메서드는 Pydantic v1에서 dict() 메서드를 대체하는 메서드예요. dict()는 v2부터 더이상 쓰지 않거든요(deprecated). mode="python" 인자를 사용하면 Python 자료형, 각 필드에 각주한 Python 자료형으로 만들어진 객체를 담은 dict 객체를 반환해요. mode="json" 인자로 전달하면 JSON에서 다루는 자료형으로 변환하고요. 만약 필드가 JSON 필드라면 mode="json"으로 직렬화한 데이터를 사용하면 돼요.

테스트 코드로 동작 검증

서비스 계층은 종단점(진입점) 계층에서 유효성을 거친 데이터를 넘겨받아 사용해요. 테스트 코드에도 이 의도를 드러내야겠죠. 빈번하게 사용되는 적재 데이터이니 Pytest Fixture로 각 테스트에서 주입받아 사용하도록 작성할게요.

# tests/apps/todo/test_todo_service.py

import pytest

from sqlalchemy.ext.asyncio import AsyncSession

from pudding_todo.apps.todo.models import Todo
from pudding_todo.apps.todo.schemas import TodoCreateSchema
from pudding_todo.apps.todo.services import TodoService


@pytest.fixture()
def todo_service(db_session: AsyncSession) -> TodoService:
    return TodoService(db_session)


@pytest.fixture()
def payload() -> TodoCreateSchema:
    return TodoCreateSchema.model_validate({
	    "name": "Test Todo",
	})

앞서 설명드렸듯이 model_validate() 클래스 메서드로 데이터 유효성 검사를 해요. 나중에 TodoCreateSchema 스키마의 필드 구성이 달라지면 테스트를 수행하자마자 ValidationError가 발생해서 빠르게 문제 가능성이 있는 부분을 찾을 수 있겠지요?

# tests/apps/todo/test_todo_service.py

async def test_create(payload, todo_service):
    todo = await todo_service.create(payload)
    
    assert isinstance(todo, Todo)
    assert todo.name == payload.name

간단한 동작이라 가뿐하게 테스트 통과!

여기까지 진행한 코드 커밋 : 45e8902

(3) 사용자 데이터와 관계 짓기, User Relationship

나 혼자만 사용하면 모를까, 여러 사용자가 사용하는 서비스라면 각 할 일 마다 사용자를 구분해줘야 해요. 사용자 정보는 User 모델로 다루니, Todo 모델에 사용자 정보를 저장하지 말고, User 모델과 Todo 모델 간 관계를 맺어주는 방식으로 데이터를 다룰게요. 이를 데이터베이스 정규화라고 해요.

# pudding_todo/apps/todo/models.py

from typing import Optional, TYPE_CHECKING
from sqlmodel import Field, SQLModel, Text, func, Relationship


if TYPE_CHECKING:
    from pudding_todo.apps.account.models import User


class Todo(SQLModel, table=True):
    ...
   
    user_id: int = Field(nullable=False, foreign_key="user.id")
    user: "User" = Relationship(
        back_populates="todos",
        sa_relationship_kwargs={"lazy": "selectin"},
    )
	...

두 개 필드를 추가했어요. user_id는 실제 데이터베이스 테이블에 열(Column)로 추가돼요. 우리는 각 모델에 대응하는 데이터베이스 테이블의 기본키로 정수형 값을 사용하는 id 필드를 사용하고, 이를 외래키(ForeignKey)로 연결하는 경우 연결 대상 테이블 이름에 _id라 접미사를 붙이는 필드명이(Column 명) 관례여서 user_id라고 이름 지었습니다.

user 필드는 SQLAlchemy ORM의 기능인데, 관계를 맺고 있는 대상의 데이터를 ORM 객체로 접근할 수 있어요. 예를 들어, user_id가 1인 User 모델의 데이터를 가리키고 있다면, 데이터베이스에서 기본키 값이 1인 데이터를 가져와서 User 모델의 인스턴스 객체로 생성하고, 이를 Todo 인스턴스 객체의 필드인 user에 할당하는 거죠. 다음은 실제 코드에 가까운 의사 코드 예시예요.

obj = Todo(
	user_id=1,
	name="PuddingCamp",
)
session.add(obj)
session.commit()

todo = select(Todo).where(Todo.id == 980665)
assert todo.user.id == obj.id
assert todo.user_id == obj.id

관계(Relationship) 모델인 User의 데이터는 Todo의 데이터를 가져올 때 함께 가져올 수도 있고, 나중에 필요할 때 가져오게 할 수도 있어요.

실제 코드나 마찬가지인 앞선 코드를 의사 코드라고 한 이유는, 실제로는 동작하지 않기 때문이예요. 오류가 발생하거든요.

    other = self.mapper.get_property(key, _configure_mappers=False)
.venv/lib/python3.12/site-packages/sqlalchemy/orm/mapper.py:2511: in get_property
    raise sa_exc.InvalidRequestError(
E   sqlalchemy.exc.InvalidRequestError: Mapper 'Mapper[User(user)]' has no property 'todos'.  If this property was indicated from other mappers or configure events, ensure registry.configure() has been called

이 오류는 Todo 모델이 User 모델을 가리키면서 back_populates를 todos로 지정했는데, 정작 User 모델엔 todos라 명명된 모델 관계 정보가 없기 때문이죠.

class Todo(SQLModel, table=True):
    ...
   
    user_id: int = Field(nullable=False, foreign_key="user.id")
    user: "User" = Relationship(
        back_populates="todos",
        sa_relationship_kwargs={"lazy": "selectin"},
    )
	...

Relationship() 함수의 back_populates 인자는 관계 맺은 대상 모델에서 사용할 참조(reference) 필드명이예요. 사용자는 할 일을 여러 개 만들고 관리하잖아요. 사용자 1 : 할 일 다수(N). 이런 관계를 할 일 모델 기준으로 N 대 1 관계라고 해요. Todo는 User 하나를 가리키는 거죠. User 모델 입장에서(?) 보면 나 하나에 대해 Todo 여러 개가 매달려 있는 거죠. “내 할 일 모여라!”라는 얘기는 user_id 값이 1인 Todo 모두 모이라는 말이고요. 즉, User 모델을 가리키는 Todo 모델의 데이터를 걸러낼 수 있어요. 그래서 별도 데이터베이스 열을 만들지 않고 논리적인, 다르게 표현하면 Python 영역에서 동작하는 ORM의 속성으로써 User 모델을 가리키는 Todo 모델 데이터들을 접근할 수 있는 기능을 제공할 수 있어요. 이것도 Relationship() 함수를 이용해서 선언해주면 돼요.

# pudding_todo/apps/account/models.py

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from pudding_todo.apps.todo.models import Todo, TodoGroup

class User(BaseUser, SQLModel, table=True):
    ...
    todos: list["Todo"] = Relationship(
        back_populates="user",
        sa_relationship_kwargs={"lazy": "selectin"},
    )

Todo 모델에서 User 모델을 향해 관계를 맺어주는 것의 반대 방향이죠. User 모델을 가리키는 건 Todo들이니까 list 자료형으로 각주했고(list["Todo"]), Todo 모델에서 User 모델 자신을 향하는 관계 필드 이름이 user이므로 back_populates="user" 인자를 전달했어요.

TodoService에도 User 적용

Todo에 반영할 사용자 정보를 Todo 생성 API에 추가합니다.

# pudding_todo/apps/todo/services.py

class TodoService:
    session: AsyncSession

    def __init__(self, session: DbSessionDep):
        self.session = session

    async def create(self, user_id: int, payload: TodoCreateSchema) -> Todo:
        todo = Todo(user_id=user_id, **payload.model_dump(mode="python"))
        self.session.add(todo)
        await self.session.commit()
        return todo

payload는 사용자가 할 일을 생성할 때 입력하는 정보를 담습니다. 여기에 사용자 정보(user_id)를 담으면 의도하지 않은 동작이 발생하게 돼요. id가 1인 사용자가 user_id를 3으로 지정해서 id가 3인 이용자인 척 할 일을 만들 수 있게 되거든요. 그래서 Todo를 생성하는 사용자는 로그인 한 사용자의 것으로 제한해야 해요. 그래서 user_id를 payload와 분리하여 인자로 받았어요.

조금 더 안전하게 구현한다면 user_id를 토대로 유효한 사용자인지 검사해야 해요. 가령, 계정을 활성화하거나 비활성화하는 정보가 있다면, 비활성화 이용자는 Todo를 생성하지 못하게 막는 거죠. 우리의 할 일 관리 앱엔 그런 기능이 없으니 따로 구현하진 않을게요.

conftest.py 에 만든 valid_user를 사용해 사용자 정보를 추가한 create() API를 테스트해서 동작을 검증해봅니다.

# tests/apps/todo/test_todo_service.py

async def test_create(payload, todo_service, valid_user):
    todo = await todo_service.create(valid_user.id, payload)
    
    assert isinstance(todo, Todo)
    assert todo.name == payload.name
    assert todo.user.id == valid_user.id

물론 통과하지요. 🥳

여기까지 진행한 코드 커밋 : 389f6f0

(4) 할 일 생성 Endpoint

할 일 생성을 HTTP Request로 요청받을 진입점(종단점) 함수는 아주 단순한 구조예요. HTTP Request로 전달받은 적재 데이터(payload), 그리고 진입점 API에 접근한 사용자 정보를 Todo Service의 create() 메서드에 전달하기만 하거든요. 적재물 데이터가 유효한지 검증하는 건 FastAPI가 Pydantic 모델인 TodoCreateSchema로 알아서 해주고, 로그인한 사용자만 접근 가능하며 접근한 사용자가 누구인지를 처리하는 건 CurrentUserDep로 처리하죠.

# pudding_todo/apps/todo/endpoints.py

from fastapi import APIRouter, status

from pudding_todo.authentication import CurrentUserDep

from .deps import TodoServiceDep
from .models import Todo
from .schemas import TodoCreateSchema

router = APIRouter(prefix="/todos")


@router.post(
    "/",
    response_model=Todo,
    name="create-todo",
    status_code=status.HTTP_201_CREATED,
)
async def create_todo(
    payload: TodoCreateSchema,
    user: CurrentUserDep,
    service: TodoServiceDep,
) -> Todo:
    todo = await service.create(user.id, payload)
    return todo

진입점 함수 자체엔 HTTP와 관련된 코드가 직접적으로 드러나지 않아서 한결 읽기에 명료하죠?

# tests/apps/todo/test_todo_endpoint.py

import pytest

from fastapi import status
from fastapi.testclient import TestClient

from pudding_todo.app import app
from pudding_todo.apps.account.models import User
from pudding_todo.apps.todo.schemas import TodoCreateSchema
from pudding_todo.authentication import auth_backend


@pytest.fixture()
def payload() -> dict:
    schema = TodoCreateSchema.model_validate({
        "name": "Test Todo",
    })
    return schema.model_dump(mode="json")


async def test_create_todo(payload: dict, client: TestClient, valid_user: User):
    url = app.router.url_path_for("create-todo")
    token = await auth_backend.get_strategy().write_token(valid_user)
    cookies = {auth_backend.transport.cookie_name: token}

    res = client.post(url, json=payload, cookies=cookies)
    assert res.status_code == status.HTTP_201_CREATED
    data = res.json()
    assert data["name"] == payload["name"]
    assert data["user_id"] == valid_user.id

로그인하지 않은 사용자는 할 일을 생성하지 못하는 테스트도 무난히 통과합니다.

async def test_cannot_create_todo_without_auth(payload, client, valid_user):
    url = app.router.url_path_for("create-todo")

    res = client.post(url, json=payload)
    assert res.status_code == status.HTTP_401_UNAUTHORIZED

여기까지 진행한 코드 커밋 : 61ff4d2

할 일 그룹 구현

(1) 모델링

모든 할 일 그룹은 사용자와 관계를 맺고 있어요. 사용자 한 명이 할 일 그룹을 여러 개 만들 수 있으니 할 일 그룹과 사용자 간 관계는 N 대 1이지요.

# pudding_todo/apps/todo/models.py

from typing import TYPE_CHECKING
from sqlmodel.main import declared_attr, UniqueConstraint

if TYPE_CHECKING:
    from pudding_todo.apps.account.models import User

class TodoGroup(SQLModel, table=True):
    id: int = Field(
        primary_key=True,
        nullable=False,
        sa_column_kwargs={"autoincrement": True},
    )
    name: str = Field()

    user_id: int = Field(nullable=False, foreign_key="user.id")
    user: "User" = Relationship(
        back_populates="todo_groups",
        sa_relationship_kwargs={"lazy": "selectin"},
    )
    
    
    @declared_attr
    @classmethod
    def __tablename__(cls) -> str:
        return "todo_group"

    @declared_attr
    @classmethod
    def __table_args__(cls) -> tuple:
        return (
            UniqueConstraint(
	            "user_id",
	            "name",
	            name="unique_user_group_name",
			),
        )

못보던 게 있네요. __table_args__ 속성은 SQLAlchemy의 선언형 테이블의 설정 정보를 다뤄요. SQLAlchemy이 참조하는 __tablename__ 속성과 마찬가지로 멤버 변수로써 Python 속성으로 선언해도 되고, @declared_attr 클래스를 사용해 클래스 메서드를 Python 속성으로 선언해도 돼요. 저는 클래스 메서드로 선언하는 방식을 선호하죠.

UniqueConstraint는 고유 속성을 선언하는 데 사용하는데, 모델 필드 하나에만 고유 속성을 주는 경우엔 unique=True 인자를 전달하고, 여러 필드를 묶어서 고유 속성을 주는 경우엔 UniqueConstraint 클래스를 쓰죠.

사용자 별 할 일 그룹 이름을 고유하게 할 것이므로 user_id와 name 필드를 묶어서 고유 속성을 부여합니다.

TodoGroup이 User 모델에 대해 관계를 맺으니 User 모델에도 관계 필드를 추가해야 할게요.

if TYPE_CHECKING:
    from pudding_todo.apps.todo.models import TodoGroup


class User(BaseUser, SQLModel, table=True):
    ...

    todo_groups: list["TodoGroup"] = Relationship(
        back_populates="user",
        sa_relationship_kwargs={"lazy": "selectin"},
    )

(2) Todo 모델과 관계 짓기

각 할 일을 모아놓은 단위, 할 일 그룹은 할 일에 사용자를 관계짓는 것과 개념 상 그리 다르지 않아요. 각 할 일은 할 일 그룹 하나를 가리키거든요.

# pudding_todo/apps/todo/models.py

class Todo(SQLModel, table=True):
    ...
    
    group_id: int = Field(nullable=False, foreign_key=f"{TodoGroup.__tablename__}.id")
    group: TodoGroup = Relationship(
        back_populates="todos",
        sa_relationship_kwargs={"lazy": "selectin"},
    )

역방향도 관계가 있다는 참조를 시켜주고요.

# pudding_todo/apps/todo/models.py

class TodoGroup(SQLModel, table=True):
    ...
    
    todos: list["Todo"] = Relationship(
        back_populates="group",
        sa_relationship_kwargs={"lazy": "selectin"},
    )

    ...

진행한 코드 커밋 : 92136ed

(3) Todo 모델에서 사용자 관계 필드 제거

현재 Todo 모델은 User 모델과 TodoGroup 모델에 대해 각각 N 대 1로 관계를 맺고 있어요. 그리고 TodoGroup 모델은 User 모델에 대해 N 대 1 관계를 맺고 있죠.

Todo ───N:1─── TodoGroup ───N:1─── User
   │                                │
   +──────────────N:1───────────────+

그렇다면 Todo 모델은 User 모델과 관계를 맺을 필요는 없어요. 각 Todo는 사용자를 가리키는 TodoGroup를 가리키고 있기 때문이죠. 그래서 다음과 같이 정규화할 수 있어요.

Todo ───N:1─── TodoGroup ───N:1─── User

Todo 모델에서 user_id 필드를 제거하겠습니다. Todo 모델에서 user_id와 user 필드를 제거하고, User 모델에서는 todos 필드를 제거하세요.

(4) 비즈니스 로직에서 Todo의 user_id 제거

이제 비즈니스 로직에서도 제거합니다. 먼저 Todo Service의 create()의 동작을 변경해야 해요. user_id 인자를 제거하는 거죠.

그 다음엔 create_todo() 종단점 함수에도 변경 사항을 반영해요. Todo Service의 create()는 TodoCreateSchema로 생성된 인스턴스 객체인 payload의 group_id를 Todo와 TodoGroup 모델 관계 맺는 데 사용하는데요. 서비스 계층에선 이 정보가 유효하다고 전제합니다. 그 대신 서비스 계층을 사용하는 종단점 계층에서 유효한 TodoGroup을 전달해야 해요. Todo Service는 Todo에 관한 처리에만 집중하는 거죠.

# pudding_todo/apps/todo/endpoints.py

@router.post(
    "/",
    response_model=Todo,
    name="create-todo",
    status_code=status.HTTP_201_CREATED,
)
async def create_todo(
    payload: TodoCreateSchema,
    user: CurrentUserDep,
    service: TodoServiceDep,
    todo_group_service: TodoGroupServiceDep,
) -> Todo:
    group = await todo_group_service.get_users_group_by_id(user.id, payload.group_id)
    if not group:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
    todo = await service.create(payload)
    return todo

현재 로그인한 이용자의 id 값과 적재 데이터 중 하나인 group_id로 TodoGroup 데이터를 가져와요. 현 로그인한, 즉 HTTP Request가 일어난 세션의 사용자 정보를 기반으로 TodoGroup를 가져오므로 다른 사용자의 TodoGroup을 가져오진 않아요.

# pudding_todo/apps/todo/services.py
from sqlmodel import select

class TodoGroupService:
    ...
    
    async def get_users_group_by_id(self, user_id: int, group_id: int) -> TodoGroup | None:
        stmt = select(TodoGroup).where(TodoGroup.user_id == user_id, TodoGroup.id == group_id)
        result = await self.session.execute(stmt)
        return result.one_or_none()

SQLModel에서 제공하는 select() 함수는 SQLAlchemy에 있는 함수와 사실상 동일한 동작을 해요. SQLModel에 있는 select() 함수의 자료형 각주가 더 친절해서 저는 SQLModel의 select() 함수를 사용해요.

select() 함수는 이름에서 알 수 있듯이 SQL의 select 구문을 생성해요. select() 함수는 전달되는 인자에 따라 SQLAlchemy 1 버전대 API나 2버전대 API를 자동으로 골라 사용하죠. SQL 구문을 다루는 Python 표현식과 SQLAlchemy 세션에 의존하지 않아서 직관성과 사용성 모두 좋아요. 게다가 SQL문과 비슷하게 Python 구문을 작성해서 SQL을 알면 더 직관적이죠.

# SQLAlchemy Select 표현식
select(Model).where(Model.id == 1).order_by(Model.id.asc()).limit(1)

# SQL 문
SELECT * FROM table WHERE id = 1 ORDER BY id ASC LIMIT 1;

SQLAlchemy Session 객체의 execute() 메서드는 Result 객체를 반환해요. 앞서 Select 구문은 기본키 매칭 조건이 있으므로 반환되는 데이터(레코드)는 하나이거나 없지요. 그래서 Result 객체의 one_or_none() 메서드로 데이터 하나를 가져오거나 None을 반환받도록 했어요.

# tests/apps/todo/test_todo_service.py

@pytest.fixture()
async def todo_group(db_session, valid_user) -> TodoGroup:
    service = TodoGroupService(db_session)
    payload = TodoGroupCreateSchema.model_validate({
        "name": "Test Group",
    })
    group = await service.create(valid_user.id, payload)
    return group


@pytest.fixture()
def payload(todo_group) -> TodoCreateSchema:
    return TodoCreateSchema.model_validate({
        "name": "Test Todo",
        "group_id": todo_group.id,
    })


async def test_create_todo(payload, client, valid_user):
    url = app.router.url_path_for("create-todo")
    token = await auth_backend.get_strategy().write_token(valid_user)
    cookies = {auth_backend.transport.cookie_name: token}

    res = client.post(url, json=payload, cookies=cookies)
    assert res.status_code == status.HTTP_201_CREATED
    data = res.json()
    assert data["name"] == payload["name"]
    assert data["group_id"] == payload["group_id"]

문제 없는 상황으로 환경을 조성해 테스트를 수행하면 무난히 통과합니다.

(5) 남의 TodoGroup 을 가져오지 못하는지 테스트

로그인한 이용자의 할 일 그룹만 할 일에 연결하는 구현이 종단점 함수에 있지요. 그러니 이 종단점 함수가 의도대로 동작하는지 테스트로 검증할게요.

먼저 사용자를 하나 더 만들고요.

# tests/conftest.py


@pytest.fixture()
async def valid_user2(db_session, user_manager):
    password = "PuddingCamp2024"
    hashed_password = user_manager.password_helper.hash(password)

    user = User(username="puddingcamp2", hashed_password=hashed_password)
    db_session.add(user)
    await db_session.commit()
    yield user

valid_user2의 할 일 그룹도 테스트 픽스쳐로 만들어요.

# tests/apps/todo/test_todo_endpoint.py

@pytest.fixture()
async def todo_group2(db_session, valid_user2):
    service = TodoGroupService(db_session)
    payload = TodoGroupCreateSchema.model_validate({
        "name": "Test Group",
    })
    group = await service.create(valid_user2.id, payload)
    return group

valid_user가 valid_user2의 할 일 그룹을 자신의 할 일에 사용하는 테스트를 작성하고요.

# tests/apps/todo/test_todo_endpoint.py

async def test_cannot_create_todo_with_others_group(
    payload: dict,
    client: TestClient,
    todo_group2: TodoGroup,
    valid_user: User,
):
    payload["group_id"] = todo_group2.id
    url = app.router.url_path_for("create-todo")
    token = await auth_backend.get_strategy().write_token(valid_user)
    cookies = {auth_backend.transport.cookie_name: token}

    res = client.post(url, json=payload, cookies=cookies)
    assert res.status_code == status.HTTP_404_NOT_FOUND

todo_group2는 valid_user2의 할 일 그룹이예요. valid_user로 할 일을 생성할 때 적재 데이터 중 group_id로 todo_group2의 기본키값을 사용하면 의도대로 HTTP 404 응답을 받아요.

남의 데이터에 접근할 권한이 없는 것이니 HTTP 403 응답을 할 수도 있는데요. 저는 권한이 없는 것인지 데이터가 없는 것인지 판별하지 못하게 HTTP 404 응답을 하는 편이예요.

여기까지 진행한 코드 커밋 : f1a026e

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

이전 컨텐츠

다음 컨텐츠