'Returns the task instances for this dag run'

@provide_session
def get_task_instances(self, state=None, session=None):

    TI = TaskInstance
    tis = session.query(TI).filter((TI.dag_id == self.dag_id), (TI.execution_date == self.execution_date))
    if state:
        if isinstance(state, six.string_types):
            tis = tis.filter((TI.state == state))
        elif (None in state):
            tis = tis.filter(or_(TI.state.in_(state), TI.state.is_(None)))
        else:
            tis = tis.filter(TI.state.in_(state))
    if (self.dag and self.dag.partial):
        tis = tis.filter(TI.task_id.in_(self.dag.task_ids))
    return tis.all()
