Skip to content

Write action log to DB when DAG run is trigged via API #28998

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

Merged
uranusjr merged 10 commits into from
Feb 3, 2023
Merged

Write action log to DB when DAG run is trigged via API #28998

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
a0c7fa7
add log entry for a dag run via api
kolfild26 Jan 17, 2023
e36c174
Merge branch 'main' into main
kolfild26 Jan 17, 2023
2757a96
Merge branch 'main' into main
kolfild26 Jan 18, 2023
b13a9d5
change Log to action_logging
kolfild26 Jan 18, 2023
89a7f8d
Merge branch 'main' into main
kolfild26 Jan 24, 2023
d1169bd
fix change to action_logging decorator
kolfild26 Jan 25, 2023
a078934
Merge branch 'main' into main
kolfild26 Jan 25, 2023
b1fbd92
Merge branch 'main' into main
kolfild26 Jan 31, 2023
e2b670c
add test _check_last_log() for dag_run
kolfild26 Jan 31, 2023
506b621
Merge branch 'main' into main
kolfild26 Feb 2, 2023
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
10 changes: 10 additions & 0 deletions airflow/api_connexion/endpoints/dag_run_endpoint.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,9 +56,13 @@
from airflow.models import DagModel, DagRun
from airflow.security import permissions
from airflow.utils.airflow_flask_app import get_airflow_app
from airflow.utils.log.action_logger import action_event_from_permission
from airflow.utils.session import NEW_SESSION, provide_session
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunType
from airflow.www.decorators import action_logging

RESOURCE_EVENT_PREFIX = "dag_run"


@security.requires_access(
Expand Down Expand Up @@ -281,6 +285,12 @@ def get_dag_runs_batch(*, session: Session = NEW_SESSION) -> APIResponse:
],
)
@provide_session
@action_logging(
event=action_event_from_permission(
prefix=RESOURCE_EVENT_PREFIX,
permission=permissions.ACTION_CAN_CREATE,
),
)
def post_dag_run(*, dag_id: str, session: Session = NEW_SESSION) -> APIResponse:
"""Trigger a DAG."""
dm = session.query(DagModel).filter(DagModel.dag_id == dag_id).first()
Expand Down
4 changes: 3 additions & 1 deletion tests/api_connexion/endpoints/test_dag_run_endpoint.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
from tests.test_utils.api_connexion_utils import assert_401, create_user, delete_roles, delete_user
from tests.test_utils.config import conf_vars
from tests.test_utils.db import clear_db_dags, clear_db_runs, clear_db_serialized_dags
from tests.test_utils.www import _check_last_log


@pytest.fixture(scope="module")
Expand Down Expand Up @@ -1032,7 +1033,7 @@ class TestPostDagRun(TestDagRunEndpoint):
pytest.param(None, None, None, id="all-missing"),
],
)
def test_should_respond_200(self, logical_date_field_name, dag_run_id, logical_date, note):
def test_should_respond_200(self, session, logical_date_field_name, dag_run_id, logical_date, note):
self._create_dag("TEST_DAG_ID")

# We'll patch airflow.utils.timezone.utcnow to always return this so we
Expand Down Expand Up @@ -1077,6 +1078,7 @@ def test_should_respond_200(self, logical_date_field_name, dag_run_id, logical_d
"run_type": "manual",
"note": note,
}
_check_last_log(session, dag_id="TEST_DAG_ID", event="dag_run.create", execution_date=None)

def test_should_respond_400_if_a_dag_has_import_errors(self, session):
"""Test that if a dagmodel has import errors, dags won't be triggered"""
Expand Down