'Returns a set of dag runs for the given search criteria.
:param dag_id: the dag_id to find dag runs for
:type dag_id: integer, list
:param run_id: defines the the run id for this dag run
:type run_id: string
:param execution_date: the execution date
:type execution_date: datetime
:param state: the state of the dag run
:type state: State
:param external_trigger: whether this dag run is externally triggered
:type external_trigger: bool
:param no_backfills: return no backfills (True), return all (False).
Defaults to False
:type no_backfills: bool
:param session: database session
:type session: Session'

@staticmethod
@provide_session
def find(dag_id=None, run_id=None, execution_date=None, state=None, external_trigger=None, no_backfills=False, session=None):

    DR = DagRun
    qry = session.query(DR)
    if dag_id:
        qry = qry.filter((DR.dag_id == dag_id))
    if run_id:
        qry = qry.filter((DR.run_id == run_id))
    if execution_date:
        if isinstance(execution_date, list):
            qry = qry.filter(DR.execution_date.in_(execution_date))
        else:
            qry = qry.filter((DR.execution_date == execution_date))
    if state:
        qry = qry.filter((DR.state == state))
    if (external_trigger is not None):
        qry = qry.filter((DR.external_trigger == external_trigger))
    if no_backfills:
        from airflow.jobs import BackfillJob
        qry = qry.filter(DR.run_id.notlike((BackfillJob.ID_PREFIX + u'%')))
    dr = qry.order_by(DR.execution_date).all()
    return dr
