# Path: helper/utils/tasks.py
# @shared_task
# def dmap(iter, callback):
#     # Map a callback over an iterator and return as a group
#     callback = subtask(callback)
#     return group(callback.clone([arg,]) for arg in iter)()
# 
# Path: helper/utils/dedup/tasks.py
# @shared_task
# def dedup_effect_wrapper(data, dedup_key, task_pair_id, effect):
#     from helper.models import DedupEvent # avoid recursive import.
#     if DedupEvent.objects.get_or_create(task_pair_id=task_pair_id,
#                                         key=data[dedup_key])[1]:
#         effect.apply_async(
#             (data,), link_error=remove_from_dedup_cache_on_error.s(
#                 dedup_key, task_pair_id, data))
#         return data
# 
# @shared_task
# def create_dedup_event(data, dedup_key, task_pair_id):
#     from helper.models import DedupEvent # avoid recursive import.
#     return DedupEvent.objects.get_or_create(
#         task_pair_id=task_pair_id, key=data[dedup_key])[1]


from collections import OrderedDict
from importlib import import_module
from django.db import models
from django.core.urlresolvers import reverse
from django.core.exceptions import ValidationError
from django.contrib.postgres.fields import HStoreField
from celery import chain
from helper.utils.tasks import dmap
from helper.utils.dedup.tasks import dedup_effect_wrapper, create_dedup_event
                       self.effect_task.replace('_', ' ').capitalize())

    @property
    def schedule(self):
        try:
            return self.cause.every
        except (AttributeError, ImportError):
            pass

    def run(self):
        cause_options = self.cause_agent.options or {}
        cause_options.update({k: v for k, v in self.cause_options.items()
                              if not k.startswith('_')})
        cause_options['task_pair_id'] = self.id
        effect_options = self.effect_agent.options or {}
        effect_options.update({k: v for k, v in self.effect_options.items()
                               if not k.startswith('_')})
        effect_options['task_pair_id'] = self.id

        cause = self.cause.s(**cause_options)

        if getattr(self.cause, 'dedup_key', None) is not None:
            effect = dedup_effect_wrapper.s(
                dedup_key=self.cause.dedup_key,
                task_pair_id=self.id,
                effect=self.effect.s(**effect_options),
            )
        else:
            effect = self.effect.s(**effect_options)


# Path: helper/utils/tasks.py
# @shared_task
# def dmap(iter, callback):
#     # Map a callback over an iterator and return as a group
#     callback = subtask(callback)
#     return group(callback.clone([arg,]) for arg in iter)()
# 
# Path: helper/utils/dedup/tasks.py
# @shared_task
# def dedup_effect_wrapper(data, dedup_key, task_pair_id, effect):
#     from helper.models import DedupEvent # avoid recursive import.
#     if DedupEvent.objects.get_or_create(task_pair_id=task_pair_id,
#                                         key=data[dedup_key])[1]:
#         effect.apply_async(
#             (data,), link_error=remove_from_dedup_cache_on_error.s(
#                 dedup_key, task_pair_id, data))
#         return data
# 
# @shared_task
# def create_dedup_event(data, dedup_key, task_pair_id):
#     from helper.models import DedupEvent # avoid recursive import.
#     return DedupEvent.objects.get_or_create(
#         task_pair_id=task_pair_id, key=data[dedup_key])[1]


# Path: helper/models.py
from collections import OrderedDict
from importlib import import_module
from django.db import models
from django.core.urlresolvers import reverse
from django.core.exceptions import ValidationError
from django.contrib.postgres.fields import HStoreField
from celery import chain
from helper.utils.tasks import dmap
from helper.utils.dedup.tasks import dedup_effect_wrapper, create_dedup_event

                       self.effect_task.replace('_', ' ').capitalize())

    @property
    def schedule(self):
        try:
            return self.cause.every
        except (AttributeError, ImportError):
            pass

    def run(self):
        cause_options = self.cause_agent.options or {}
        cause_options.update({k: v for k, v in self.cause_options.items()
                              if not k.startswith('_')})
        cause_options['task_pair_id'] = self.id
        effect_options = self.effect_agent.options or {}
        effect_options.update({k: v for k, v in self.effect_options.items()
                               if not k.startswith('_')})
        effect_options['task_pair_id'] = self.id

        cause = self.cause.s(**cause_options)

        if getattr(self.cause, 'dedup_key', None) is not None:
            effect = dedup_effect_wrapper.s(
                dedup_key=self.cause.dedup_key,
                task_pair_id=self.id,
                effect=self.effect.s(**effect_options),
            )
        else:
            effect = self.effect.s(**effect_options)


        return chain(cause, dmap.s(effect))()