할 일, 취소 상태로 변경

할 일을 취소하거나 취소 해제하는 기능은 할 일을 완료 처리하거나 완료 해제하는 기능과 거의 동일해요. completed_at 모델 필드 대신 cancelled_at 모델 필드의 값을 다룬다는 점이 다르죠. 그래서 이 부분 구현은 여러분께 과제로 남겨둘게요. 😁

할 일 상태 별 목록 가져오기

할 일을 완료 처리하거나 미완료 처리하는 기능을 구현했으니, 이번엔 완료한 일만, 완료하지 않은 일만 걸러내는 기능을 구현해보아요.

TodoService의 findall() API에 is_completed 필터 추가

지정한 조건으로 할 일을 가져오는 TodoService 의 findall() API에 필터를 하나 추가할게요. 바로 is_completed 로 True 이면 완료한 일만, False 이면 완료하지 않은 일만 걸러내 가져오도록 SQL 질의를 작성하는 데 사용해요.

# pudding_todo/apps/todo/services.py

from sqlmodel import null

class TodoService:
    ...
    
    async def findall(
            self,
            user_id: int,
            *,
            group_id: Optional[int] = None,
            offset: int = 0,
            limit: int = 10,
            is_completed: Optional[bool] = None,
        ) -> Sequence[Todo]:
        stmt = (
            select(Todo)
            .join(TodoGroup)
            .where(TodoGroup.user_id == user_id)
            .offset(offset)
            .limit(limit)
        )

        if group_id:
            stmt = stmt.where(Todo.group_id == group_id)

        if is_completed is True:
            stmt = stmt.where(Todo.completed_at.isnot(null()))
        elif is_completed is False:
            stmt = stmt.where(Todo.completed_at.is_(null()))

        result = await self.session.execute(stmt)
        return result.unique().scalars().all()

완료한 일은 Todo 모델의 completed_at 모델 필드의 값이 None 이 아닌 데이터예요. 데이터베이스에서는 NULL 이 아닌 값이죠. 다음 SQL로 표현돼요.

SELECT * FROM todo WHERE completed_at IS NULL

SELECT * FROM todo WHERE completed_at IS NOT NULL

SQLAlchemy로 이와 같은 SQL 질의를 작성하는 코드가 바로 다음 코드예요.

from sqlmodel import null

select(Todo).where(Todo.completed_at.is_(null()))

select(Todo).where(Todo.completed_at.isnot(null()))

null() 은 SQLAlchemy에서 제공하는 함수예요. 데이터베이스 SQL에서 NULL을 표현하는 데 사용하죠.

Todo 모델에 있는 is_completed 프로퍼티는 SQLAlchemy가 질의를 다루는 데 사용할 수 없어요. 이 프로퍼티는 Python 영역에서만 다룰 수 있으며, SQLAlchemy가 이 프로퍼티를 SQL 질의문을 작성하는 데 필요한 정보는 없죠.

list_todo HTTP API에 status 쿼리 인자 추가

TodoService의 API를 구현했으니 HTTP API에 반영합니다. 할 일의 완료 여부, 취소 여부는 서로 배타적이라서 HTTP QueryString의 status 인자 하나로 할 일의 상태를 지정하도록 할게요. FastAPI는 함수의 인자 정보를 참고하여 URL 경로 인자인지, QueryString 인자인지 따위를 알아서 처리해줘요. 그래서 list_todo() 종단점 함수에 status 인자를 추가하기만 하면 돼요.

# pudding_todo/apps/todo/endpoints.py

from typing import Optional, Literal

@router.get("/todo-groups/{pk}/todos", name="list-todo-page")
@tpl.page("pages/todo-list.jinja2")
async def list_todo(
    pk: int,
    user: CurrentUserDep,
    service: TodoServiceDep,
    todo_group_service: TodoGroupServiceDep,
    status: Optional[Literal["all", "completed", "incompleted"]] = None,
) -> dict:
    ...
    
    params = {}
    match status:
        case "completed":
            params["is_completed"] = True
        case "incompleted":
            params["is_completed"] = False
        case _:
            params = {}
    
    todos = await service.findall(user.id, group_id=pk, **params)

status 인자의 값이 completed 이면 findall() 메서드에 is_completed 키워드 인자 값으로 True 를, incompleted 이면 False 를 전달합니다. 이 코드 구조에 취소한 일만 걸러내는 구현을 쉽게 추가할 수 있어요.

추가한 코드를 반영한 전체 코드는 다음과 같아요.

# pudding_todo/apps/todo/endpoints.py

from typing import Optional, Literal

@router.get("/todo-groups/{pk}/todos", name="list-todo-page")
@tpl.page("pages/todo-list.jinja2")
async def list_todo(
    pk: int,
    user: CurrentUserDep,
    service: TodoServiceDep,
    todo_group_service: TodoGroupServiceDep,
    status: Optional[Literal["all", "completed", "incompleted"]] = None,
) -> dict:
    todo_group = await todo_group_service.get_users_group_by_id(user.id, pk)
    if not todo_group:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Group not found")
    
    params = {}
    match status:
        case "completed":
            params["is_completed"] = True
        case "incompleted":
            params["is_completed"] = False
        case _:
            params = {}
    
    todos = await service.findall(user.id, group_id=pk, **params)
    ctx = {
        "todo_group": todo_group,
        "todos": todos,
    }

    return ctx

