Published on

Django ORM 의 bulk_create 와 bulk_update에서 auto_now와 auto_now_add는 의미가 없다

Authors

최근 스케쥴링 작업 중 bulk_update()를 통해 대량의 데이터를 갱신해야 했고, 이 과정에서 DateTimeField로 갱신이 잘되는지 확인해야 했습니다.

새로운 필드를 생성하는 대신에, 자동으로 갱신(auto_now)되던 updated_at 필드(DateTimeField)를 그대로 사용하고자 했습니다.

대규모 데이터를 처리할 때 Django ORM의 bulk_create()bulk_update() 메서드는 성능 향상을 위해 매우 유용합니다.

하지만, 일반적인 create, update 와는 다르게 auto_now 는 bulk 에는 적용이 되지 않습니다.

그 과정과 Django 의 소스코드를 보면서 동작 원리와 ORM이 쿼리를 처리하는 방식을 보고 의도를 알아보도록 하겠습니다.(Django == 4.2.3 의 소스코드 입니다.)

(주인공 updated_at 필드)

updated_at = models.DateTimeField(verbose_name="수정일시", auto_now=True)

1. auto_now()

Django model 에서 auto_now=True처럼 옵션을 설정하면 해당 설정된 필드를 객체가 저장될 때마다 현재 시간으로 자동 설정되게 할 수 있습니다.

1.1 auto_now=True, auto_now_add=True

  • auto_now=True: 모델이 저장될 때마다 현재 시간을 자동으로 해당 필드에 설정합니다. 주로 "수정 시간"을 기록.
  • auto_now_add=True: 모델 인스턴스가 처음 생성될 때 현재 시간을 자동으로 해당 필드에 설정합니다. 이후에는 변경되지 않습니다.주로 "생성 시간"을 기록.

이 필드 옵션들은 Django가 모델을 저장할 때 특정 시점에 자동으로 값을 설정하도록 합니다.

그러나 bulk_update는 이런 자동 설정을 무시합니다.

1.2 DateField

DateTimeField(혹은 DateField)의 pre_save 메서드에서 auto_nowauto_now_add 여부를 확인합니다.

여기서 True 여부에 따라 그 시간(날짜)이 타임스탬프 처럼 찍히게(저장이) 되는 것 입니다.

# DateTimeField class
class DateTimeField(DateField):
		...

    def pre_save(self, model_instance, add):
        if self.auto_now or (self.auto_now_add and add):
            value = datetime.date.today()
            setattr(model_instance, self.attname, value)
            return value
        else:
            return super().pre_save(model_instance, add)

pre_save 메서드의 호출 시점 은 아래와 같습니다.

  • Model.save(),Model.objects.create() , QuerySet.create()
  • Model.objects.get_or_create(),Model.objects.update_or_create()
  • Django admin에서 저장 시

1.3 save_base()

auto_nowpre_save() 안에서 그 여부를 확인하고, pre_save()save_base() 안에서 호출됩니다.

save_base() 메서드는 실제로 모델 인스턴스를 데이터베이스에 저장하는 핵심 로직을 담당하고, 다음과 같은 일을 수행합니다.

  • 데이터베이스와 모델 유효성 검사
  • 시그널 전송 (pre_savepost_save)
  • 트랜잭션 관리
  • 부모 모델의 저장 처리
  • 실제 데이터베이스에 객체를 저장
