장고 고급 #5 Signals 깊이와 트랜잭션 후 처리

프레임워크 장고 signals transaction
7 분 소요

중급 #3 Signals와 Middleware에서 시그널의 기본을 다졌다면, 이제 운영에서 시그널을 안전하게 쓰기 위한 도구상자를 정리합니다. 핵심 키워드는 **transaction.on_commit**과 절제.

시그널의 위험 — 왜 “절제” 인가 #

시그널은 강력하지만, 큰 프로젝트에서 가장 자주 비난받는 기능이기도 합니다. 이유:

  • 흐름이 안 보임. User.save() 한 줄 뒤에 무엇이 일어나는지 코드를 따라가도 안 보임 — 시그널 핸들러가 다른 파일에 있어서.
  • 순서가 모호. 같은 시그널에 여러 receiver가 붙으면 등록 순서대로 호출되지만, 등록 자체가 import 순서에 의존.
  • 트랜잭션과 미묘하게 어긋남. post_save가 트랜잭션 안에서 실행됨 — 이게 외부 호출이면 큰 문제.
  • 테스트가 새는 추상화. 다른 테스트의 시그널 등록이 영향을 줘 깜빡임 (flaky)의 원인.

장고 자신의 권장은 “가능하면 명시적 호출, 시그널은 정말 분리가 필요한 경우만”. 좋은 사용처:

  • 다른 앱이 우리 모델의 변경을 알아야 할 때 (auth.User의 변경에 우리 앱이 반응)
  • 캐시 무효화
  • 검색 인덱싱 (Elasticsearch, Algolia 등)

나쁜 사용처:

  • 같은 앱 안의 명시적 후처리 (그냥 save 다음 줄에 적자)
  • 결제, 잔액 같은 비즈니스 로직의 핵심 (시그널이 누락되면 데이터가 깨짐)

트랜잭션 — 짧은 복습 #

장고 view는 기본적으로 AUTOCOMMIT 모드. 명시적으로 묶으려면:

@transaction.atomic 
from django.db import transaction

@transaction.atomic
def transfer(from_id, to_id, amount):
    src = Account.objects.select_for_update().get(pk=from_id)
    dst = Account.objects.select_for_update().get(pk=to_id)
    src.balance -= amount
    dst.balance += amount
    src.save()
    dst.save()

블록 안의 모든 SQL이 한 트랜잭션. 예외가 나면 자동 rollback.

ATOMIC_REQUESTS #

settings.py 
DATABASES = {
    "default": {
        "ENGINE": "...",
        "ATOMIC_REQUESTS": True,
        ...
    }
}

이걸 켜면 모든 view가 자동으로 트랜잭션으로 감쌉니다. 5xx 응답이면 rollback. 편하지만:

  • 트랜잭션이 길어져 락 시간 증가
  • view 안에서 외부 API 호출이 있으면 그동안 트랜잭션 잡음

큰 트래픽 사이트는 명시적 atomic이 더 정밀합니다.

컨텍스트 매니저 + savepoint #

중첩 atomic = savepoint 
@transaction.atomic
def outer():
    Order.objects.create(...)
    try:
        with transaction.atomic():    # 내부 — savepoint로 동작
            risky_step()
    except SomeError:
        # outer 트랜잭션은 살아 있음
        log_failure()
    Order.objects.filter(...).update(status="processed")

내부 atomic은 savepoint — 부분 rollback이 가능합니다. 외부 트랜잭션은 영향 안 받음.

명시적 비활성화 
with transaction.atomic(savepoint=False):
    ...

극히 드문 경우입니다. 보통 기본 (savepoint=True)으로 충분.

transaction.on_commit — 후처리의 표준 #

가장 중요한 도구. 트랜잭션이 성공적으로 commit 된 후에 실행할 콜백을 등록합니다.

기본 
from django.db import transaction

@transaction.atomic
def create_order(user, items):
    order = Order.objects.create(user=user, total=0)
    for item in items:
        OrderItem.objects.create(order=order, ...)

    transaction.on_commit(lambda: send_confirmation_email(order.id))
    return order

흐름:

  1. order 생성, items 생성
  2. commit 됨
  3. 그제서야 send_confirmation_email 실행

만약 트랜잭션이 rollback 되면? 콜백은 호출되지 않습니다. 정확히 우리가 원하는 동작입니다.

왜 이게 필수인가 #

🚫 atomic 안에서 직접 호출 
@transaction.atomic
def create_order(user, items):
    order = Order.objects.create(user=user, total=0)
    send_confirmation_email(order.id)   # ← 위험
    raise SomeError()

이메일은 보내졌는데 트랜잭션은 rollback. 존재하지 않는 주문에 대한 이메일이 사용자에게 갑니다. 비슷한 패턴이 결제, 알림, 외부 인덱싱 등 모든 경우에서 같은 위험을 만듭니다.

on_commit이 정확히 이 문제를 막습니다.

시그널 안에서 #

