@staticmethod<EOL><INDENT>@provide_session<EOL>def find(dag_id=None, run_id=None, execution_date=None,<EOL>state=None, external_trigger=None, no_backfills=False,<EOL>session=None):<DEDENT>
DR = DagRun<EOL>qry = session.query(DR)<EOL>if dag_id:<EOL><INDENT>qry = qry.filter(DR.dag_id == dag_id)<EOL><DEDENT>if run_id:<EOL><INDENT>qry = qry.filter(DR.run_id == run_id)<EOL><DEDENT>if execution_date:<EOL><INDENT>if isinstance(execution_date, list):<EOL><INDENT>qry = qry.filter(DR.execution_date.in_(execution_date))<EOL><DEDENT>else:<EOL><INDENT>qry = qry.filter(DR.execution_date == execution_date)<EOL><DEDENT><DEDENT>if state:<EOL><INDENT>qry = qry.filter(DR.state == state)<EOL><DEDENT>if external_trigger is not None:<EOL><INDENT>qry = qry.filter(DR.external_trigger == external_trigger)<EOL><DEDENT>if no_backfills:<EOL><INDENT>from airflow.jobs import BackfillJob<EOL>qry = qry.filter(DR.run_id.notlike(BackfillJob.ID_PREFIX + '<STR_LIT:%>'))<EOL><DEDENT>dr = qry.order_by(DR.execution_date).all()<EOL>return dr<EOL>
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: int, list
:param run_id: defines the the run id for this dag run
:type run_id: str
:param execution_date: the execution date
:type execution_date: datetime.datetime
:param state: the state of the dag run
:type state: airflow.utils.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: sqlalchemy.orm.session.Session