def METHOD_NAME(self, *args, **options):
    tasks = self.get_tasks()
    task_info = tasks.get(options.get('task_info'))
    if not task_info:
        raise CommandError(
            'Unknown task provided. Options are: %s' % ', '.join(tasks.keys())
        )
    base_qs = self.get_base_queryset(options)
    pks = self.get_pks(
        base_qs,
        task_info['queryset_filters'],
        distinct=task_info.get('distinct'),
    )
    if options.get('limit'):
        pks = pks[: options.get('limit')]
    if 'pre' in task_info:
        # This is run in process to ensure its run before the tasks.
        pks = task_info['pre'](pks)
    if pks:
        kwargs = task_info.get('kwargs', {})
        if task_info.get('allowed_kwargs'):
            kwargs.update(
                {arg: options.get(arg, None) for arg in task_info['allowed_kwargs']}
            )
        # All the remaining tasks go in one group.
        grouping = []
        for chunk in chunked(pks, options.get('batch_size')):
            grouping.append(task_info['task'].subtask(args=[chunk], kwargs=kwargs))
        # Add the post task on to the end.
        post = None
        if 'post' in task_info:
            post = task_info['post'].subtask(args=[], kwargs=kwargs, immutable=True)
            ts = chord(grouping, post)
        else:
            ts = group(grouping)
        ts.apply_async()