Python-Как передать объект Feedparser в задачу сельдерея?

Question

У меня есть модуль feedparser для анализа RSS-каналов. Мне нужно передать объект feedparser к задаче сельдерея.

При попытке передать объект, я получаю сообщение об ошибке говорящее time.struct_time(tm_year=2015, tm_mon=2, tm_mday=12, tm_hour=8, tm_min=19, tm_sec=11, tm_wday=3, tm_yday=43, tm_isdst=0) is not JSON serializable

Как передать объект feedparser к задаче сельдерея?

Вот мой код: -

rss_content = feedparser.parse(rss_link) 
content_entries = rss_content['entries'] 
for content in content_entries: 
    parse_link.apply_async(args=[content, link, json_id, news_category], queue= news_source) #celery task 

Как мне это сделать?

Answer 1

Вы должны создать свой собственный кодер и декодер, который будет в основном превратить ваш time.time_struct объект в нечто сериализуемой (а Словаре), а затем зарегистрировать их в реестре комбу сериализатора, как описано в docs для того, чтобы позволить сельдерей использовать свой новый сериализатор в своей задаче.

import json 
import time 
import types 
import datetime 

class FeedContentEncoder(json.JSONEncoder): 
    def default(self, obj): 
     if isinstance(obj, time_struct): 
      epoch = int(time.mktime(time_struct)) 
      return {'__type__': '__time__', 'time': epoch} 
     else: 
      return json.FeedContentEncoder.default(self, obj) 

def decode_feed_content(obj): 
    if isinstance(obj, types.DictionaryType) and '__type__' in obj: 
     if obj['__type__'] == '__time__': 
      return datetime.datetime.fromtimestamp(obj['time']).timetuple() 
    return obj 

Необходимо уведомить kombu о своей новой сериализации, зарегистрировав их в реестре serializer.

from kombu.serialization import register 

def feed_content_json_dumps(obj): 
    return json.dumps(obj, cls=FeedContentEncoder) 

def feed_content_json_loads(obj): 
    return json.loads(obj, object_hook=decode_feed_content) 

register('feedcontentjson', 
     feed_content_json_dumps, 
     feed_content_json_loads, 
     content_type='application/x-feedcontent-json', 
     content_encoding='utf-8') 

Наконец, следует сказать, сельдерей использовать новый сериалайзер для сериализации задачи так же, как в celery docs; вы должны вызвать свою задачу с параметром serializer.

parse_link.apply_async(args=[content, link, json_id, news_category], queue= news_source, serializer='feedcontentjson') 

Надеюсь, что это поможет.