post_save 시그널은 트랜잭션 안에서 호출됩니다. 외부 호출은 반드시 on_commit으로 미루세요.

signals.py 
from django.db import transaction
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Order

@receiver(post_save, sender=Order)
def index_order(sender, instance, created, **kwargs):
    if not created:
        return
    transaction.on_commit(lambda: search_index.add(instance.id))

search_index.add가 commit 후에만 실행. rollback 되면 인덱스에도 안 들어갑니다.

Celery와의 결합 #

비동기 작업 큐를 쓴다면:

Celery와 
@receiver(post_save, sender=Order)
def queue_processing(sender, instance, created, **kwargs):
    if not created:
        return
    transaction.on_commit(lambda: process_order_task.delay(instance.id))

여기서 on_commit이 결정적. 그렇지 않으면 Celery 워커가 받아서 DB를 조회했을 때 아직 commit 전이라 주문이 보이지 않을 수 있습니다. 미묘한 race condition의 흔한 원인. 자세한 Celery 통합은 DRF #4에서.

체이닝 / 여러 콜백 #

여러 번 호출 가능. 등록 순서대로 실행됩니다.

여러 콜백 
@transaction.atomic
def f():
    ...
    transaction.on_commit(callback_a)
    transaction.on_commit(callback_b)
    transaction.on_commit(callback_c)
    # commit 후 a → b → c 순

트랜잭션 밖에서 부르면 #

on_commit을 트랜잭션 밖에서 부르면 즉시 실행됩니다 (이미 “commit 된 것” 으로 간주). 그래서 라이브러리/유틸 함수에 안전하게 넣을 수 있습니다 — 호출자가 atomic으로 감쌌든 아니든.

시그널 — 깊이 있는 사용 #

dispatch_uid — 중복 등록 방지 #

signals.py 
@receiver(post_save, sender=Order, dispatch_uid="order_index_handler")
def index_order(...):
    ...

dispatch_uid가 유일한 ID. 같은 ID로 두 번 등록되면 무시됩니다. 모듈이 두 번 import 되어도 핸들러가 두 번 등록되는 일 없음.

apps.py에서 등록 #

시그널 모듈이 자동 import 되도록 권장 패턴.

myapp/apps.py 
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "myapp"

    def ready(self):
        from . import signals    # noqa
myapp/__init__.py 
default_app_config = "myapp.apps.MyAppConfig"   # Django 3.2+ 는 자동 감지

ready() 안에서 import만 하면 데코레이터가 등록을 자동.

⚠ ready() 안에서 DB 조회는 금지. migration도중에 호출될 수 있습니다.

자주 쓰는 시그널 #

시그널발생 시점
pre_save / post_savesave() 전/후
pre_delete / post_deletedelete() 전/후
pre_migrate / post_migratemigrate 명령 전/후
pre_init / post_init모델 인스턴스 생성
m2m_changedM2M 관계 변경 (add/remove/clear)
request_started / request_finished요청 시작/끝
got_request_exceptionview 안에서 예외 발생
user_logged_in / user_logged_out / user_login_failed인증

m2m_changed — 의외로 까다로움 #

m2m_changed 
from django.db.models.signals import m2m_changed

@receiver(m2m_changed, sender=Post.tags.through)
def on_tags_changed(sender, instance, action, pk_set, **kwargs):
    if action in {"post_add", "post_remove", "post_clear"}:
        cache.delete(f"post_tags:{instance.id}")

action 값:

  • pre_add, post_add
  • pre_remove, post_remove
  • pre_clear, post_clear

대부분 post_*만 보면 됩니다. pre_* + post_* 둘 다에 같은 핸들러를 붙이면 두 번 호출되는 흔한 실수가 있습니다.

Custom signal — 우리 앱의 시그널 정의 #

내장 시그널만 있는 게 아닙니다. 우리 도메인 이벤트를 시그널로.

myapp/signals.py — 정의 
from django.dispatch import Signal

order_paid = Signal()    # providing_args는 5.x에서 제거됨, 그냥 docstring으로
order_cancelled = Signal()
발행 
from .signals import order_paid

def mark_paid(order):
    order.status = "paid"
    order.save()
    order_paid.send(sender=Order, order=order, paid_at=timezone.now())
구독 
from django.dispatch import receiver
from myapp.signals import order_paid

@receiver(order_paid)
def reward_user(sender, order, paid_at, **kwargs):
    transaction.on_commit(lambda: give_points(order.user_id, 100))

@receiver(order_paid)
def notify_seller(sender, order, **kwargs):
    transaction.on_commit(lambda: notify_seller_task.delay(order.seller_id))

여러 모듈이 같은 이벤트에 반응하는 경우에 어울립니다. 다만 위에서 본 “흐름이 안 보임” 단점이 그대로 적용 — 명시적 함수 호출이 가능하면 그게 더 단순.

send vs send_robust #

