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

Skip deleted apps when updating dag_registry #588

Open
wants to merge 3 commits into
base: develop
Choose a base branch
from
Open

Skip deleted apps when updating dag_registry #588

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4abd118
Skip deleted apps when updating dag_registry
marioncottard Dec 16, 2024
967ca36
Update actions/checkout and actions/setup-python to v4
marioncottard Dec 16, 2024
656b116
Merge branch 'develop' into bugfix/old_apps
ggsdc Dec 18, 2024
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
4 changes: 2 additions & 2 deletions .github/workflows/test_cornflow_dags.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,9 +31,9 @@ jobs:
os: [ubuntu-latest]

steps:
- uses: actions/checkout@v1
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v1
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install cbc
Expand Down
16 changes: 11 additions & 5 deletions cornflow-dags/DAG/update_dag_registry.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from datetime import datetime, timedelta
import logging

from airflow import DAG
from airflow.models import DagModel
Expand All @@ -21,6 +22,8 @@
"catchup": False,
}

logger = logging.getLogger("airflow.task")


def update_dag_registry(**kwargs):
with create_session() as session:
Expand All @@ -30,16 +33,19 @@ def update_dag_registry(**kwargs):
for tag in dag.tags
if tag.name == "model"
]
print(f"MODEL DAGS: {model_dags}")
logger.info(f"MODEL DAGS: {model_dags}")
cf_client = connect_to_cornflow(EnvironmentVariablesBackend())
deployed_dags = [
dag["id"] for dag in cf_client.get_deployed_dags(encoding="br")
]
print(f"DEPLOYED DAGS: {deployed_dags}")
logger.info(f"DEPLOYED DAGS: {deployed_dags}")
all_apps = dict()
for app in get_new_apps():
all_apps[app.name] = app
for model in model_dags:
if model.dag_id not in all_apps:
logger.warning(f"App {model.dag_id} is registered in Airflow database but there is no app for it. Skipping")
continue
app = all_apps[model.dag_id]
if model.dag_id not in deployed_dags:
response = cf_client.create_deployed_dag(
Expand All @@ -54,7 +60,7 @@ def update_dag_registry(**kwargs):
config_schema=app.schema,
encoding="br",
)
print(f"DAG: {response['id']} registered")
logger.info(f"DAG: {response['id']} registered")
else:
# Even if the dag is registered, we still update its schemas
response = cf_client.put_deployed_dag(
Expand All @@ -71,7 +77,7 @@ def update_dag_registry(**kwargs):
),
encoding="br",
)
print(f"DAG: {model.dag_id} registered")
logger.info(f"DAG: {model.dag_id} registered")


dag = DAG(
Expand All @@ -90,4 +96,4 @@ def update_dag_registry(**kwargs):
)

if __name__ == "__main__":
update_dag_registry()
update_dag_registry()
Loading