'Returns the number of task instances in the given DAG.
:param session: ORM session
:param dag_id: ID of the DAG to get the task concurrency of
:type dag_id: unicode
:param task_ids: A list of valid task IDs for the given DAG
:type task_ids: list[unicode]
:param states: A list of states to filter by if supplied
:type states: list[state]
:return: The number of running tasks
:rtype: int'

@staticmethod
@provide_session
def get_num_task_instances(dag_id, task_ids, states=None, session=None):

    qry = session.query(func.count(TaskInstance.task_id)).filter((TaskInstance.dag_id == dag_id), TaskInstance.task_id.in_(task_ids))
    if (states is not None):
        if (None in states):
            qry = qry.filter(or_(TaskInstance.state.in_(states), TaskInstance.state.is_(None)))
        else:
            qry = qry.filter(TaskInstance.state.in_(states))
    return qry.scalar()