둘의 차이 
order_paid.send(sender=Order, order=order)         # 한 receiver 라도 예외 → 전파
order_paid.send_robust(sender=Order, order=order)  # 예외를 받아 로그만, 다른 receiver는 계속

send_robust는 결과로 **(receiver, response_or_exception)**의 리스트를 반환. 응답을 검사하려면 이쪽.

이메일 알림처럼 한 receiver가 죽어도 다른 게 돌아야 하는 경우는 send_robust. 트랜잭션의 일부라면 send (예외로 rollback 시키고 싶음).

시그널 테스트 #

시그널 비활성화 — mute_signals #

factory_boy의 도구가 표준입니다.

설치 
pip install factory_boy
tests 
from factory.django import mute_signals
from django.db.models.signals import post_save

@mute_signals(post_save)
def test_create_without_signals():
    user = User.objects.create(email="a@b.com")
    # post_save 핸들러가 호출되지 않음

데이터 시드, 픽스처 로딩 등 부수효과를 원치 않는 경우에 결정적. 클래스 데코레이터로도 사용 가능.

시그널 직접 disconnect / reconnect #

setUp/tearDown 
from django.test import TestCase
from django.db.models.signals import post_save
from myapp.signals import index_order

class MyTest(TestCase):
    def setUp(self):
        post_save.disconnect(index_order, sender=Order)

    def tearDown(self):
        post_save.connect(index_order, sender=Order)

mute_signals가 더 깔끔하지만, 특정 receiver만 빼고 싶으면 직접.

시그널이 호출됐는지 검증 #

mock 
from unittest.mock import MagicMock

def test_signal_fires():
    handler = MagicMock()
    order_paid.connect(handler, sender=Order)
    try:
        mark_paid(order)
        handler.assert_called_once()
    finally:
        order_paid.disconnect(handler, sender=Order)

bulk_*와 시그널 #

#3에서 본 bulk_create, bulk_update, update, delete (queryset의)는 시그널을 발생시키지 않습니다. 큰 함정.

🚫 시그널 가정 
Order.objects.filter(status="pending").update(status="cancelled")
# post_save 호출 안 됨!

데이터 마이그레이션 같은 경우에서 의도적으로 시그널을 우회하는 건 OK. 하지만 평소 코드에서 시그널이 핵심 로직이라면 bulk_* 사용 시 직접 후처리를 코드에서 해줘야 합니다.

✅ 명시적 후처리 
ids = list(qs.values_list("id", flat=True))
Order.objects.filter(id__in=ids).update(status="cancelled")
for order_id in ids:
    transaction.on_commit(lambda i=order_id: post_cancel_hook(i))

lambda i=order_id의 기본 인자 트릭에 주의 — 안 그러면 모든 lambda가 마지막 order_id를 캡처합니다. 클로저 함정.

자주 만나는 함정 #

1) post_save 안에서 instance.save() 다시 호출 #

무한 재귀. update_fields로 끊거나, pre_save에서 처리하거나, Model.objects.filter(pk=instance.pk).update(...) (시그널 안 일으킴).

2) signal 안의 외부 API 호출이 트랜잭션을 묶음 #

위에서 본 그것. 반드시 on_commit으로.

3) m2m_changed가 두 번 호출 #

pre_*와 post_* 둘 다 받으면 두 번. 하나만.

4) 다른 앱의 시그널 등록 누락 #

apps.py의 ready()가 import를 트리거하지 않으면 시그널이 등록 안 됩니다. 정적 분석 도구가 못 잡는 흔한 회귀.

5) async view에서 시그널 핸들러 #

핸들러가 동기인데 비동기 컨텍스트에서 호출됩니다. 내부적으로 sync_to_async가 적용되긴 하지만, 핸들러가 무거우면 이벤트 루프를 막을 수 있습니다. 비동기 view가 메인이라면 시그널 대신 명시적 호출 + await가 더 정직.

정리 #

이번 글에서 잡은 것:

  • 시그널은 강력하지만 “흐름이 안 보임” 단점 — 절제 권장
  • @transaction.atomic의 쓰임, ATOMIC_REQUESTS의 트레이드오프
  • 중첩 atomic = savepoint, 부분 rollback가능
  • transaction.on_commit(callback) — 시그널/atomic 안에서 외부 호출의 표준
  • dispatch_uid로 중복 등록 방지, apps.py의 ready()에서 import
  • m2m_changed의 action 종류와 두 번 호출 함정
  • Custom Signal() — 도메인 이벤트, send vs send_robust
  • 테스트: mute_signals, disconnect/reconnect, MagicMock으로 호출 검증
  • bulk_* / queryset update/delete는 시그널 발생 안 함 — 명시 후처리
  • Celery와 결합: 반드시 on_commit 안에서 task.delay

다음 글(#6 Django Channels — WebSocket)에서는 #1 ASGI 위에 WebSocket을 얹습니다. AsyncWebsocketConsumer, group broadcast, HTTP view에서 push, daphne 배포까지 — 실시간 양방향 통신을 구현하는 방법을 다룹니다.

X