# django/db/models/base.py

    def save_base(
        self,
        raw=False,
        force_insert=False,
        force_update=False,
        using=None,
        update_fields=None,
    ):
        """
        Handle the parts of saving which should be done only once per save,
        yet need to be done in raw saves, too. This includes some sanity
        checks and signal sending.

        The 'raw' argument is telling save_base not to save any parent
        models and not to do any changes to the values before save. This
        is used by fixture loading.
        """
        using = using or router.db_for_write(self.__class__, instance=self)
        assert not (force_insert and (force_update or update_fields))
        assert update_fields is None or update_fields
        cls = origin = self.__class__
        # Skip proxies, but keep the origin as the proxy model.
        if cls._meta.proxy:
            cls = cls._meta.concrete_model
        meta = cls._meta
        if not meta.auto_created:
            pre_save.send(
                sender=origin,
                instance=self,
                raw=raw,
                using=using,
                update_fields=update_fields,
            )
        # A transaction isn't needed if one query is issued.
        if meta.parents:
            context_manager = transaction.atomic(using=using, savepoint=False)
        else:
            context_manager = transaction.mark_for_rollback_on_error(using=using)
        with context_manager:
            parent_inserted = False
            if not raw:
                parent_inserted = self._save_parents(cls, using, update_fields)
            updated = self._save_table(
                raw,
                cls,
                force_insert or parent_inserted,
                force_update,
                using,
                update_fields,
            )
        # Store the database on which the object was saved
        self._state.db = using
        # Once saved, this is no longer a to-be-added instance.
        self._state.adding = False

        # Signal that the save is complete
        if not meta.auto_created:
            post_save.send(
                sender=origin,
                instance=self,
                created=(not updated),
                update_fields=update_fields,
                raw=raw,
                using=using,
            )

pre_save 는 DB에 저장되기 전에 발생하고 데이터를 수정하고 유효성을 검사합니다. post_save 는 DB에 저장된 직후에 발생하고 관련된 객체를 생성하고 저장 결과를 확인할 수 있습니다.

소스 코드를 확인해보면 pre_savepost_save 사이에서 부모 모델 저장, 테이블 저장, 객체의 데이터베이스 상태를 업데이트 합니다.

2. bulk_update()

2.1 소스코드

bulk_update는 성능 최적화를 위해 Django의 Model.save() 메서드를 호출하지 않고, 직접 SQL 쿼리를 생성하고 실행합니다.

    def bulk_update(self, objs, fields, batch_size=None):
        """
        Update the given fields in each of the given objects in the database.
        """


        # 1. 검증

        if batch_size is not None and batch_size <= 0:
            raise ValueError("Batch size must be a positive integer.")
        if not fields:
            raise ValueError("Field names must be given to bulk_update().")
        objs = tuple(objs)
        if any(obj.pk is None for obj in objs):
            raise ValueError("All bulk_update() objects must have a primary key set.")
        fields = [self.model._meta.get_field(name) for name in fields]
        if any(not f.concrete or f.many_to_many for f in fields):
            raise ValueError("bulk_update() can only be used with concrete fields.")
        if any(f.primary_key for f in fields):
            raise ValueError("bulk_update() cannot be used with primary key fields.")

        # 2. 필드 준비
        if not objs:
            return 0
        for obj in objs:
            obj._prepare_related_fields_for_save(
                operation_name="bulk_update", fields=fields
            )

        # PK is used twice in the resulting update query, once in the filter
        # and once in the WHEN. Each field will also have one CAST.

				# 3. 배치 처리 및 쿼리 준비
        self._for_write = True
        connection = connections[self.db]
        max_batch_size = connection.ops.bulk_batch_size(["pk", "pk"] + fields, objs)
        batch_size = min(batch_size, max_batch_size) if batch_size else max_batch_size
        requires_casting = connection.features.requires_casted_case_in_updates
        batches = (objs[i : i + batch_size] for i in range(0, len(objs), batch_size))
        updates = []

        # 4. CASE 문 생성 및 UPDATE 쿼리 준비
        for batch_objs in batches:
            update_kwargs = {}
            for field in fields:
                when_statements = []
                for obj in batch_objs:
                    attr = getattr(obj, field.attname)
                    if not hasattr(attr, "resolve_expression"):
                        attr = Value(attr, output_field=field)
                    when_statements.append(When(pk=obj.pk, then=attr))
                case_statement = Case(*when_statements, output_field=field)
                if requires_casting:
                    case_statement = Cast(case_statement, output_field=field)
                update_kwargs[field.attname] = case_statement
            updates.append(([obj.pk for obj in batch_objs], update_kwargs))

        # 5. SQL UPDATE 문 실행
        rows_updated = 0
        queryset = self.using(self.db)
        with transaction.atomic(using=self.db, savepoint=False):
            for pks, update_kwargs in updates:
                rows_updated += queryset.filter(pk__in=pks).update(**update_kwargs)
        return rows_updated

