Python-如何将 Feedparser 对象传递给 celery 任务？

Question

我使用 feedparser 模块来解析 RSS 提要。我需要将 feedparser 对象传递给 celery 任务。

在尝试传递对象时，我收到一条错误消息 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) 不可 JSON 序列化

如何将 feedparser 对象传递给 celery 任务？

这是我的代码:-

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 对象转换为可序列化的对象(字典)，然后将它们注册到 kombu 序列化程序注册表中，如 docs 中所述，以便让 celery 在其任务中使用新的序列化器。

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。

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 使用新的序列化器来序列化任务，就像 celery docs 一样；您应该使用 serializer 参数调用您的任务。

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

希望这有帮助。