jason810496 commented on code in PR #72715:
URL: https://github.com/apache/airflow/pull/72715#discussion_r4367445232
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_bundles.py:
##########
@@ -435,3 +440,26 @@ def get_dag_bundle_files(
dag_bundle_files=[DagBundleFileResponse(**row) for row in
ordered[start:end]],
total_entries=len(ordered),
)
+
+
+@dag_bundles_router.post(
+ "/{bundle_name}/refresh",
+ status_code=status.HTTP_202_ACCEPTED,
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ dependencies=[Depends(action_logging()),
Depends(requires_access_dag_bundle(method="PUT"))],
+)
+def refresh_dag_bundle(bundle_name: str, session: SessionDep) ->
DagBundleRefreshResponse:
+ """Request an asynchronous Dag bundle refresh."""
+ try:
+ with session.begin_nested():
+ session.add(
+ DagPriorityParsingRequest(
+ bundle_name=bundle_name,
+ relative_fileloc=None,
+ )
+ )
+ session.flush()
+ except IntegrityError:
+ # The deterministic request key means an equivalent refresh is already
pending.
+ pass
Review Comment:
It would be better to raise out the 409 status code and mention the reason
out explicitly instead of hiding it silently.
##########
airflow-core/src/airflow/api_fastapi/core_api/security.py:
##########
@@ -228,6 +228,30 @@ def inner(
return inner
+def requires_access_dag_bundle(method: ResourceMethod) -> Callable[[str,
BaseUser, Session], None]:
+ """Authorize a Dag bundle action against its owning team."""
+
+ def inner(bundle_name: str, user: GetUserDep, session: SessionDep) -> None:
+ bundle = session.scalar(
+ select(DagBundleModel).where(
+ DagBundleModel.name == bundle_name,
+ DagBundleModel.active.is_(True),
+ )
+ )
+ if bundle is None:
+ raise HTTPException(status.HTTP_404_NOT_FOUND, "Dag bundle not
found")
Review Comment:
```suggestion
raise HTTPException(status.HTTP_404_NOT_FOUND, "Dag bundle is
not found or inactive")
```
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_bundles.py:
##########
@@ -944,3 +949,112 @@ def
test_404_for_a_bundle_whose_dags_are_not_readable(self, dag_scoped_client):
def test_404_for_an_unknown_bundle(self, dag_scoped_client):
assert
dag_scoped_client.get("/dagBundles/no_such_bundle/files").status_code == 404
+
+
+class TestRefreshDagBundle:
+ def setup_method(self):
+ clear_db_dag_parsing_requests()
+ clear_db_dags()
+ clear_db_dag_bundles()
+ clear_db_logs()
+
+ def teardown_method(self):
+ clear_db_dag_parsing_requests()
+ clear_db_dags()
+ clear_db_dag_bundles()
+ clear_db_logs()
+
+ @pytest.fixture
+ def auth_manager(self, test_client):
+ auth_manager = get_auth_manager()
+ with mock.patch.object(auth_manager, "is_authorized_dag"):
+ yield auth_manager
+
+ def test_global_bundle_refresh_uses_dag_permission(self, auth_manager,
session, test_client):
+ session.add(DagBundleModel(name="global-bundle"))
+ session.commit()
+ auth_manager.is_authorized_dag.return_value = True
+
+ response = test_client.post("/dagBundles/global-bundle/refresh")
+
+ assert response.status_code == 202
+ assert response.json() == {"bundle_name": "global-bundle"}
+ auth_manager.is_authorized_dag.assert_called_with(
+ method="PUT",
+ details=DagDetails(id=None, team_name=None),
+ user=mock.ANY,
+ )
+ parsing_requests =
session.scalars(select(DagPriorityParsingRequest)).all()
+ assert len(parsing_requests) == 1
+ assert parsing_requests[0].bundle_name == "global-bundle"
+ assert parsing_requests[0].relative_fileloc is None
+ _check_last_log(session, dag_id=None, event="refresh_dag_bundle",
logical_date=None)
+
+ def test_repeated_refresh_uses_existing_pending_request(self,
auth_manager, session, test_client):
+ session.add(DagBundleModel(name="global-bundle"))
+ session.commit()
+ auth_manager.is_authorized_dag.return_value = True
+
+ first_response = test_client.post("/dagBundles/global-bundle/refresh")
+ second_response = test_client.post("/dagBundles/global-bundle/refresh")
+
+ assert first_response.status_code == second_response.status_code == 202
Review Comment:
Then we will need to update the test case to expect 409 status code for the
second request.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_bundles.py:
##########
@@ -435,3 +440,26 @@ def get_dag_bundle_files(
dag_bundle_files=[DagBundleFileResponse(**row) for row in
ordered[start:end]],
total_entries=len(ordered),
)
+
+
+@dag_bundles_router.post(
Review Comment:
It should be `put` instead of `post` btw.
##########
airflow-core/src/airflow/api_fastapi/core_api/security.py:
##########
@@ -228,6 +228,30 @@ def inner(
return inner
+def requires_access_dag_bundle(method: ResourceMethod) -> Callable[[str,
BaseUser, Session], None]:
Review Comment:
Please reference the following authorization methods.
https://github.com/apache/airflow/blob/e87c35e344ee4ba6ef41d879d401e2b77d4dcf86/airflow-core/src/airflow/api_fastapi/core_api/security.py#L231-L238
https://github.com/apache/airflow/blob/e87c35e344ee4ba6ef41d879d401e2b77d4dcf86/airflow-core/src/airflow/api_fastapi/core_api/security.py#L201-L228
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_bundles.py:
##########
@@ -435,3 +440,26 @@ def get_dag_bundle_files(
dag_bundle_files=[DagBundleFileResponse(**row) for row in
ordered[start:end]],
total_entries=len(ordered),
)
+
+
+@dag_bundles_router.post(
+ "/{bundle_name}/refresh",
+ status_code=status.HTTP_202_ACCEPTED,
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+ dependencies=[Depends(action_logging()),
Depends(requires_access_dag_bundle(method="PUT"))],
+)
+def refresh_dag_bundle(bundle_name: str, session: SessionDep) ->
DagBundleRefreshResponse:
+ """Request an asynchronous Dag bundle refresh."""
+ try:
+ with session.begin_nested():
+ session.add(
+ DagPriorityParsingRequest(
+ bundle_name=bundle_name,
+ relative_fileloc=None,
+ )
+ )
+ session.flush()
+ except IntegrityError:
+ # The deterministic request key means an equivalent refresh is already
pending.
+ pass
+ return DagBundleRefreshResponse(bundle_name=bundle_name)
Review Comment:
I think we should follow the convention to return empty string instead of
bundle name.
https://github.com/apache/airflow/blob/c5faeb1555b4076739665b9434f616db37559727/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_parsing.py#L32-L44
There's no need to return the bundle_name back to user again since they pass
bundle_name itself to API server so they should know it already.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
