상태 관리는 시스템을 개발하면서 필연적으로 만나게 되는 문제입니다. 피플펀드의 대출로 예를 들면 “REPAYING” 이라는 상태에서 “REPAID” 상태로의 변경은 가능해야 하나, “REPAID”에서 “FUNDING” 상태로의 변경은 이루어지면 안 됩니다. 상태의 개수가 많지 않을때에는 if 문을 이용하여 문제를 해결할 수 있습니다. naive한 구현의 예시입니다.

class LoanApplication(models.Model):

    status = models.CharField(default="RECEIVED", max_length=10)

    def validate_transition(self, next_status):
        prev_status = self.status

        if prev_status == "RECEIVED" and next_status == "JUDGING":
            return
        if prev_status == "REPAYING" and next_status == "REPAID":
            return
        ...

        raise ValueError("상태를 [{}]에서 [{}]으로 변경할 수 없습니다.".format(prev_status, next_status))

시스템을 계속해서 개선해 가면서 요건은 점점 복잡해졌고 대출의 상태는 초기의 구현보다 훨씬 많아졌습니다. 심지어 대출 신청의 상태가 전부 변경되기도 하는 일도 발생했습니다. 누구보다 빠르게 남들과는 다르게 시스템을 구현해야 하는 스타트업에서, 위처럼 naive한 구현으로는 빠트린 것이 없는지 확인이 어렵고 무엇보다도 유지보수 하기가 정말 어려워집니다. 모든 상태에 대하여 테스트 케이스를 만드는 것은 현실적으로 어렵습니다. (상태가 20개이면 자기 자신으로 변경되는 것을 포함하여 가능한 변경의 수는 자그마치 20 * 20 = 400개입니다!) if 문을 400번 사용하는 것보다 더 나은 방법은 없을까요?

유한 상태 기계 (Finite State Machine, FSM)

유한 상태 기계는 유한한 개수의 상태를 가질 수 있는 오토마타, 즉 추상 기계라고 할 수 있다. 이러한 기계는 한 번에 오로지 하나의 상태만을 가지게 되며, 현재 상태(Current State)란 임의의 주어진 시간의 상태를 칭한다. 이러한 기계는 어떠한 사건(Event)에 의해 한 상태에서 다른 상태로 변화할 수 있으며, 이를 전이(Transition)이라 한다. 특정한 유한 오토마톤은 현재 상태로부터 가능한 전이 상태와, 이러한 전이를 유발하는 조건들의 집합으로서 정의된다. - 위키백과

오토마타라고도 불리우는 유한 상태 기계는 대학교에서 컴파일러 과목을 수강하시게 되면 이론적으로 아주 깊게 배우실 수 있습니다. 이론은 잠시 두고, 간단하게 개념에 대해서만 설명드리면, 위키백과의 내용처럼 한 번에 하나의 상태만 가질 수 있으며 상태의 변경은 특정 조건이 만족되었을 때만 정해진 방향으로 전이(Transition) 되는 시스템이라고 생각하시면 조금 쉬울 것 같습니다. (전이는 상태의 변경을 말합니다)

이 시스템은 오브젝트의 상태를 관리해주는 시스템입니다. (게임에서는 이미 수많은 곳에서 이를 사용하고 있다고 합니다) 이를 Python으로 구현하기 위해 10분 정도 코딩하다가, 이미 잘 만들어 놓은 라이브러리가 있을 것 같아 조금 찾아보았습니다.

라이브러리

Python의 가장 큰 장점은 내가 필요한 많은 것들을 누군가가 이미 만들어 놓았다는 점입니다. 조금만 찾아보았는데도 django-fsm, viewflow, transitions 등 많은 라이브러리를 쉽게 찾을 수 있었습니다. 여러 라이브러리들을 검토해 본 후, 추후 Django 버전을 올릴 때 문제가 될 가능성을 최대한 줄이고자, Django에 의존성이 없는 transitions 라이브러리를 사용하였습니다. 라이브러리를 pip를 이용하여 설치해 줍니다.

pip install transitions==0.6.9 pygraphviz==1.5

소스 코드

이 글에서는 피플펀드 대출의 상태를 가독성을 위하여 일부 축약하였습니다. 여기서 triggersource에서 dest으로 상태를 변환하기 위하여 일어나는 행동이라고 보시면 됩니다. Machine 객체의 get_transitions 함수의 반환값의 리스트 길이가 0이면 변경할 수 없는 그래프로 판단하였습니다. 자세한 사용법은 라이브러리의 문서를 참조해 주세요.

from transitions import Machine

LOAN_TRANSITION = [
    {"trigger": "start_judge", "source": "RECEIVED", "dest": "JUDGING"},
    {"trigger": "approved", "source": "JUDGING", "dest": "FUNDING"},
    {"trigger": "rejected", "source": "JUDGING", "dest": "REJECTED"},
    {"trigger": "funded", "source": "FUNDING", "dest": "REPAYING"},
    {"trigger": "repay_completed", "source": "REPAYING", "dest": "REPAID"},
    {"trigger": "cancel", "source": ["RECEIVED", "JUDGING"], "dest": "CANCELED"},
]

