python - 从 celery 功能发出推送通知

Question

我正在尝试使用 Django 消息显示消息，但是当@shared_task 上发生特定事件时应该执行此警报。这是我在 tasks.py 中的 celery 函数代码

def show_message_in_time(self,response):
    while True:
        # get_info_to_show() this function will do something 
        messages.info(response, 'Your email is added successfully!')
        sleep(2*60)

这是views.py中的函数

def index(response):
    
    show_message_in_time.delay(response)
    return render(response, "Sales.html", {'zipped_list': objs})

下面是我收到的错误

Exception Value:    
Object of type 'WSGIRequest' is not JSON serializable

我已尝试将 WSGIRequest 转换为 JSON，但也没有用。任何帮助将不胜感激!

Answer 1

解决您的问题的最简单方法是根据您真正需要的 WSGIRequest/HttpRequest 片段创建一个字典。

假设您的 WSGIRequest 有“用户名”和“电子邮件”部分。然后你将你的索引重构为:

def index(response):
    data = { 
        "method": response.method,
        "username": response["username"],
        "email": response["email"]
    }

    if response.method == "POST":
        data["response_data"] = response.POST

    if response.method == "GET":
        data["response_data"] = response.GET

    show_message_in_time.delay(data)
    return render(response, "Sales.html", {'zipped_list': objs})

如果您知道这是一个 post 方法，那么最快的测试是

def index(response):
    # POST is a dictionary, so it is definitely serializable
    show_message_in_time.delay(response.POST)
    return render(response, "Sales.html", {'zipped_list': objs})

免责声明:我从未使用过 Django 或任何其他 Web 框架，但我对 Celery 知之甚少。