Skip to content

Add pendingActions onto ui/dags#55041

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
bbovenzi merged 8 commits into from
Sep 2, 2025
Merged

Add pendingActions onto ui/dags #55041

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
fe10736
Add pendingActions onto ui/dags
bbovenzi Aug 28, 2025
1641ad3
refactor(api_fastapi): unify usage of the term Dag
Lee-W Aug 29, 2025
804043c
refactor(api_fastapi): simplify HITL loading logic
Lee-W Aug 29, 2025
11ba363
test: add test case test_should_return_200_with_hitl
Lee-W Aug 29, 2025
bd17685
Refactor naming scheme and filter
bbovenzi Aug 29, 2025
98eab6f
fix(hitl): fix has_pending_actions to use >= 1 and == 0 to query
Lee-W Sep 1, 2025
d9597b7
feat(zh-tw): update Taiwanese Mandarin translation
Lee-W Sep 1, 2025
60e13bc
Reduce i18n surface area and parameterize dags test
bbovenzi Sep 2, 2025
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
54 changes: 45 additions & 9 deletions airflow-core/src/airflow/api_fastapi/common/parameters.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -114,7 +114,7 @@ def depends(cls, offset: NonNegativeInt = 0) -> OffsetFilter:


class _FavoriteFilter(BaseParam[bool]):
"""Filter DAGs by favorite status."""
"""Filter Dags by favorite status."""