이 메서드는 objs라는 객체 리스트의 특정 필드를 업데이트하는 기능을 합니다. fields는 업데이트할 필드의 리스트를, batch_size는 배치의 크기를 설정합니다.

2.1.1 검증

  • batch_size가 양수인지 확인합니다.
  • fields가 제공되었는지 확인하고 유효성을 확인합니다.
  • 모든 객체에 pk가 설정되어 있는지 확인합니다.

2.1.2 필드 준비

  • 객체의 관련 필드를 업데이트를 위한 준비 작업을 수행합니다. 이는 관련 객체들이 올바르게 업데이트되도록 하기 위한 것입니다.
  • PK가 없는 모델 인스턴스가 할당되지 않았는지 확인하세요. 이 모델의 ForeignKey, GenericForeignKey 또는 OneToOneField. 만약에 필드는 null을 허용하므로 저장을 허용하면 자동으로 데이터가 손실됩니다.
  • PK는 결과 업데이트 쿼리에서 두 번 사용됩니다(필터에서 한 번, WHEN에서 한 번). 각 필드에는 하나의 CAST도 있습니다.

2.1.3 배치 처리 및 쿼리 준비

  • 데이터베이스를 연결하고 관련된 정보를 가져옵니다.
  • 데이터베이스에 따라 배치의 최대 크기를 계산합니다. 배치 크기가 주어진 값과 최대 값 중 작은 값으로 설정됩니다.
  • requires_casting을 사용하여 데이터베이스가 CASE 문에서 캐스팅이 필요한지 확인합니다.
  • 객체들을 배치로 나눕니다.

2.1.4 CASE 문 생성 및 UPDATE 쿼리 준비

  • 각 배치에 대해 update_kwargs를 준비합니다.
  • 특정 pk 값에 대해 다른 데이터 값들을 업데이트 하기 위해서,when_statements 리스트에 각 객체의 When 문을 아래 처럼추가합니다. ⇒ When(pk=obj.pk, then=attr))
  • Case 문(CASE 문)을 생성하고 필요한지(requires_casting)확인 후, Cast 를 진행해서 DB종류에 대한 필드 타입과 일치하도록 합니다.
  • 각 필드에 대해 update_kwargs를 설정하고, 업데이트할 객체의 pk 리스트와 함께 updates에 추가합니다.

2.1.5 SQL UPDATE 문 실행

  • transaction.atomic 이 걸려 있고, updates 리스트에 있는 각 배치에 대해 filter(pk__in=pks)로 필터링하고, update(**update_kwargs)로 업데이트를 수행합니다.
  • 업데이트된 행의 수를 rows_updated에 추가하고 마지막에 반환합니다.

2.4 save_base()와 bulk_update()의 차이점

위의 코드를 바탕으로 정리해보면

  • save_base()
    • Model 클래스에서 개별 객체를 저장할 때 호출됩니다.
    • 데이터베이스 트랜잭션을 처리하고, 시그널(pre_save, post_save)을 전송하며, 부모 모델을 저장합니다.
    • 자동 생성된 필드(auto_now 등)와 관련된 작업을 처리합니다.
  • bulk_update()
    • QuerySet 클래스에서 여러 객체를 SQL 쿼리로 업데이트 합니다. (Model.save()를 호출하지 않으며, save_base()를 호출하지도 않습니다.)
    • 성능을 최적화하기 위해 객체의 상태를 직접 수정합니다.

