[Python][Django][Celery] Serializer를 pickle로!!

회사업무를 하면서 celery 작업 중 날짜로 argument를 받는 작업이 많다. 근데 여기서 귀찮은 점은 기본 celery serializer가 json으로 되어 있어서 이를 받았을 때 다시 변환하는 작업을 해줘야 하는데 이 과정에서 휴먼 에러가 너무 발생한 것을 많이 느낀 나는 python 객체로 그냥 넘기는 방법이 뭐가 있을까? 하고 고민을 하게 되었습니다.

 

https://docs.celeryproject.org/projects/kombu/en/stable/userguide/serialization.html

Serialization — Kombu 5.2.2 documentation

여기서 설명하기를 'json', 'pickle', 'msgpack' 이 세가지를 지원한다고 되어 있습니다.

 

The pickle format is very convenient as it can serialize and deserialize almost any object, but this is also a concern for security.
Carefully crafted pickle payloads can do almost anything a regular Python program can do, so if you let your consumer automatically decode pickled objects you must make sure to limit access to the broker so that untrusted parties do not have the ability to send messages!

여기서 경고하는 것은 신뢰할 수 없는 서드파티에 대해서 메세지를 주고 받지 말라! 그리고 보안성에 대해서 취약한 점을 가지고 있다. 이정도이다.

하지만 우리는 검토 끝에 쓰기로 했다. 왜냐하면 우리는 기껏해야 날짜만 전달하기 때문에 보안에 대해서 크게 위험할 것 같지 않다는 생각을 했기 때문이다.

 

그럼 실제로 한 번 확인해보는 시간을 가져보자.

 

기본적인 세팅은 https://vixxcode.tistory.com/162 해당 글과 유사하므로 참고하시길 바랍니다.

 

그래도 핵심적인 부분을 간략히 소개하고자 한다.

 

settings.py

 

INSTALLED_APPS

INSTALLED_APPS = [
	...
    'celery',
    'domain',
    'django_extensions'
]

CELERY에 관한 설정

CELERY_BROKER_URL = 'redis://127.0.0.1:6379'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379'
# 기본 CELERY 세팅 값
# CELERY_ACCEPT_CONTENT = ['application/json']
# CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Seoul'
CELERY_ENABLE_UTC = False
# PICKLE SERIALIZER 로 전환 했을 때
CELERY_ACCEPT_CONTENT = ['pickle', 'json']
CELERY_TASK_SERIALIZER = 'pickle'

 

 

worker.py

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "batch.settings")

app = Celery('batch')

app.config_from_object('django.conf:settings', namespace="CELERY")
app.autodiscover_tasks()
today = datetime.today()
app.conf.beat_schedule = {
    'serializer_test': {
        'task': 'domain.tasks.serializer_test',
        'schedule': crontab(),
        'args': (today,)
    }
}

여기서 스케줄러가 1분마다 현재 시간을 함수에 전달하는 것을 볼 수가 있다.

 

 

tasks.py

@shared_task
def serializer_test(base_date):
    print(type(base_date))
    return str(type(base_date))

 

 

그럼 기본적인 세팅은 소개 했고 바로 테스트를 해보는 시간을 가져볼까요?

 

기본 serializer를 썼을 때

celery -A batch worker 

celery -A batch beat -l info

셀러리 비트를 통해 전달받은 값을 찍었을 때 str 타입으로 반환 받은 것을 확인할 수가 있다.

 

 

하지만 serializer를 pickle로 바꿨을 때는 어떨까?

 

pickle serializer를 썼을 때

celery -A batch worker 

celery -A batch beat -l info

결과를 보면 datetime.datetime으로 넘어오는 것을 확인 할수가 있다.

 

필자는 다음과 같이 serializer를 pickle로 전환함에 따른 컨버팅 작업을 안하게 되었고, 휴먼에러도 줄일 수 있게 되었다!!

 

아 그리고 참고로 docker를 띄울 때에는 `C_FORCE_ROOT=true` 을 환경변수로 넣어줘야 함을 잊지 말도록 하자

 

위 글의 내용은 해당 github자료로 확인할 수 있습니다.

 

https://github.com/VIXXPARK/django-remind/tree/main/django-celery

GitHub - VIXXPARK/django-remind