芹菜任务调用另一个微服务api-身份验证

问题描述

我有一个包含多个微服务项目的Django系统。我正在使用djangorestframework-sso在它们之间进行身份验证。

在我的一个微服务中，我有一个celery任务，该任务从另一个微服务中调用api函数。

由于未提供身份验证头，因此呼叫失败。

我尝试使用celery任务注入所需的标头，但无济于事：

class MyTask(Task):
abstract = True
def apply_async(self,args=None,kwargs=None,task_id=None,producer=None,link=None,link_error=None,**options):
    """ invoked either directly or via .delay() to fork a task from the main process """

    token = 'XXX'
    options['headers'] = options.get('headers',{})
    options['headers'].update({
        'Authorization': 'Bearer ' + token,})
    return super(MyTask,self).apply_async(
        args,kwargs,task_id,producer,link,link_error,**options)

使用python的请求库发出请求。因此，我认为尝试捕获所有请求并在执行之前注入标头。不知道该怎么做。

任何帮助将不胜感激。

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）