You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by fo...@apache.org on 2018/06/20 18:32:56 UTC
incubator-airflow git commit: [AIRFLOW-1919] Add option to query for
DAG runs given a DAG ID
Repository: incubator-airflow
Updated Branches:
refs/heads/master 862204678 -> 6ed2fb716
[AIRFLOW-1919] Add option to query for DAG runs given a DAG ID
Closes #3515 from feng-tao/airflow-1919
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/6ed2fb71
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/6ed2fb71
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/6ed2fb71
Branch: refs/heads/master
Commit: 6ed2fb716a64728d525837fadaa28e7ed7a0588b
Parents: 8622046
Author: Tao feng <tf...@lyft.com>
Authored: Wed Jun 20 20:32:50 2018 +0200
Committer: Fokko Driesprong <fo...@godatadriven.com>
Committed: Wed Jun 20 20:32:50 2018 +0200
----------------------------------------------------------------------
airflow/bin/cli.py | 72 ++++++++++++++++++++++++++++++++++++++++++++++++-
tests/core.py | 8 ++++++
2 files changed, 79 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/6ed2fb71/airflow/bin/cli.py
----------------------------------------------------------------------
diff --git a/airflow/bin/cli.py b/airflow/bin/cli.py
index b56e325..32ee483 100644
--- a/airflow/bin/cli.py
+++ b/airflow/bin/cli.py
@@ -1288,6 +1288,58 @@ def create_user(args):
raise SystemExit('Failed to create user.')
+@cli_utils.action_logging
+def list_dag_runs(args, dag=None):
+ if dag:
+ args.dag_id = dag.dag_id
+
+ dagbag = DagBag()
+
+ if args.dag_id not in dagbag.dags:
+ error_message = "Dag id {} not found".format(args.dag_id)
+ raise AirflowException(error_message)
+
+ dag_runs = list()
+ state = args.state.lower() if args.state else None
+ for run in DagRun.find(dag_id=args.dag_id,
+ state=state,
+ no_backfills=args.no_backfill):
+ dag_runs.append({
+ 'id': run.id,
+ 'run_id': run.run_id,
+ 'state': run.state,
+ 'dag_id': run.dag_id,
+ 'execution_date': run.execution_date.isoformat(),
+ 'start_date': ((run.start_date or '') and
+ run.start_date.isoformat()),
+ })
+ if not dag_runs:
+ print('No dag runs for {dag_id}'.format(dag_id=args.dag_id))
+
+ s = textwrap.dedent("""\n
+ {line}
+ DAG RUNS
+ {line}
+ {dag_run_header}
+ """)
+
+ dag_runs.sort(key=lambda x: x['execution_date'], reverse=True)
+ dag_run_header = '%-3s | %-20s | %-10s | %-20s | %-20s |' % ('id',
+ 'run_id',
+ 'state',
+ 'execution_date',
+ 'state_date')
+ print(s.format(dag_run_header=dag_run_header,
+ line='-' * 120))
+ for dag_run in dag_runs:
+ record = '%-3s | %-20s | %-10s | %-20s | %-20s |' % (dag_run['id'],
+ dag_run['run_id'],
+ dag_run['state'],
+ dag_run['execution_date'],
+ dag_run['start_date'])
+ print(record)
+
+
Arg = namedtuple(
'Arg', ['flags', 'help', 'action', 'default', 'nargs', 'type', 'choices', 'metavar'])
Arg.__new__.__defaults__ = (None, None, None, None, None, None, None)
@@ -1335,6 +1387,15 @@ class CLIFactory(object):
"store_true",
default=False),
+ # list_dag_runs
+ 'no_backfill': Arg(
+ ("--no_backfill",),
+ "filter all the backfill dagruns given the dag id", "store_true"),
+ 'state': Arg(
+ ("--state",),
+ "Only list the dag runs corresponding to the state"
+ ),
+
# backfill
'mark_success': Arg(
("-m", "--mark_success"),
@@ -1719,6 +1780,15 @@ class CLIFactory(object):
'reset_dag_run', 'rerun_failed_tasks',
)
}, {
+ 'func': list_dag_runs,
+ 'help': "List dag runs given a DAG id. If state option is given, it will only"
+ "search for all the dagruns with the given state. "
+ "If no_backfill option is given, it will filter out"
+ "all backfill dagruns for given dag id.",
+ 'args': (
+ 'dag_id', 'no_backfill', 'state'
+ )
+ }, {
'func': list_tasks,
'help': "List the tasks within a DAG",
'args': ('dag_id', 'tree', 'subdir'),
@@ -1855,7 +1925,7 @@ class CLIFactory(object):
)
subparsers_dict = {sp['func'].__name__: sp for sp in subparsers}
dag_subparsers = (
- 'list_tasks', 'backfill', 'test', 'run', 'pause', 'unpause')
+ 'list_tasks', 'backfill', 'test', 'run', 'pause', 'unpause', 'list_dag_runs')
@classmethod
def get_parser(cls, dag_parser=False):
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/6ed2fb71/tests/core.py
----------------------------------------------------------------------
diff --git a/tests/core.py b/tests/core.py
index c3adb3c..8e68be3 100644
--- a/tests/core.py
+++ b/tests/core.py
@@ -1004,6 +1004,14 @@ class CliTests(unittest.TestCase):
args = self.parser.parse_args(['list_dags', '--report'])
cli.list_dags(args)
+ def test_cli_list_dag_runs(self):
+ cli.trigger_dag(self.parser.parse_args([
+ 'trigger_dag', 'example_bash_operator', ]))
+ args = self.parser.parse_args(['list_dag_runs',
+ 'example_bash_operator',
+ '--no_backfill'])
+ cli.list_dag_runs(args)
+
def test_cli_create_user_random_password(self):
args = self.parser.parse_args([
'create_user', '-u', 'test1', '-l', 'doe', '-f', 'jon',