def __init__(self, user_id: str, value: T | None = None, skip_none: bool = True) -> None:
super().__init__(skip_none=skip_none)
Expand DownExpand Up@@ -607,7 +607,7 @@ def depends_float(
DateTimeQuery = Annotated[str, AfterValidator(_safe_parse_datetime)]
OptionalDateTimeQuery = Annotated[str | None, AfterValidator(_safe_parse_datetime_optional)]

# DAG
# Dag
QueryLimit = Annotated[LimitFilter, Depends(LimitFilter.depends)]
QueryOffset = Annotated[OffsetFilter, Depends(OffsetFilter.depends)]
QueryPausedFilter = Annotated[
Expand DownExpand Up@@ -638,7 +638,7 @@ def depends_float(


class _HasAssetScheduleFilter(BaseParam[bool]):
"""Filter DAGs that have asset-based scheduling."""
"""Filter Dags that have asset-based scheduling."""

def to_orm(self, select: Select) -> Select:
if self.value is None and self.skip_none:
Expand All@@ -647,22 +647,22 @@ def to_orm(self, select: Select) -> Select:
asset_ref_subquery = sql_select(DagScheduleAssetReference.dag_id).distinct()

if self.value:
# Filter DAGs that have asset-based scheduling
# Filter Dags that have asset-based scheduling
return select.where(DagModel.dag_id.in_(asset_ref_subquery))

# Filter DAGs that do NOT have asset-based scheduling
# Filter Dags that do NOT have asset-based scheduling
Comment thread
Lee-W marked this conversation as resolved.
return select.where(DagModel.dag_id.notin_(asset_ref_subquery))

@classmethod
def depends(
cls,
has_asset_schedule: bool | None = Query(None, description="Filter DAGs with asset-based scheduling"),
has_asset_schedule: bool | None = Query(None, description="Filter Dags with asset-based scheduling"),
) -> _HasAssetScheduleFilter:
return cls().set_value(has_asset_schedule)


class _AssetDependencyFilter(BaseParam[str]):
"""Filter DAGs by specific asset dependencies."""
"""Filter Dags by specific asset dependencies."""

def to_orm(self, select: Select) -> Select:
if self.value is None and self.skip_none:
Expand All@@ -681,7 +681,7 @@ def to_orm(self, select: Select) -> Select:
def depends(
cls,
asset_dependency: str | None = Query(
None, description="Filter DAGs by asset dependency (name or URI)"
None, description="Filter Dags by asset dependency (name or URI)"
),
) -> _AssetDependencyFilter:
return cls().set_value(asset_dependency)
Expand All@@ -690,6 +690,42 @@ def depends(
QueryHasAssetScheduleFilter = Annotated[_HasAssetScheduleFilter, Depends(_HasAssetScheduleFilter.depends)]
QueryAssetDependencyFilter = Annotated[_AssetDependencyFilter, Depends(_AssetDependencyFilter.depends)]


class _PendingActionsFilter(BaseParam[bool]):
"""Filter Dags by having pending HITL actions (more than 1)."""

def to_orm(self, select: Select) -> Select:
if self.value is None and self.skip_none:
return select

from airflow.models.hitl import HITLDetail
from airflow.models.taskinstance import TaskInstance

# Join with HITLDetail and TaskInstance to find Dags
pending_actions_count_subquery = (
sql_select(func.count(HITLDetail.ti_id))
.join(TaskInstance, HITLDetail.ti_id == TaskInstance.id)
.where(HITLDetail.response_at.is_(None))
.where(TaskInstance.dag_id == DagModel.dag_id)
.scalar_subquery()
)

if self.value is True:
# Filter to show only Dags with pending actions
where_clause = pending_actions_count_subquery >= 1
else:
# Filter to show only Dags without pending actions
where_clause = pending_actions_count_subquery == 0

return select.where(where_clause)

@classmethod
def depends(cls, has_pending_actions: bool | None = Query(None)) -> _PendingActionsFilter:
return cls().set_value(has_pending_actions)


QueryPendingActionsFilter = Annotated[_PendingActionsFilter, Depends(_PendingActionsFilter.depends)]

# DagRun
QueryLastDagRunStateFilter = Annotated[
FilterParam[DagRunState | None],
Expand DownExpand Up@@ -748,7 +784,7 @@ def _transform_dag_run_types(types: list[str] | None) -> list[DagRunType | None]
),
]

# DAGTags
# DagTags
QueryDagTagPatternSearch = Annotated[
_SearchParam, Depends(search_param_factory(DagTag.name, "tag_name_pattern"))
]
Expand Down
2 changes: 2 additions & 0 deletions airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -20,13 +20,15 @@
from airflow.api_fastapi.core_api.base import BaseModel
from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse
from airflow.api_fastapi.core_api.datamodels.hitl import HITLDetail


class DAGWithLatestDagRunsResponse(DAGResponse):
"""DAG with latest dag runs response serializer."""

asset_expression: dict | None
latest_dag_runs: list[DAGRunResponse]
pending_actions: list[HITLDetail]


class DAGWithLatestDagRunsCollectionResponse(BaseModel):
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
Add pendingActions onto ui/dags by bbovenzi · Pull Request #55041 · apache/airflow · GitHub
Skip to content
Add pendingActions onto ui/dags#55041

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

bbovenzi
merged 8 commits into

from

Sep 2, 2025
 Merged


Add pendingActions onto ui/dags
#55041
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
fe10736

Add pendingActions onto ui/dags


bbovenzi Aug 28, 2025
1641ad3

refactor(api_fastapi): unify usage of the term Dag


Lee-W Aug 29, 2025
804043c

refactor(api_fastapi): simplify HITL loading logic


Lee-W Aug 29, 2025
11ba363

test: add test case test_should_return_200_with_hitl


Lee-W Aug 29, 2025
bd17685

Refactor naming scheme and filter


bbovenzi Aug 29, 2025
98eab6f

fix(hitl): fix has_pending_actions to use >= 1 and == 0 to query


Lee-W Sep 1, 2025
d9597b7

feat(zh-tw): update Taiwanese Mandarin translation


Lee-W Sep 1, 2025
60e13bc

Reduce i18n surface area and parameterize dags test


bbovenzi Sep 2, 2025

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


54 changes: 45 additions & 9 deletions
airflow-core/src/airflow/api_fastapi/common/parameters.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -114,7 +114,7 @@ def depends(cls, offset: NonNegativeInt = 0) -> OffsetFilter:


class _FavoriteFilter(BaseParam[bool]):
 """Filter DAGs by favorite status."""
 """Filter Dags by favorite status."""

 def __init__(self, user_id: str, value: T | None = None, skip_none: bool = True) -> None:
 super().__init__(skip_none=skip_none)
Expand DownExpand Up@@ -607,7 +607,7 @@ def depends_float(
DateTimeQuery = Annotated[str, AfterValidator(_safe_parse_datetime)]
OptionalDateTimeQuery = Annotated[str | None, AfterValidator(_safe_parse_datetime_optional)]

# DAG
# Dag
QueryLimit = Annotated[LimitFilter, Depends(LimitFilter.depends)]
QueryOffset = Annotated[OffsetFilter, Depends(OffsetFilter.depends)]
QueryPausedFilter = Annotated[
Expand DownExpand Up@@ -638,7 +638,7 @@ def depends_float(


class _HasAssetScheduleFilter(BaseParam[bool]):
 """Filter DAGs that have asset-based scheduling."""
 """Filter Dags that have asset-based scheduling."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -647,22 +647,22 @@ def to_orm(self, select: Select) -> Select:
 asset_ref_subquery = sql_select(DagScheduleAssetReference.dag_id).distinct()

 if self.value:
 # Filter DAGs that have asset-based scheduling
 # Filter Dags that have asset-based scheduling
 return select.where(DagModel.dag_id.in_(asset_ref_subquery))

 # Filter DAGs that do NOT have asset-based scheduling
 # Filter Dags that do NOT have asset-based scheduling
Comment thread
Lee-W marked this conversation as resolved.

 return select.where(DagModel.dag_id.notin_(asset_ref_subquery))

 @classmethod
 def depends(
 cls,
 has_asset_schedule: bool | None = Query(None, description="Filter DAGs with asset-based scheduling"),
 has_asset_schedule: bool | None = Query(None, description="Filter Dags with asset-based scheduling"),
 ) -> _HasAssetScheduleFilter:
 return cls().set_value(has_asset_schedule)


class _AssetDependencyFilter(BaseParam[str]):
 """Filter DAGs by specific asset dependencies."""
 """Filter Dags by specific asset dependencies."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -681,7 +681,7 @@ def to_orm(self, select: Select) -> Select:
 def depends(
 cls,
 asset_dependency: str | None = Query(
 None, description="Filter DAGs by asset dependency (name or URI)"
 None, description="Filter Dags by asset dependency (name or URI)"
 ),
 ) -> _AssetDependencyFilter:
 return cls().set_value(asset_dependency)
Expand All@@ -690,6 +690,42 @@ def depends(
QueryHasAssetScheduleFilter = Annotated[_HasAssetScheduleFilter, Depends(_HasAssetScheduleFilter.depends)]
QueryAssetDependencyFilter = Annotated[_AssetDependencyFilter, Depends(_AssetDependencyFilter.depends)]


class _PendingActionsFilter(BaseParam[bool]):
 """Filter Dags by having pending HITL actions (more than 1)."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
 return select

 from airflow.models.hitl import HITLDetail
 from airflow.models.taskinstance import TaskInstance

 # Join with HITLDetail and TaskInstance to find Dags
 pending_actions_count_subquery = (
 sql_select(func.count(HITLDetail.ti_id))
 .join(TaskInstance, HITLDetail.ti_id == TaskInstance.id)
 .where(HITLDetail.response_at.is_(None))
 .where(TaskInstance.dag_id == DagModel.dag_id)
 .scalar_subquery()
 )

 if self.value is True:
 # Filter to show only Dags with pending actions
 where_clause = pending_actions_count_subquery >= 1
 else:
 # Filter to show only Dags without pending actions
 where_clause = pending_actions_count_subquery == 0

 return select.where(where_clause)

 @classmethod
 def depends(cls, has_pending_actions: bool | None = Query(None)) -> _PendingActionsFilter:
 return cls().set_value(has_pending_actions)


QueryPendingActionsFilter = Annotated[_PendingActionsFilter, Depends(_PendingActionsFilter.depends)]

# DagRun
QueryLastDagRunStateFilter = Annotated[
 FilterParam[DagRunState | None],
Expand DownExpand Up@@ -748,7 +784,7 @@ def _transform_dag_run_types(types: list[str] | None) -> list[DagRunType | None]
 ),
]

# DAGTags
# DagTags
QueryDagTagPatternSearch = Annotated[
 _SearchParam, Depends(search_param_factory(DagTag.name, "tag_name_pattern"))
]
Expand Down

2 changes: 2 additions & 0 deletions
airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,13 +20,15 @@
from airflow.api_fastapi.core_api.base import BaseModel
from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse
from airflow.api_fastapi.core_api.datamodels.hitl import HITLDetail


class DAGWithLatestDagRunsResponse(DAGResponse):
 """DAG with latest dag runs response serializer."""

 asset_expression: dict | None
 latest_dag_runs: list[DAGRunResponse]
 pending_actions: list[HITLDetail]


class DAGWithLatestDagRunsCollectionResponse(BaseModel):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add pendingActions onto ui/dags by bbovenzi · Pull Request #55041 · apache/airflow · GitHub
Skip to content
Add pendingActions onto ui/dags#55041

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

bbovenzi
merged 8 commits into

from

Sep 2, 2025
 Merged


Add pendingActions onto ui/dags
#55041
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
fe10736

Add pendingActions onto ui/dags


bbovenzi Aug 28, 2025
1641ad3

refactor(api_fastapi): unify usage of the term Dag


Lee-W Aug 29, 2025
804043c

refactor(api_fastapi): simplify HITL loading logic


Lee-W Aug 29, 2025
11ba363

test: add test case test_should_return_200_with_hitl


Lee-W Aug 29, 2025
bd17685

Refactor naming scheme and filter


bbovenzi Aug 29, 2025
98eab6f

fix(hitl): fix has_pending_actions to use >= 1 and == 0 to query


Lee-W Sep 1, 2025
d9597b7

feat(zh-tw): update Taiwanese Mandarin translation


Lee-W Sep 1, 2025
60e13bc

Reduce i18n surface area and parameterize dags test


bbovenzi Sep 2, 2025

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


54 changes: 45 additions & 9 deletions
airflow-core/src/airflow/api_fastapi/common/parameters.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -114,7 +114,7 @@ def depends(cls, offset: NonNegativeInt = 0) -> OffsetFilter:


class _FavoriteFilter(BaseParam[bool]):
 """Filter DAGs by favorite status."""
 """Filter Dags by favorite status."""

 def __init__(self, user_id: str, value: T | None = None, skip_none: bool = True) -> None:
 super().__init__(skip_none=skip_none)
Expand DownExpand Up@@ -607,7 +607,7 @@ def depends_float(
DateTimeQuery = Annotated[str, AfterValidator(_safe_parse_datetime)]
OptionalDateTimeQuery = Annotated[str | None, AfterValidator(_safe_parse_datetime_optional)]

# DAG
# Dag
QueryLimit = Annotated[LimitFilter, Depends(LimitFilter.depends)]
QueryOffset = Annotated[OffsetFilter, Depends(OffsetFilter.depends)]
QueryPausedFilter = Annotated[
Expand DownExpand Up@@ -638,7 +638,7 @@ def depends_float(


class _HasAssetScheduleFilter(BaseParam[bool]):
 """Filter DAGs that have asset-based scheduling."""
 """Filter Dags that have asset-based scheduling."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -647,22 +647,22 @@ def to_orm(self, select: Select) -> Select:
 asset_ref_subquery = sql_select(DagScheduleAssetReference.dag_id).distinct()

 if self.value:
 # Filter DAGs that have asset-based scheduling
 # Filter Dags that have asset-based scheduling
 return select.where(DagModel.dag_id.in_(asset_ref_subquery))

 # Filter DAGs that do NOT have asset-based scheduling
 # Filter Dags that do NOT have asset-based scheduling
Comment thread
Lee-W marked this conversation as resolved.

 return select.where(DagModel.dag_id.notin_(asset_ref_subquery))

 @classmethod
 def depends(
 cls,
 has_asset_schedule: bool | None = Query(None, description="Filter DAGs with asset-based scheduling"),
 has_asset_schedule: bool | None = Query(None, description="Filter Dags with asset-based scheduling"),
 ) -> _HasAssetScheduleFilter:
 return cls().set_value(has_asset_schedule)


class _AssetDependencyFilter(BaseParam[str]):
 """Filter DAGs by specific asset dependencies."""
 """Filter Dags by specific asset dependencies."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -681,7 +681,7 @@ def to_orm(self, select: Select) -> Select:
 def depends(
 cls,
 asset_dependency: str | None = Query(
 None, description="Filter DAGs by asset dependency (name or URI)"
 None, description="Filter Dags by asset dependency (name or URI)"
 ),
 ) -> _AssetDependencyFilter:
 return cls().set_value(asset_dependency)
Expand All@@ -690,6 +690,42 @@ def depends(
QueryHasAssetScheduleFilter = Annotated[_HasAssetScheduleFilter, Depends(_HasAssetScheduleFilter.depends)]
QueryAssetDependencyFilter = Annotated[_AssetDependencyFilter, Depends(_AssetDependencyFilter.depends)]


class _PendingActionsFilter(BaseParam[bool]):
 """Filter Dags by having pending HITL actions (more than 1)."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
 return select

 from airflow.models.hitl import HITLDetail
 from airflow.models.taskinstance import TaskInstance

 # Join with HITLDetail and TaskInstance to find Dags
 pending_actions_count_subquery = (
 sql_select(func.count(HITLDetail.ti_id))
 .join(TaskInstance, HITLDetail.ti_id == TaskInstance.id)
 .where(HITLDetail.response_at.is_(None))
 .where(TaskInstance.dag_id == DagModel.dag_id)
 .scalar_subquery()
 )

 if self.value is True:
 # Filter to show only Dags with pending actions
 where_clause = pending_actions_count_subquery >= 1
 else:
 # Filter to show only Dags without pending actions
 where_clause = pending_actions_count_subquery == 0

 return select.where(where_clause)

 @classmethod
 def depends(cls, has_pending_actions: bool | None = Query(None)) -> _PendingActionsFilter:
 return cls().set_value(has_pending_actions)


QueryPendingActionsFilter = Annotated[_PendingActionsFilter, Depends(_PendingActionsFilter.depends)]

# DagRun
QueryLastDagRunStateFilter = Annotated[
 FilterParam[DagRunState | None],
Expand DownExpand Up@@ -748,7 +784,7 @@ def _transform_dag_run_types(types: list[str] | None) -> list[DagRunType | None]
 ),
]

# DAGTags
# DagTags
QueryDagTagPatternSearch = Annotated[
 _SearchParam, Depends(search_param_factory(DagTag.name, "tag_name_pattern"))
]
Expand Down

2 changes: 2 additions & 0 deletions
airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,13 +20,15 @@
from airflow.api_fastapi.core_api.base import BaseModel
from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse
from airflow.api_fastapi.core_api.datamodels.hitl import HITLDetail


class DAGWithLatestDagRunsResponse(DAGResponse):
 """DAG with latest dag runs response serializer."""

 asset_expression: dict | None
 latest_dag_runs: list[DAGRunResponse]
 pending_actions: list[HITLDetail]


class DAGWithLatestDagRunsCollectionResponse(BaseModel):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add pendingActions onto ui/dags by bbovenzi · Pull Request #55041 · apache/airflow · GitHub
Skip to content
Add pendingActions onto ui/dags#55041

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

bbovenzi
merged 8 commits into

from

Sep 2, 2025
 Merged


Add pendingActions onto ui/dags
#55041
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
fe10736

Add pendingActions onto ui/dags


bbovenzi Aug 28, 2025
1641ad3

refactor(api_fastapi): unify usage of the term Dag


Lee-W Aug 29, 2025
804043c

refactor(api_fastapi): simplify HITL loading logic


Lee-W Aug 29, 2025
11ba363

test: add test case test_should_return_200_with_hitl


Lee-W Aug 29, 2025
bd17685

Refactor naming scheme and filter


bbovenzi Aug 29, 2025
98eab6f

fix(hitl): fix has_pending_actions to use >= 1 and == 0 to query


Lee-W Sep 1, 2025
d9597b7

feat(zh-tw): update Taiwanese Mandarin translation


Lee-W Sep 1, 2025
60e13bc

Reduce i18n surface area and parameterize dags test


bbovenzi Sep 2, 2025

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


54 changes: 45 additions & 9 deletions
airflow-core/src/airflow/api_fastapi/common/parameters.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -114,7 +114,7 @@ def depends(cls, offset: NonNegativeInt = 0) -> OffsetFilter:


class _FavoriteFilter(BaseParam[bool]):
 """Filter DAGs by favorite status."""
 """Filter Dags by favorite status."""

 def __init__(self, user_id: str, value: T | None = None, skip_none: bool = True) -> None:
 super().__init__(skip_none=skip_none)
Expand DownExpand Up@@ -607,7 +607,7 @@ def depends_float(
DateTimeQuery = Annotated[str, AfterValidator(_safe_parse_datetime)]
OptionalDateTimeQuery = Annotated[str | None, AfterValidator(_safe_parse_datetime_optional)]

# DAG
# Dag
QueryLimit = Annotated[LimitFilter, Depends(LimitFilter.depends)]
QueryOffset = Annotated[OffsetFilter, Depends(OffsetFilter.depends)]
QueryPausedFilter = Annotated[
Expand DownExpand Up@@ -638,7 +638,7 @@ def depends_float(


class _HasAssetScheduleFilter(BaseParam[bool]):
 """Filter DAGs that have asset-based scheduling."""
 """Filter Dags that have asset-based scheduling."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -647,22 +647,22 @@ def to_orm(self, select: Select) -> Select:
 asset_ref_subquery = sql_select(DagScheduleAssetReference.dag_id).distinct()

 if self.value:
 # Filter DAGs that have asset-based scheduling
 # Filter Dags that have asset-based scheduling
 return select.where(DagModel.dag_id.in_(asset_ref_subquery))

 # Filter DAGs that do NOT have asset-based scheduling
 # Filter Dags that do NOT have asset-based scheduling
Comment thread
Lee-W marked this conversation as resolved.

 return select.where(DagModel.dag_id.notin_(asset_ref_subquery))

 @classmethod
 def depends(
 cls,
 has_asset_schedule: bool | None = Query(None, description="Filter DAGs with asset-based scheduling"),
 has_asset_schedule: bool | None = Query(None, description="Filter Dags with asset-based scheduling"),
 ) -> _HasAssetScheduleFilter:
 return cls().set_value(has_asset_schedule)


class _AssetDependencyFilter(BaseParam[str]):
 """Filter DAGs by specific asset dependencies."""
 """Filter Dags by specific asset dependencies."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -681,7 +681,7 @@ def to_orm(self, select: Select) -> Select:
 def depends(
 cls,
 asset_dependency: str | None = Query(
 None, description="Filter DAGs by asset dependency (name or URI)"
 None, description="Filter Dags by asset dependency (name or URI)"
 ),
 ) -> _AssetDependencyFilter:
 return cls().set_value(asset_dependency)
Expand All@@ -690,6 +690,42 @@ def depends(
QueryHasAssetScheduleFilter = Annotated[_HasAssetScheduleFilter, Depends(_HasAssetScheduleFilter.depends)]
QueryAssetDependencyFilter = Annotated[_AssetDependencyFilter, Depends(_AssetDependencyFilter.depends)]


class _PendingActionsFilter(BaseParam[bool]):
 """Filter Dags by having pending HITL actions (more than 1)."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
 return select

 from airflow.models.hitl import HITLDetail
 from airflow.models.taskinstance import TaskInstance

 # Join with HITLDetail and TaskInstance to find Dags
 pending_actions_count_subquery = (
 sql_select(func.count(HITLDetail.ti_id))
 .join(TaskInstance, HITLDetail.ti_id == TaskInstance.id)
 .where(HITLDetail.response_at.is_(None))
 .where(TaskInstance.dag_id == DagModel.dag_id)
 .scalar_subquery()
 )

 if self.value is True:
 # Filter to show only Dags with pending actions
 where_clause = pending_actions_count_subquery >= 1
 else:
 # Filter to show only Dags without pending actions
 where_clause = pending_actions_count_subquery == 0

 return select.where(where_clause)

 @classmethod
 def depends(cls, has_pending_actions: bool | None = Query(None)) -> _PendingActionsFilter:
 return cls().set_value(has_pending_actions)


QueryPendingActionsFilter = Annotated[_PendingActionsFilter, Depends(_PendingActionsFilter.depends)]

# DagRun
QueryLastDagRunStateFilter = Annotated[
 FilterParam[DagRunState | None],
Expand DownExpand Up@@ -748,7 +784,7 @@ def _transform_dag_run_types(types: list[str] | None) -> list[DagRunType | None]
 ),
]

# DAGTags
# DagTags
QueryDagTagPatternSearch = Annotated[
 _SearchParam, Depends(search_param_factory(DagTag.name, "tag_name_pattern"))
]
Expand Down

2 changes: 2 additions & 0 deletions
airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,13 +20,15 @@
from airflow.api_fastapi.core_api.base import BaseModel
from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse
from airflow.api_fastapi.core_api.datamodels.hitl import HITLDetail


class DAGWithLatestDagRunsResponse(DAGResponse):
 """DAG with latest dag runs response serializer."""

 asset_expression: dict | None
 latest_dag_runs: list[DAGRunResponse]
 pending_actions: list[HITLDetail]


class DAGWithLatestDagRunsCollectionResponse(BaseModel):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '
Add pendingActions onto ui/dags by bbovenzi · Pull Request #55041 · apache/airflow · GitHub
Skip to content
Add pendingActions onto ui/dags#55041

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

bbovenzi
merged 8 commits into

from

Sep 2, 2025
 Merged


Add pendingActions onto ui/dags
#55041
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
fe10736

Add pendingActions onto ui/dags


bbovenzi Aug 28, 2025
1641ad3

refactor(api_fastapi): unify usage of the term Dag


Lee-W Aug 29, 2025
804043c

refactor(api_fastapi): simplify HITL loading logic


Lee-W Aug 29, 2025
11ba363

test: add test case test_should_return_200_with_hitl


Lee-W Aug 29, 2025
bd17685

Refactor naming scheme and filter


bbovenzi Aug 29, 2025
98eab6f

fix(hitl): fix has_pending_actions to use >= 1 and == 0 to query


Lee-W Sep 1, 2025
d9597b7

feat(zh-tw): update Taiwanese Mandarin translation


Lee-W Sep 1, 2025
60e13bc

Reduce i18n surface area and parameterize dags test


bbovenzi Sep 2, 2025

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


54 changes: 45 additions & 9 deletions
airflow-core/src/airflow/api_fastapi/common/parameters.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -114,7 +114,7 @@ def depends(cls, offset: NonNegativeInt = 0) -> OffsetFilter:


class _FavoriteFilter(BaseParam[bool]):
 """Filter DAGs by favorite status."""
 """Filter Dags by favorite status."""

 def __init__(self, user_id: str, value: T | None = None, skip_none: bool = True) -> None:
 super().__init__(skip_none=skip_none)
Expand DownExpand Up@@ -607,7 +607,7 @@ def depends_float(
DateTimeQuery = Annotated[str, AfterValidator(_safe_parse_datetime)]
OptionalDateTimeQuery = Annotated[str | None, AfterValidator(_safe_parse_datetime_optional)]

# DAG
# Dag
QueryLimit = Annotated[LimitFilter, Depends(LimitFilter.depends)]
QueryOffset = Annotated[OffsetFilter, Depends(OffsetFilter.depends)]
QueryPausedFilter = Annotated[
Expand DownExpand Up@@ -638,7 +638,7 @@ def depends_float(


class _HasAssetScheduleFilter(BaseParam[bool]):
 """Filter DAGs that have asset-based scheduling."""
 """Filter Dags that have asset-based scheduling."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -647,22 +647,22 @@ def to_orm(self, select: Select) -> Select:
 asset_ref_subquery = sql_select(DagScheduleAssetReference.dag_id).distinct()

 if self.value:
 # Filter DAGs that have asset-based scheduling
 # Filter Dags that have asset-based scheduling
 return select.where(DagModel.dag_id.in_(asset_ref_subquery))

 # Filter DAGs that do NOT have asset-based scheduling
 # Filter Dags that do NOT have asset-based scheduling
Comment thread
Lee-W marked this conversation as resolved.

 return select.where(DagModel.dag_id.notin_(asset_ref_subquery))

 @classmethod
 def depends(
 cls,
 has_asset_schedule: bool | None = Query(None, description="Filter DAGs with asset-based scheduling"),
 has_asset_schedule: bool | None = Query(None, description="Filter Dags with asset-based scheduling"),
 ) -> _HasAssetScheduleFilter:
 return cls().set_value(has_asset_schedule)


class _AssetDependencyFilter(BaseParam[str]):
 """Filter DAGs by specific asset dependencies."""
 """Filter Dags by specific asset dependencies."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -681,7 +681,7 @@ def to_orm(self, select: Select) -> Select:
 def depends(
 cls,
 asset_dependency: str | None = Query(
 None, description="Filter DAGs by asset dependency (name or URI)"
 None, description="Filter Dags by asset dependency (name or URI)"
 ),
 ) -> _AssetDependencyFilter:
 return cls().set_value(asset_dependency)
Expand All@@ -690,6 +690,42 @@ def depends(
QueryHasAssetScheduleFilter = Annotated[_HasAssetScheduleFilter, Depends(_HasAssetScheduleFilter.depends)]
QueryAssetDependencyFilter = Annotated[_AssetDependencyFilter, Depends(_AssetDependencyFilter.depends)]


class _PendingActionsFilter(BaseParam[bool]):
 """Filter Dags by having pending HITL actions (more than 1)."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
 return select

 from airflow.models.hitl import HITLDetail
 from airflow.models.taskinstance import TaskInstance

 # Join with HITLDetail and TaskInstance to find Dags
 pending_actions_count_subquery = (
 sql_select(func.count(HITLDetail.ti_id))
 .join(TaskInstance, HITLDetail.ti_id == TaskInstance.id)
 .where(HITLDetail.response_at.is_(None))
 .where(TaskInstance.dag_id == DagModel.dag_id)
 .scalar_subquery()
 )

 if self.value is True:
 # Filter to show only Dags with pending actions
 where_clause = pending_actions_count_subquery >= 1
 else:
 # Filter to show only Dags without pending actions
 where_clause = pending_actions_count_subquery == 0

 return select.where(where_clause)

 @classmethod
 def depends(cls, has_pending_actions: bool | None = Query(None)) -> _PendingActionsFilter:
 return cls().set_value(has_pending_actions)


QueryPendingActionsFilter = Annotated[_PendingActionsFilter, Depends(_PendingActionsFilter.depends)]

# DagRun
QueryLastDagRunStateFilter = Annotated[
 FilterParam[DagRunState | None],
Expand DownExpand Up@@ -748,7 +784,7 @@ def _transform_dag_run_types(types: list[str] | None) -> list[DagRunType | None]
 ),
]

# DAGTags
# DagTags
QueryDagTagPatternSearch = Annotated[
 _SearchParam, Depends(search_param_factory(DagTag.name, "tag_name_pattern"))
]
Expand Down

2 changes: 2 additions & 0 deletions
airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,13 +20,15 @@
from airflow.api_fastapi.core_api.base import BaseModel
from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse
from airflow.api_fastapi.core_api.datamodels.hitl import HITLDetail


class DAGWithLatestDagRunsResponse(DAGResponse):
 """DAG with latest dag runs response serializer."""

 asset_expression: dict | None
 latest_dag_runs: list[DAGRunResponse]
 pending_actions: list[HITLDetail]


class DAGWithLatestDagRunsCollectionResponse(BaseModel):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add pendingActions onto ui/dags by bbovenzi · Pull Request #55041 · apache/airflow · GitHub
Skip to content
Add pendingActions onto ui/dags#55041

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

bbovenzi
merged 8 commits into

from

Sep 2, 2025
 Merged


Add pendingActions onto ui/dags
#55041
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
fe10736

Add pendingActions onto ui/dags


bbovenzi Aug 28, 2025
1641ad3

refactor(api_fastapi): unify usage of the term Dag


Lee-W Aug 29, 2025
804043c

refactor(api_fastapi): simplify HITL loading logic


Lee-W Aug 29, 2025
11ba363

test: add test case test_should_return_200_with_hitl


Lee-W Aug 29, 2025
bd17685

Refactor naming scheme and filter


bbovenzi Aug 29, 2025
98eab6f

fix(hitl): fix has_pending_actions to use >= 1 and == 0 to query


Lee-W Sep 1, 2025
d9597b7

feat(zh-tw): update Taiwanese Mandarin translation


Lee-W Sep 1, 2025
60e13bc

Reduce i18n surface area and parameterize dags test


bbovenzi Sep 2, 2025

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


54 changes: 45 additions & 9 deletions
airflow-core/src/airflow/api_fastapi/common/parameters.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -114,7 +114,7 @@ def depends(cls, offset: NonNegativeInt = 0) -> OffsetFilter:


class _FavoriteFilter(BaseParam[bool]):
 """Filter DAGs by favorite status."""
 """Filter Dags by favorite status."""

 def __init__(self, user_id: str, value: T | None = None, skip_none: bool = True) -> None:
 super().__init__(skip_none=skip_none)
Expand DownExpand Up@@ -607,7 +607,7 @@ def depends_float(
DateTimeQuery = Annotated[str, AfterValidator(_safe_parse_datetime)]
OptionalDateTimeQuery = Annotated[str | None, AfterValidator(_safe_parse_datetime_optional)]

# DAG
# Dag
QueryLimit = Annotated[LimitFilter, Depends(LimitFilter.depends)]
QueryOffset = Annotated[OffsetFilter, Depends(OffsetFilter.depends)]
QueryPausedFilter = Annotated[
Expand DownExpand Up@@ -638,7 +638,7 @@ def depends_float(


class _HasAssetScheduleFilter(BaseParam[bool]):
 """Filter DAGs that have asset-based scheduling."""
 """Filter Dags that have asset-based scheduling."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -647,22 +647,22 @@ def to_orm(self, select: Select) -> Select:
 asset_ref_subquery = sql_select(DagScheduleAssetReference.dag_id).distinct()

 if self.value:
 # Filter DAGs that have asset-based scheduling
 # Filter Dags that have asset-based scheduling
 return select.where(DagModel.dag_id.in_(asset_ref_subquery))

 # Filter DAGs that do NOT have asset-based scheduling
 # Filter Dags that do NOT have asset-based scheduling
Comment thread
Lee-W marked this conversation as resolved.

 return select.where(DagModel.dag_id.notin_(asset_ref_subquery))

 @classmethod
 def depends(
 cls,
 has_asset_schedule: bool | None = Query(None, description="Filter DAGs with asset-based scheduling"),
 has_asset_schedule: bool | None = Query(None, description="Filter Dags with asset-based scheduling"),
 ) -> _HasAssetScheduleFilter:
 return cls().set_value(has_asset_schedule)


class _AssetDependencyFilter(BaseParam[str]):
 """Filter DAGs by specific asset dependencies."""
 """Filter Dags by specific asset dependencies."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -681,7 +681,7 @@ def to_orm(self, select: Select) -> Select:
 def depends(
 cls,
 asset_dependency: str | None = Query(
 None, description="Filter DAGs by asset dependency (name or URI)"
 None, description="Filter Dags by asset dependency (name or URI)"
 ),
 ) -> _AssetDependencyFilter:
 return cls().set_value(asset_dependency)
Expand All@@ -690,6 +690,42 @@ def depends(
QueryHasAssetScheduleFilter = Annotated[_HasAssetScheduleFilter, Depends(_HasAssetScheduleFilter.depends)]
QueryAssetDependencyFilter = Annotated[_AssetDependencyFilter, Depends(_AssetDependencyFilter.depends)]


class _PendingActionsFilter(BaseParam[bool]):
 """Filter Dags by having pending HITL actions (more than 1)."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
 return select

 from airflow.models.hitl import HITLDetail
 from airflow.models.taskinstance import TaskInstance

 # Join with HITLDetail and TaskInstance to find Dags
 pending_actions_count_subquery = (
 sql_select(func.count(HITLDetail.ti_id))
 .join(TaskInstance, HITLDetail.ti_id == TaskInstance.id)
 .where(HITLDetail.response_at.is_(None))
 .where(TaskInstance.dag_id == DagModel.dag_id)
 .scalar_subquery()
 )

 if self.value is True:
 # Filter to show only Dags with pending actions
 where_clause = pending_actions_count_subquery >= 1
 else:
 # Filter to show only Dags without pending actions
 where_clause = pending_actions_count_subquery == 0

 return select.where(where_clause)

 @classmethod
 def depends(cls, has_pending_actions: bool | None = Query(None)) -> _PendingActionsFilter:
 return cls().set_value(has_pending_actions)


QueryPendingActionsFilter = Annotated[_PendingActionsFilter, Depends(_PendingActionsFilter.depends)]

# DagRun
QueryLastDagRunStateFilter = Annotated[
 FilterParam[DagRunState | None],
Expand DownExpand Up@@ -748,7 +784,7 @@ def _transform_dag_run_types(types: list[str] | None) -> list[DagRunType | None]
 ),
]

# DAGTags
# DagTags
QueryDagTagPatternSearch = Annotated[
 _SearchParam, Depends(search_param_factory(DagTag.name, "tag_name_pattern"))
]
Expand Down

2 changes: 2 additions & 0 deletions
airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,13 +20,15 @@
from airflow.api_fastapi.core_api.base import BaseModel
from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse
from airflow.api_fastapi.core_api.datamodels.hitl import HITLDetail


class DAGWithLatestDagRunsResponse(DAGResponse):
 """DAG with latest dag runs response serializer."""

 asset_expression: dict | None
 latest_dag_runs: list[DAGRunResponse]
 pending_actions: list[HITLDetail]


class DAGWithLatestDagRunsCollectionResponse(BaseModel):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Add pendingActions onto ui/dags by bbovenzi · Pull Request #55041 · apache/airflow · GitHub
Skip to content
Add pendingActions onto ui/dags#55041

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

bbovenzi
merged 8 commits into

from

Sep 2, 2025
 Merged


Add pendingActions onto ui/dags
#55041
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
fe10736

Add pendingActions onto ui/dags


bbovenzi Aug 28, 2025
1641ad3

refactor(api_fastapi): unify usage of the term Dag


Lee-W Aug 29, 2025
804043c

refactor(api_fastapi): simplify HITL loading logic


Lee-W Aug 29, 2025
11ba363

test: add test case test_should_return_200_with_hitl


Lee-W Aug 29, 2025
bd17685

Refactor naming scheme and filter


bbovenzi Aug 29, 2025
98eab6f

fix(hitl): fix has_pending_actions to use >= 1 and == 0 to query


Lee-W Sep 1, 2025
d9597b7

feat(zh-tw): update Taiwanese Mandarin translation


Lee-W Sep 1, 2025
60e13bc

Reduce i18n surface area and parameterize dags test


bbovenzi Sep 2, 2025

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


54 changes: 45 additions & 9 deletions
airflow-core/src/airflow/api_fastapi/common/parameters.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -114,7 +114,7 @@ def depends(cls, offset: NonNegativeInt = 0) -> OffsetFilter:


class _FavoriteFilter(BaseParam[bool]):
 """Filter DAGs by favorite status."""
 """Filter Dags by favorite status."""

 def __init__(self, user_id: str, value: T | None = None, skip_none: bool = True) -> None:
 super().__init__(skip_none=skip_none)
Expand DownExpand Up@@ -607,7 +607,7 @@ def depends_float(
DateTimeQuery = Annotated[str, AfterValidator(_safe_parse_datetime)]
OptionalDateTimeQuery = Annotated[str | None, AfterValidator(_safe_parse_datetime_optional)]

# DAG
# Dag
QueryLimit = Annotated[LimitFilter, Depends(LimitFilter.depends)]
QueryOffset = Annotated[OffsetFilter, Depends(OffsetFilter.depends)]
QueryPausedFilter = Annotated[
Expand DownExpand Up@@ -638,7 +638,7 @@ def depends_float(


class _HasAssetScheduleFilter(BaseParam[bool]):
 """Filter DAGs that have asset-based scheduling."""
 """Filter Dags that have asset-based scheduling."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -647,22 +647,22 @@ def to_orm(self, select: Select) -> Select:
 asset_ref_subquery = sql_select(DagScheduleAssetReference.dag_id).distinct()

 if self.value:
 # Filter DAGs that have asset-based scheduling
 # Filter Dags that have asset-based scheduling
 return select.where(DagModel.dag_id.in_(asset_ref_subquery))

 # Filter DAGs that do NOT have asset-based scheduling
 # Filter Dags that do NOT have asset-based scheduling
Comment thread
Lee-W marked this conversation as resolved.

 return select.where(DagModel.dag_id.notin_(asset_ref_subquery))

 @classmethod
 def depends(
 cls,
 has_asset_schedule: bool | None = Query(None, description="Filter DAGs with asset-based scheduling"),
 has_asset_schedule: bool | None = Query(None, description="Filter Dags with asset-based scheduling"),
 ) -> _HasAssetScheduleFilter:
 return cls().set_value(has_asset_schedule)


class _AssetDependencyFilter(BaseParam[str]):
 """Filter DAGs by specific asset dependencies."""
 """Filter Dags by specific asset dependencies."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -681,7 +681,7 @@ def to_orm(self, select: Select) -> Select:
 def depends(
 cls,
 asset_dependency: str | None = Query(
 None, description="Filter DAGs by asset dependency (name or URI)"
 None, description="Filter Dags by asset dependency (name or URI)"
 ),
 ) -> _AssetDependencyFilter:
 return cls().set_value(asset_dependency)
Expand All@@ -690,6 +690,42 @@ def depends(
QueryHasAssetScheduleFilter = Annotated[_HasAssetScheduleFilter, Depends(_HasAssetScheduleFilter.depends)]
QueryAssetDependencyFilter = Annotated[_AssetDependencyFilter, Depends(_AssetDependencyFilter.depends)]


class _PendingActionsFilter(BaseParam[bool]):
 """Filter Dags by having pending HITL actions (more than 1)."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
 return select

 from airflow.models.hitl import HITLDetail
 from airflow.models.taskinstance import TaskInstance

 # Join with HITLDetail and TaskInstance to find Dags
 pending_actions_count_subquery = (
 sql_select(func.count(HITLDetail.ti_id))
 .join(TaskInstance, HITLDetail.ti_id == TaskInstance.id)
 .where(HITLDetail.response_at.is_(None))
 .where(TaskInstance.dag_id == DagModel.dag_id)
 .scalar_subquery()
 )

 if self.value is True:
 # Filter to show only Dags with pending actions
 where_clause = pending_actions_count_subquery >= 1
 else:
 # Filter to show only Dags without pending actions
 where_clause = pending_actions_count_subquery == 0

 return select.where(where_clause)

 @classmethod
 def depends(cls, has_pending_actions: bool | None = Query(None)) -> _PendingActionsFilter:
 return cls().set_value(has_pending_actions)


QueryPendingActionsFilter = Annotated[_PendingActionsFilter, Depends(_PendingActionsFilter.depends)]

# DagRun
QueryLastDagRunStateFilter = Annotated[
 FilterParam[DagRunState | None],
Expand DownExpand Up@@ -748,7 +784,7 @@ def _transform_dag_run_types(types: list[str] | None) -> list[DagRunType | None]
 ),
]

# DAGTags
# DagTags
QueryDagTagPatternSearch = Annotated[
 _SearchParam, Depends(search_param_factory(DagTag.name, "tag_name_pattern"))
]
Expand Down

2 changes: 2 additions & 0 deletions
airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,13 +20,15 @@
from airflow.api_fastapi.core_api.base import BaseModel
from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse
from airflow.api_fastapi.core_api.datamodels.hitl import HITLDetail


class DAGWithLatestDagRunsResponse(DAGResponse):
 """DAG with latest dag runs response serializer."""

 asset_expression: dict | None
 latest_dag_runs: list[DAGRunResponse]
 pending_actions: list[HITLDetail]


class DAGWithLatestDagRunsCollectionResponse(BaseModel):
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();
Add pendingActions onto ui/dags by bbovenzi · Pull Request #55041 · apache/airflow · GitHub
Skip to content
Add pendingActions onto ui/dags#55041

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

bbovenzi
merged 8 commits into

from

Sep 2, 2025
 Merged


Add pendingActions onto ui/dags
#55041
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


8 commits


Select commit
Hold shift + click to select a range
fe10736

Add pendingActions onto ui/dags


bbovenzi Aug 28, 2025
1641ad3

refactor(api_fastapi): unify usage of the term Dag


Lee-W Aug 29, 2025
804043c

refactor(api_fastapi): simplify HITL loading logic


Lee-W Aug 29, 2025
11ba363

test: add test case test_should_return_200_with_hitl


Lee-W Aug 29, 2025
bd17685

Refactor naming scheme and filter


bbovenzi Aug 29, 2025
98eab6f

fix(hitl): fix has_pending_actions to use >= 1 and == 0 to query


Lee-W Sep 1, 2025
d9597b7

feat(zh-tw): update Taiwanese Mandarin translation


Lee-W Sep 1, 2025
60e13bc

Reduce i18n surface area and parameterize dags test


bbovenzi Sep 2, 2025

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


54 changes: 45 additions & 9 deletions
airflow-core/src/airflow/api_fastapi/common/parameters.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -114,7 +114,7 @@ def depends(cls, offset: NonNegativeInt = 0) -> OffsetFilter:


class _FavoriteFilter(BaseParam[bool]):
 """Filter DAGs by favorite status."""
 """Filter Dags by favorite status."""

 def __init__(self, user_id: str, value: T | None = None, skip_none: bool = True) -> None:
 super().__init__(skip_none=skip_none)
Expand DownExpand Up@@ -607,7 +607,7 @@ def depends_float(
DateTimeQuery = Annotated[str, AfterValidator(_safe_parse_datetime)]
OptionalDateTimeQuery = Annotated[str | None, AfterValidator(_safe_parse_datetime_optional)]

# DAG
# Dag
QueryLimit = Annotated[LimitFilter, Depends(LimitFilter.depends)]
QueryOffset = Annotated[OffsetFilter, Depends(OffsetFilter.depends)]
QueryPausedFilter = Annotated[
Expand DownExpand Up@@ -638,7 +638,7 @@ def depends_float(


class _HasAssetScheduleFilter(BaseParam[bool]):
 """Filter DAGs that have asset-based scheduling."""
 """Filter Dags that have asset-based scheduling."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -647,22 +647,22 @@ def to_orm(self, select: Select) -> Select:
 asset_ref_subquery = sql_select(DagScheduleAssetReference.dag_id).distinct()

 if self.value:
 # Filter DAGs that have asset-based scheduling
 # Filter Dags that have asset-based scheduling
 return select.where(DagModel.dag_id.in_(asset_ref_subquery))

 # Filter DAGs that do NOT have asset-based scheduling
 # Filter Dags that do NOT have asset-based scheduling
Comment thread
Lee-W marked this conversation as resolved.

 return select.where(DagModel.dag_id.notin_(asset_ref_subquery))

 @classmethod
 def depends(
 cls,
 has_asset_schedule: bool | None = Query(None, description="Filter DAGs with asset-based scheduling"),
 has_asset_schedule: bool | None = Query(None, description="Filter Dags with asset-based scheduling"),
 ) -> _HasAssetScheduleFilter:
 return cls().set_value(has_asset_schedule)


class _AssetDependencyFilter(BaseParam[str]):
 """Filter DAGs by specific asset dependencies."""
 """Filter Dags by specific asset dependencies."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
Expand All@@ -681,7 +681,7 @@ def to_orm(self, select: Select) -> Select:
 def depends(
 cls,
 asset_dependency: str | None = Query(
 None, description="Filter DAGs by asset dependency (name or URI)"
 None, description="Filter Dags by asset dependency (name or URI)"
 ),
 ) -> _AssetDependencyFilter:
 return cls().set_value(asset_dependency)
Expand All@@ -690,6 +690,42 @@ def depends(
QueryHasAssetScheduleFilter = Annotated[_HasAssetScheduleFilter, Depends(_HasAssetScheduleFilter.depends)]
QueryAssetDependencyFilter = Annotated[_AssetDependencyFilter, Depends(_AssetDependencyFilter.depends)]


class _PendingActionsFilter(BaseParam[bool]):
 """Filter Dags by having pending HITL actions (more than 1)."""

 def to_orm(self, select: Select) -> Select:
 if self.value is None and self.skip_none:
 return select

 from airflow.models.hitl import HITLDetail
 from airflow.models.taskinstance import TaskInstance

 # Join with HITLDetail and TaskInstance to find Dags
 pending_actions_count_subquery = (
 sql_select(func.count(HITLDetail.ti_id))
 .join(TaskInstance, HITLDetail.ti_id == TaskInstance.id)
 .where(HITLDetail.response_at.is_(None))
 .where(TaskInstance.dag_id == DagModel.dag_id)
 .scalar_subquery()
 )

 if self.value is True:
 # Filter to show only Dags with pending actions
 where_clause = pending_actions_count_subquery >= 1
 else:
 # Filter to show only Dags without pending actions
 where_clause = pending_actions_count_subquery == 0

 return select.where(where_clause)

 @classmethod
 def depends(cls, has_pending_actions: bool | None = Query(None)) -> _PendingActionsFilter:
 return cls().set_value(has_pending_actions)


QueryPendingActionsFilter = Annotated[_PendingActionsFilter, Depends(_PendingActionsFilter.depends)]

# DagRun
QueryLastDagRunStateFilter = Annotated[
 FilterParam[DagRunState | None],
Expand DownExpand Up@@ -748,7 +784,7 @@ def _transform_dag_run_types(types: list[str] | None) -> list[DagRunType | None]
 ),
]

# DAGTags
# DagTags
QueryDagTagPatternSearch = Annotated[
 _SearchParam, Depends(search_param_factory(DagTag.name, "tag_name_pattern"))
]
Expand Down

2 changes: 2 additions & 0 deletions
airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/dags.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -20,13 +20,15 @@
from airflow.api_fastapi.core_api.base import BaseModel
from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
from airflow.api_fastapi.core_api.datamodels.dags import DAGResponse
from airflow.api_fastapi.core_api.datamodels.hitl import HITLDetail


class DAGWithLatestDagRunsResponse(DAGResponse):
 """DAG with latest dag runs response serializer."""

 asset_expression: dict | None
 latest_dag_runs: list[DAGRunResponse]
 pending_actions: list[HITLDetail]


class DAGWithLatestDagRunsCollectionResponse(BaseModel):
Expand Down
Loading