DRF 성능 향상을 위한 PrefetchLatestObjectField와 효율적인 Serializer의 내부캐싱 활용방법

Serializer가 동작하는 원리만 제대로 이해해도 N+1을 방지하고, 불필요한 기술이전을 안해도 됩니다. 그러기 위해서 가장 간단한 방법으로 View단에서 ORM을 최적화하는 것이 중요한데, Prefetch 기능을 자주 사용하게 됩니다.

모델 구조상 하위 모델에서 여러개의 데이터에 대한 단일 객체를 가져와야 하는 경우가 있을 수도 있는데, 이렇게 되면 쿼리낭비가 반드시 발생하게 됩니다. 하지만 이 낭비를 최소화하는 방법이 존재하며, Serializer에서 그것을 최적화 할 수 있는 방법이 존재합니다.

이 글에서는 불필요한 중복 쿼리를 최대한 효율적으로 개선하여 Serializer를 커스텀하고 쿼리를 덜 낭비하는 방법을 기록합니다.

1. serializers.Field 커스텀하여 PrefetchLatestObjectField 생성하기
2. Serializer 캐싱 전략
- 2.1 내부에서 반복적인 DB 접근을 최소화
  - 2.1.1 many=True 상황에서는 그냥 모델과 view를 최적화 하는것이 정답!
- 2.2 context를 활용 (many=True 상황)
3. 참고 자료

1. serializers.Field 커스텀하여 PrefetchLatestObjectField 생성하기

1.1 동일한 하위 모델을 나누기 to_attr

예를 들어, 특정 모델과 연관된 메모(Memo) 데이터를 두 가지 유형으로 나누어 가져온다고 가정하고,두 유형의 메모는 각각 "A 타입 메모"와 "B 타입 메모"라고 하겠습니다.

이때 Prefetch 기능을 사용하여 아래와 같이 가져올 수 있습니다.

queryset 자체도 미리 선언하여 최적화 할 수 있겠지만, 다음과 같이 필터링하여도 문제는 없습니다.

from django.db.models import Prefetch

prefetch_queryset = YourModel.objects.prefetch_related(
    Prefetch(
        "memos",
        queryset=Memo.objects.filter(memo_type="a", is_active=True)
        .select_related("creator")
        .order_by("-created_at"),
        to_attr="a_memos",
    ),
    Prefetch(
        "memos",
        queryset=Memo.objects.filter(memo_type="b", is_active=True)
        .select_related("creator")
        .order_by("-created_at"),
        to_attr="b_memos",
    ),
)

이렇게 하면 각각의 조건으로 필터링된 메모가 미리 불러와져 a_memos, b_memos 속성으로 접근할 수 있게 됩니다.

그러나 Serializer에서 이 데이터를 직렬화할 때, 여러 개의 메모 중 가장 최근의 메모 하나만을 직렬화하고 싶은 경우가 많습니다.

기본 Serializer로는 이를 깔끔하게 처리하기 어렵습니다.

1.2 PrefetchLatestObjectField 커스텀 필드 작성

PrefetchLatestObjectField 라는 이름은 공식적으로 사용하는 이름은 아니고 제가 지어낸 것 입니다.

이름 그대로 이 커스텀을 통해서 Prefetch로 가져온 리스트의 첫 번째 객체만 직렬화 합니다.

DRF의 기본클래스 중 serializers.Field를 확장하여, Prefetch로 가져온 리스트의 첫 번째 객체만 직렬화하는 부분입니다.

from rest_framework import serializers

class PrefetchLatestObjectField(serializers.Field):
    def __init__(self, serializer_class, **kwargs):
        self.serializer_class = serializer_class
        super().__init__(**kwargs)

    def to_representation(self, value):
        if not value:
            return None
        latest_obj = value[0]  # 최신 객체만 가져오기
        return self.serializer_class(latest_obj).data

이 필드는 미리 Prefetch로 가져온 리스트에서 첫 번째 항목만 직렬화합니다.

1.3 PrefetchLatestObjectField 커스텀 필드 사용하기

이제 커스텀 필드를 사용하여 Serializer를 작성하면 다음과 같이 깔끔해집니다.

class MemoSerializer(serializers.ModelSerializer):
    class Meta:
        model = Memo
        fields = ["id", "content", "creator", "created_at"]