2.5 SQL 쿼리 예시

예를 들어, bulk_update를 사용하여 여러 Ticker 객체(진행한 프로젝트의 현재가 모델이름)의 priceupdated_at 필드를 업데이트하는 경우, 생성되는 SQL 쿼리는 다음과 유사합니다:


UPDATE ticker
SET price = CASE id
    WHEN 1 THEN 100.0
    WHEN 2 THEN 200.0
    WHEN 3 THEN 300.0
END,
updated_at = CASE id
    WHEN 1 THEN '2024-07-07 10:00:00'
    WHEN 2 THEN '2024-07-07 10:00:00'
    WHEN 3 THEN '2024-07-07 10:00:00'
END
WHERE id IN (1, 2, 3);

이 쿼리는 id가 1, 2, 3인 행들의 priceupdated_at 필드를 각각의 값으로 업데이트합니다.

bulk_update() 메서드는 이렇게 여러 객체를 한 번에 업데이트하기 위해 직접 SQL 쿼리를 생성하고 실행합니다.

3. 해결 방법

bulk_update를 사용하면서 auto_now 로 적용되어있는 필드를 갱신하려면 수동으로 값을 설정해야 했습니다.

이를 해결하는 몇 가지 방법을 살펴보겠습니다.

3.1 수동으로 updated_at 필드 업데이트

가장 간단한 방법은 bulk_update를 호출함에 앞서 updated_at 객체를 갱신하고 bulk_update 를 호출하는 것 입니다.

from django.utils import timezone

for new_ticker in new_tickers:
    key = (new_ticker.currency, new_ticker.exchange)
    if key in existing_tickers:
        existing_ticker = existing_tickers[key]
        existing_ticker.price = new_ticker.price
        existing_ticker.updated_at = timezone.now()  # 수동으로 현재 시간 설정
        to_update.append(existing_ticker)
    else:
        to_create.append(new_ticker)

Ticker.objects.bulk_update(to_update, ["price", "updated_at"])

3.2 Manager 커스텀

저는 위의 방법 보다는 bulk_update 을 사용하는 곳에서 동일하게 updated_at을 확인해야 했기 때문에 재사용 가능한 커스텀 매니저를 만들어 bulk_update 메서드에서 updated_at 필드를 자동으로 갱신하도록 했습니다.

from django.db import models
from django.utils import timezone

class CustomManager(models.Manager):
    def bulk_update_with_auto_now(self, objs, fields, batch_size=None):
        if not objs:
            return 0  # 빈 리스트인 경우 early return

        model = objs[0].__class__
        auto_now_fields = [
            field.name for field in model._meta.fields
            if isinstance(field, (models.DateTimeField, models.DateField)) and field.auto_now
        ]

        # auto_now 필드가 없으면 일반 bulk_update 실행
        if not auto_now_fields:
            return super().bulk_update(objs, fields, batch_size)

        now = timezone.now()
        # auto_now 필드 업데이트
        for obj in objs:
            for field_name in auto_now_fields:
                setattr(obj, field_name, now)

        # fields에 auto_now 필드 추가 (중복 제거)
        fields = list(set(fields + auto_now_fields))

        # bulk_update 호출
        return super().bulk_update(objs, fields, batch_size)

class BaseModelMixin(models.Model):
    created_at = models.DateTimeField(verbose_name="생성일시", auto_now_add=True)
    updated_at = models.DateTimeField(verbose_name="수정일시", auto_now=True)

    objects = CustomManager()

    class Meta:
        abstract = True

# 사용할때 이렇게 하면 커스텀 메서드 임을 명시할 수 있지만, 실제로는 bulk_update()로 네이밍 하고 사용중입니다.
Ticker.objects.bulk_update_with_auto_now(to_update, ["price"])

4. 도움되는 문서

hongreat 블로그의 글을 봐주셔서 감사합니다!