(self, task_id)
return self.meta_from_decoded(res)
Get task metadata for a task by id.
Get task metadata for a task by id.
Get task metadata for a task by id .
def _get_task_meta_for(self, task_id):
        """Get task metadata for a task by id."""
        obj = self.TaskModel._default_manager.get_task(task_id)
        res = obj.as_dict()
        meta = self.decode_content(obj, res.pop('meta', None)) or {}
        result = self.decode_content(obj, res.get('result'))

        task_args = res.get('task_args')
        task_kwargs = res.get('task_kwargs')
        try:
            task_args = self.decode_content(obj, task_args)
            task_kwargs = self.decode_content(obj, task_kwargs)
        except (DecodeError, binascii.Error):
            pass

        # the right names are args/kwargs, not task_args/task_kwargs,
        # keep both for backward compatibility
        res.update(
            meta,
            result=result,
            task_args=task_args,
            task_kwargs=task_kwargs,
            args=task_args,
            kwargs=task_kwargs,
        )
        return self.meta_from_decoded(res)
def _get_task_meta_for ( self , task_id ) : obj = self . TaskModel . _default_manager . get_task ( task_id ) res = obj . as_dict ( ) meta = self . decode_content ( obj , res . pop ( 'meta' , None ) ) or { } result = self . decode_content ( obj , res . get ( 'result' ) ) task_args = res . get ( 'task_args' ) task_kwargs = res . get ( 'task_kwargs' ) try : task_args = self . decode_content ( obj , task_args ) task_kwargs = self . decode_content ( obj , task_kwargs ) except ( DecodeError , binascii . Error ) : pass # the right names are args/kwargs, not task_args/task_kwargs, # keep both for backward compatibility res . update ( meta , result = result , task_args = task_args , task_kwargs = task_kwargs , args = task_args , kwargs = task_kwargs , ) return self . meta_from_decoded ( res )