class YourModelSerializer(serializers.ModelSerializer):
    latest_a_memo = PrefetchLatestObjectField(
        MemoSerializer,
        source="a_memos",
        read_only=True,
    )
    latest_b_memo = PrefetchLatestObjectField(
        MemoSerializer,
        source="b_memos",
        read_only=True,
    )

    class Meta:
        model = YourModel
        fields = ["id", "name", "latest_a_memo", "latest_b_memo"]

여기에서 중요한 것은 source 파라미터입니다. 이는 PrefetchLatestObjectField가 Prefetch된 데이터를 찾을 수 있도록, 미리 지정된 속성(to_attr)과 연결해주는 역할을 합니다.

1.4 내부 작동 방식

DRF의 Serializer는 직렬화를 수행할 때, 각 필드의 source로 지정된 속성을 인스턴스로부터 찾아 값을 가져옵니다.

위 예시에서 source="a_memos"는 다음과 같이 동작합니다

# DRF 내부의 동작 예시
data = {}
data['latest_a_memo'] = PrefetchLatestObjectField.to_representation(
    instance.a_memos
)

이처럼 내부적으로 to_representation 메서드가 호출되어 직렬화가 이루어집니다.

2. Serializer 캐싱 전략

Serializer의 동작 원리상 SerializerMethodField 는 instance 별로 동작하며 queryset에서 최적화되지 않은 경우 (대부분.. 반드시..) N+1 이 발생합니다.

필드 간 상태 공유가 되지 않아 반복적인 DB 접근이 일어나는 것이기 때문에 이점을 이용하여, 메모리 도구나 서드파티 방식이 아닌 코드 단에서 간단한 최적화를 하는 방법입니다.

2.1 내부에서 반복적인 DB 접근을 최소화

_get_cached_data 에서 미리 가져온 데이터를 캐싱하여, SerializerMethodField에서 반복적으로 DB 접근을 하지 않도록 합니다.

class ExampleSerializer(serializers.Serializer):
    field_a = serializers.SerializerMethodField()
    field_b = serializers.SerializerMethodField()

    def _get_cached_data(self, obj):
        if not hasattr(self, '_cached_data'):
            self._cached_data = SomeModel.objects.filter(related_id=obj.id).first()
        return self._cached_data

    def get_field_a(self, obj):
        data = self._get_cached_data(obj)
        return data.value_a if data else None

    def get_field_b(self, obj):
        data = self._get_cached_data(obj)
        return data.value_b if data else None

2.1.1 many=True 상황에서는 그냥 모델과 view를 최적화 하는것이 정답!

many=True로 사용할 때는 SerializerMethodField가 각 인스턴스마다 호출되므로, 위와 같은 방식으로 캐싱을 해도 데이터가 많고 복잡하다면 크게 효과가 없겠습니다. 실제로 모델과 ORM을 최적화하는 것이 정답이며, 이렇지않은 상태는 대부분 잘못된 구조이며 리팩토링이나 다른 차원의 개선이 필요하다고 생각합니다.

2.2 context를 활용 (many=True 상황)

위의 상황에서 저는 many=True 가 아니었기 때문에, 간단하게 _cached_data를 활용하여 캐싱을 하였습니다. 하지만 그럼에도 모델개선이나 ORM 최적화를 통해서 캐싱을 하려고 한다면 다른 방법으로 context를 이용하는 방법이 있습니다.(참고 자료)

view에서 queryset을 미리 가져와서 context에 저장하고, SerializerMethodField에서 이를 활용하는 방식입니다. many=True로 사용할 때는 context를 활용하여 모든 serializer 인스턴스가 캐시를 공유할 수 있도록 합니다. 이 방식을 활용하면 복잡한 데이터를 효율적이고 깔끔하게 처리할 수 있게 됩니다.

class ExampleSerializer(serializers.Serializer):
    field_a = serializers.SerializerMethodField()

    def _get_cached_data(self, obj):
        cache = self.context.setdefault('_cache', {})
        if obj.id not in cache:
            cache[obj.id] = SomeModel.objects.filter(related_id=obj.id).first()
        return cache[obj.id]

    def get_field_a(self, obj):
        data = self._get_cached_data(obj)
        return data.value_a if data else None

3. 참고 자료

get_serializer_context에 대한 이전 글