Celery: 커스텀 JSON 인코더/디코더를 쓰는 방법이 있습니까?

Celery: 커스텀 JSON 인코더/디코더를 쓰는 방법이 있습니까?

응용 프로그램의 셀러리 태스크에 보내고 싶은 개체가 있습니다.이러한 개체는 기본 json 라이브러리를 사용하여 json을 직렬화할 수 없습니다.커스텀 JSON을 사용하여 이러한 오브젝트를 시리얼화/디시리얼라이즈 할 수 있는 방법이 있습니까?Encoder/Decoder?

여기서 조금 늦었지만, 커스텀인코더와 디코더를 정의하려면 문서(http://docs.celeryproject.org/en/latest/userguide/calling.html#serializers)와 같이 Kombu 시리얼라이저 레지스트리에 등록해야 합니다.

예를 들어, 다음은 Django용 커스텀 datetime serializer/deserializer(python의 내장 json 모듈 하위 분류)입니다.

myjson.py (settings.py 파일과 동일한 폴더에 저장)

import json
from datetime import datetime
from time import mktime

class MyEncoder(json.JSONEncoder):   
    def default(self, obj):
        if isinstance(obj, datetime):
            return {
                '__type__': '__datetime__', 
                'epoch': int(mktime(obj.timetuple()))
            }
        else:
            return json.JSONEncoder.default(self, obj)

def my_decoder(obj):
    if '__type__' in obj:
        if obj['__type__'] == '__datetime__':
            return datetime.fromtimestamp(obj['epoch'])
    return obj

# Encoder function      
def my_dumps(obj):
    return json.dumps(obj, cls=MyEncoder)

# Decoder function
def my_loads(obj):
    return json.loads(obj, object_hook=my_decoder)

설정을 지정합니다.화이

# Register your new serializer methods into kombu
from kombu.serialization import register
from .myjson import my_dumps, my_loads

register('myjson', my_dumps, my_loads, 
    content_type='application/x-myjson',
    content_encoding='utf-8') 

# Tell celery to use your new serializer:
CELERY_ACCEPT_CONTENT = ['myjson']
CELERY_TASK_SERIALIZER = 'myjson'
CELERY_RESULT_SERIALIZER = 'myjson'

언급URL : https://stackoverflow.com/questions/21631878/celery-is-there-a-way-to-write-custom-json-encoder-decoder