Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Use DagBag and DagModel instead #2

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Use DagBag and DagModel instead #2

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 34 additions & 18 deletions plugins/backfill/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,35 +1,51 @@
import subprocess
import json
import subprocess

import flask
from airflow.models.dag import DagModel
from airflow.models.dagbag import DagBag
from flask import request
from flask_admin import BaseView, expose


class Backfill(BaseView):

@expose('/')
@expose("/")
def base(self):
output = subprocess.check_output(
'airflow list_dags', shell=True).split()
dags = output[(output.index(b'DAGS') + 2):]
return self.render("backfill_page.html",
dags=[dag.decode('utf8') for dag in dags])
return self.render("backfill_page.html", dags=self._list_dags())

@expose('/backfill')
@expose("/backfill")
def run_backfill(self):
cmd = self._get_backfill_command(request.args.get("dag_name"),
request.args.get("task_name"),
request.args.get("start_date"),
request.args.get("end_date"))
cmd = self._get_backfill_command(
request.args.get("dag_name"),
request.args.get("task_name"),
request.args.get("start_date"),
request.args.get("end_date"),
)

subprocess.Popen(cmd, shell=True)
response = json.dumps({'submitted': True})
return flask.Response(response, mimetype='text/json')
response = json.dumps({"submitted": True})
return flask.Response(response, mimetype="text/json")

def _list_dags(self):
dagbag = DagBag()
dags = []

for dag_id in dagbag.dags:
orm_dag = DagModel.get_current(dag_id)
# inactive DAGs can't be backfilled....
is_active = (not orm_dag.is_paused) if orm_dag is not None else False

if is_active:
dags.append(dag_id)

return dags

def _get_backfill_command(self, dag_name, task_name, start_date, end_date):
if task_name:
return 'yes | airflow backfill --reset_dagruns --rerun_failed_tasks -x -i -s {} -e {} -t "{}" {}'.format(
start_date, end_date, task_name, dag_name)
start_date, end_date, task_name, dag_name
)
else:
return 'yes | airflow backfill --reset_dagruns --rerun_failed_tasks -x -s {} -e {} {}'.format(
start_date, end_date, dag_name)
return "yes | airflow backfill --reset_dagruns --rerun_failed_tasks -x -s {} -e {} {}".format(
start_date, end_date, dag_name
)