'Operators can be assigned to one DAG, one time. Repeat assignments to
that same DAG are ok.'

@dag.setter
def dag(self, dag):

    if (not isinstance(dag, DAG)):
        raise TypeError(u'Expected    DAG;    received    {}'.format(dag.__class__.__name__))
    elif (self.has_dag() and (self.dag is not dag)):
        raise AirflowException(u'The    DAG    assigned    to    {}    can    not    be    changed.'.format(self))
    elif (self.task_id not in dag.task_dict):
        dag.add_task(self)
    self._dag = dag