class LoanApplication(models.Model):

    status = models.CharField(default="RECEIVED", max_length=10)

    def validate_transition(self, dest):
        """변경 가능한 상태인지 검증한다. 만약 변경 불가능한 상태이면 ValueError 발생한다."""
        source = self.status

        # 상태가 변경되지 않았을 경우 검증하지 않음
        if source == dest:
            return

        state_machine = Machine(
            initial=self._meta.get_field("status").get_default(),
            transitions=LOAN_TRANSITION,
        )

        if len(state_machine.get_transitions(source=source, dest=dest)) == 0:
            raise ValueError("상태를 [{}]에서 [{}]으로 변경할 수 없습니다.".format(source, dest))

테스트 작성하기

테스트를 작성해 봅시다. 신규 기능을 만들고 테스트를 작성하는 것은 추후 유지보수를 위하여 정말 중요합니다.

class TestValidateTransition(TestCase):
    def test_valid_transition(self):
        """RECEIVED 상태에서 JUDGING 상태로 변경 확인을 할 때 잘 되는지 확인한다."""

        source = "RECEIVED"
        dest = "JUDGING"
        loan_application = LoanApplication(status=source)

        loan_application.validate_transition(dest)

    def test_invalid_transition(self):
        """CANCELED 상태에서 JUDGING 상태로 변경 확인을 할 때 안 되는지 확인한다."""

        source = "CANCELED"
        dest = "JUDGING" 
        loan_application = LoanApplication(status=source)

        with self.assertRaises(ValueError) as error:
            loan_application.validate_transition(dest)

        self.assertEquals(str(error.exception), "상태를 [CANCELED]에서 [JUDGING]으로 변경할 수 없습니다.")

    def test_self_transition(self):
        """RECEIVED 상태에서 RECEIVED 상태로 변경 확인을 할 때 잘 되는지 확인한다."""

        source = "RECEIVED"
        dest = "RECEIVED"
        loan_application = LoanApplication(status=source)

        loan_application.validate_transition(dest)

테스트 코드가 올바르게 동작하는지 확인합니다.

$ python manage.py test unsecured.tests.models.test_validate_transition 

test_valid_transition (unsecured.tests.models.test_validate_transition.TestValidateTransition)
RECEIVED 상태에서 JUDGING 상태로 변경 확인을 할 때 잘 되는지 확인한다. ... ok
test_invalid_transition (unsecured.tests.models.test_validate_transition.TestValidateTransition)
CANCELED 상태에서 JUDGING 상태로 변경 확인을 할 때 안 되는지 확인한다. ... ok
test_self_transition (unsecured.tests.models.test_validate_transition.TestValidateTransition)
RECEIVED 상태에서 RECEIVED 상태로 변경 확인을 할 때 잘 되는지 확인한다. ... ok

----------------------------------------------------------------------
Ran 3 tests in 0.005s

OK

작성한 모든 테스트가 잘 통과하였습니다.

Django REST Framework의 Serializer에 적용

이제 코드가 잘 작동하는 것을 확인하였으니, API으로 들어온 요청을 검증하기 위하여 Serializer에 적용해 봅시다. status 필드의 validation이 필요하므로, validate_status 함수를 추가하여 약간의 코드를 추가해 주면 됩니다.

from rest_framework import status

class LoanApplicationSerializer(serializer.ModelSerializer):

    def validate_status(self, status):
        if self.instance:
            try:
                self.instance.validate_transition(status)

            except ValueError as e:
                raise serializers.ValidationError(detail=e, code=status.HTTP_400_BAD_REQUEST)

시험 삼아 [CANCELED] 상태인 대출을 [JUDGING] 상태로 변경하는 API를 호출해 보았습니다.

HTTP/1.1 400 Bad Request

{"detail": {"status": ["상태를 [CANCELED]에서 [JUDGING]으로 변경할 수 없습니다."]}}

검증 로직에서 잘 걸리는 것을 확인할 수 있습니다.

보너스 선물: 시각화

상태가 많지 않은 경우 손으로 그래프를 그릴 수 있지만, 상태가 많아지면 많아질수록 유지보수하기가 어려워집니다. 이 라이브러리를 사용하면 위의 작업으로 만들어진 소스 코드를 이용하여 그래프를 그릴 수 있습니다. (pygraphiz 라이브러리가 필요합니다)

from transitions.extensions import GraphMachine

LOAN_TRANSITION = [
    {"trigger": "start_judge", "source": "RECEIVED", "dest": "JUDGING"},
    {"trigger": "approved", "source": "JUDGING", "dest": "FUNDING"},
    {"trigger": "rejected", "source": "JUDGING", "dest": "REJECTED"},
    {"trigger": "funded", "source": "FUNDING", "dest": "REPAYING"},
    {"trigger": "repay_completed", "source": "REPAYING", "dest": "REPAID"},
    {"trigger": "cancel", "source": ["RECEIVED", "JUDGING"], "dest": "CANCELED"},
]

(
    GraphMachine(initial="RECEIVED", transitions=LOAN_TRANSITION)
    .get_graph()
    .draw("loan_application_fsm.png", prog="dot")
)

위의 코드를 실행하면 아래처럼 그래프가 생성됩니다. 자동 생성된 그래프 치고는 그럭저럭 꽤 괜찮은 퀄리티를 뽑아줍니다.