yuseok89 commented on code in PR #66554:
URL: https://github.com/apache/airflow/pull/66554#discussion_r3234972694
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/dag_run.py:
##########
@@ -86,3 +123,435 @@ async def wait(self) -> AsyncGenerator[str, None]:
await asyncio.sleep(self.interval)
yield await self._serialize_response(dag_run := await
self._get_dag_run())
yield "\n"
+
+
+def _format_dag_run_key(dag_id: str, dag_run_id: str) -> str:
+ return f"{dag_id}.{dag_run_id}"
+
+
+def _authorize_dag_run(
+ *,
+ session: Session,
+ user,
+ dag_id: str,
+ method: AuthMethod,
+ cache: dict[str, bool],
+) -> bool:
+ """
+ Return whether ``user`` may perform ``method`` on Dag runs of ``dag_id``.
+
+ The result is memoised in ``cache`` so a bulk request that touches many
+ runs of the same Dag only pays for one ``is_authorized_dag`` call per Dag.
+ """
+ if dag_id not in cache:
+ team_name = DagModel.get_team_name(dag_id, session=session)
+ cache[dag_id] = get_auth_manager().is_authorized_dag(
+ method=method,
+ access_entity=DagAccessEntity.RUN,
+ details=DagDetails(id=dag_id, team_name=team_name),
+ user=user,
+ )
+ return cache[dag_id]
+
+
+def _apply_state_change(
+ dag_run: DagRun,
+ new_state: DAGRunPatchStates,
+ dag: SerializedDAG,
+ session: Session,
+) -> None:
+ """Apply ``new_state`` to ``dag_run`` and fire the matching listener
hook."""
+ if new_state == DAGRunPatchStates.SUCCESS:
+ set_dag_run_state_to_success(dag=dag, run_id=dag_run.run_id,
commit=True, session=session)
Review Comment:
Without `commit=True` these helpers don't apply the change, and using it
across all three states keeps them consistent.
Only QUEUED could go lighter with a direct state set.
Would that be better?
--
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]