완료한 일만, 미완료한 일만 필터 UI 추가

마지막으로 필터 UI를 추가하면 끝나요.

{% block content %}
    <div>
        <h1>{{ todo_group.name }} 그룹함</h1>

        <nav>
            <a href="?status=all">전체</a>
            <a href="?status=completed">완료한 일만</a>
            <a href="?status=incompleted">완료하지 않은 일만</a>
        </nav>

        {% for todo in todos %}
            {{ todo_detail.render(todo, request.url_for) }}
        {% endfor %}
    </div>
{% endblock %}

SQLAlchemy의 hybrid_property로 SQL질의 선언적으로 작성

여기까지 구현하는 걸로 충분한데요. 다음 코드의 가독성이 썩 좋진 않아요.

if is_completed is True:
	stmt = stmt.where(Todo.completed_at.isnot(null()))
elif is_completed is False:
	stmt = stmt.where(Todo.completed_at.is_(null()))

비교적 간단한 구현이지만, 코드를 읽어야 비로소 어떤 의도인지 파악이 되거든요. 다음 의사 코드처럼 선언적으로 코드를 작성하면 의도가 잘 드러날 거예요.

if is_completed is True:
	stmt = stmt.where(Todo.is_completed.is_(True))
elif is_completed is False:
	stmt = stmt.where(Todo.is_completed.is_(False))

Todo.is_completed.is_(True) : Todo이 완료됐다
Todo.is_completed.is_(True) : Todo이 완료되지 않았다

is_completed 가 참인지 거짓인지를 판정하는 SQL 질의가 구체적으로 어떠한지를 코드로 드러내는 게 아니라 의도만 드러내는 거죠.

이러한 구현을 SQLAlchemy ORM 모델인 Todo 모델을 좀 더 선언적으로 구현해볼게요.

Todo 모델의 is_completed 를 SQLAlchemy 표현식으로

Todo 모델의 is_completed 프로퍼티는 SQLAlchemy가 질의를 작성하는 데 활용할 수 없다고 했는데요. SQLAlchemy에서 제공하는 hybrid_property 객체를 사용하면 활용할 수 있어요. 코드부터 보죠.

# pudding_todo/apps/todo/models.py

from sqlmodel import null
from sqlmodel.main import SQLModelConfig
from sqlalchemy.ext.hybrid import hybrid_property

class Todo(SQLModel, table=True):
    ...
 
    @hybrid_property
    def is_completed(self) -> bool:
        return self.completed_at is not None

    @is_completed.expression
    def is_completed(cls) -> bool:
        return cls.completed_at.isnot(null())

@property 로 장식되어 있던 is_completed() 를 @hybrid_property 로 장식했어요. hybrid_property 클래스로 장식해 생성한 객체는 Python 영역에서 프로퍼티처럼 동작해요. 차이점은 이 객체로 SQLAlchemy의 SQL 표현식을 만드는 객체를 만들 수 있다는 점이죠.

    @is_completed.expression
    def is_completed(cls) -> bool:
        return cls.completed_at.isnot(null())

바로 위에 있는 is_completed 의 expression() 메서드를 is_completed() 메서드에 장식자로 사용했는데요. 이러면 이 is_completed() 는 클래스 메서드처럼 동작해요.

obj = Todo()

obj.is_completed is False  # 1

Todo.is_completed.is_(null())  # 2

이 코드를 예시로 삼아 설명해드릴게요. Todo 모델의 인스턴스 객체인 obj 의 is_completed 는 @hybrid_property 로 장식한 def is_completed() 예요. 프로퍼티처럼 다루죠.

Todo 클래스(모델)의 is_completed 는 @is_completed.expression 으로 장식한 def is_completed() 예요. SQLAlchemy의 질의문을 생성할 때 사용하죠. 이 메서드의 구현을 보면 앞선 TodoService 의 findall() 구현에서 Todo.completed_at.isnot(null()) 라고 작성한 코드와 동일해요.

findall()에 반영

이제 이 구현을 TodoService 의 findall() 에 반영할게요.

# pudding_todo/apps/todo/services.py


class TodoService:
    ...
    
    async def findall(
            self,
            user_id: int,
            *,
            group_id: Optional[int] = None,
            offset: int = 0,
            limit: int = 10,
            is_completed: Optional[bool] = None,
        ) -> Sequence[Todo]:
        stmt = (
            select(Todo)
            .join(TodoGroup)
            .where(TodoGroup.user_id == user_id)
            .offset(offset)
            .limit(limit)
        )

        if group_id:
            stmt = stmt.where(Todo.group_id == group_id)

        if is_completed is True:
            stmt = stmt.where(Todo.is_completed)
        elif is_completed is False:
            stmt = stmt.where(~Todo.is_completed)
        # 또는 다음과 같이 사용할 수도 있습니다.
        # if is_completed is not None:
        #     stmt = stmt.where(Todo.is_completed.is_(is_completed))

        result = await self.session.execute(stmt)
        return result.unique().scalars().all()

~Todo.is_completed 는 Todo.is_completed 의 부정(not) 표현이예요. 어떤가요? 코드가 좀 더 선언적으로 작성되어 선언적 문법을 따르는 SQL과 잘 어울리고, 코드 가독성도 좋지요? 🙂

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