Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -1634,6 +1634,38 @@ paths:
- type: integer
- type: 'null'
title: Max Active Runs Lt
- name: duration_gte
in: query
required: false
schema:
anyOf:
- type: number
- type: 'null'
title: Duration Gte
- name: duration_gt
in: query
required: false
schema:
anyOf:
- type: number
- type: 'null'
title: Duration Gt
- name: duration_lte
in: query
required: false
schema:
anyOf:
- type: number
- type: 'null'
title: Duration Lte
- name: duration_lt
in: query
required: false
schema:
anyOf:
- type: number
- type: 'null'
title: Duration Lt
- name: reprocess_behavior
in: query
required: false
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
SortParam,
datetime_range_filter_factory,
filter_param_factory,
float_range_filter_factory,
int_range_filter_factory,
)
from airflow.api_fastapi.common.router import AirflowRouter
Expand Down Expand Up @@ -60,6 +61,7 @@ def list_backfills_ui(
created_at: Annotated[RangeFilter, Depends(datetime_range_filter_factory("created_at", Backfill))],
completed_at: Annotated[RangeFilter, Depends(datetime_range_filter_factory("completed_at", Backfill))],
max_active_runs: Annotated[RangeFilter, Depends(int_range_filter_factory("max_active_runs", Backfill))],
duration_range: Annotated[RangeFilter, Depends(float_range_filter_factory("duration", Backfill))],
reprocess_behavior: Annotated[
FilterParam, Depends(filter_param_factory(Backfill.reprocess_behavior, ReprocessBehavior | None))
],
Expand All @@ -86,6 +88,7 @@ def list_backfills_ui(
created_at,
completed_at,
max_active_runs,
duration_range,
reprocess_behavior,
],
order_by=order_by,
Expand Down
43 changes: 40 additions & 3 deletions airflow-core/src/airflow/models/backfill.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
from collections.abc import Iterable
from datetime import datetime
from enum import Enum
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any

import sqlalchemy as sa
import structlog
Expand All @@ -37,18 +37,26 @@
Integer,
String,
UniqueConstraint,
case,
delete,
func,
select,
text,
)
from sqlalchemy.exc import IntegrityError, OperationalError
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import Mapped, mapped_column, relationship, validates

from airflow._shared.timezones import timezone
from airflow.exceptions import AirflowException, DagNotFound, DagRunTypeNotAllowed
from airflow.models.base import Base, StringID
from airflow.utils.session import create_session
from airflow.utils.sqlalchemy import UtcDateTime, is_lock_not_available_error, with_row_locks
from airflow.utils.db import get_dialect_name
from airflow.utils.session import NEW_SESSION, create_session, provide_session
from airflow.utils.sqlalchemy import (
UtcDateTime,
is_lock_not_available_error,
with_row_locks,
)
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunTriggeredByType, DagRunType

Expand Down Expand Up @@ -203,6 +211,35 @@ class Backfill(Base):
def __repr__(self):
return f"Backfill({self.dag_id=}, {self.from_date=}, {self.to_date=})"

@hybrid_property
def duration(self) -> float | None:
if self.completed_at and self.created_at:
return (self.completed_at - self.created_at).total_seconds()
return None

@duration.expression # type: ignore[no-redef]
def duration(cls) -> Any:
@provide_session
def _get_dialect(*, session: Session = NEW_SESSION) -> str:
return get_dialect_name(session=session)

dialect_name = _get_dialect()

duration_expr: Any
if dialect_name == "mysql":
duration_expr = func.timestampdiff(text("SECOND"), cls.created_at, cls.completed_at)
elif dialect_name == "sqlite":
duration_expr = (func.julianday(cls.completed_at) - func.julianday(cls.created_at)) * 86400
else:
duration_expr = func.extract("epoch", cls.completed_at - cls.created_at) # type: ignore[operator, arg-type]

when_condition = (
cls.completed_at.isnot(None) & cls.created_at.isnot(None), # type: ignore[union-attr, attr-defined]
duration_expr,
)

return case(when_condition, else_=None)


class BackfillDagRunExceptionReason(str, Enum):
"""
Expand Down
8 changes: 6 additions & 2 deletions airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,7 @@ export const UseBackfillServiceListBackfillDagRunsKeyFn = ({ backfillId, limit,
export type BackfillServiceListBackfillsUiDefaultResponse = Awaited<ReturnType<typeof BackfillService.listBackfillsUi>>;
export type BackfillServiceListBackfillsUiQueryResult<TData = BackfillServiceListBackfillsUiDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
export const useBackfillServiceListBackfillsUiKey = "BackfillServiceListBackfillsUi";
export const UseBackfillServiceListBackfillsUiKeyFn = ({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }: {
export const UseBackfillServiceListBackfillsUiKeyFn = ({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, durationGt, durationGte, durationLt, durationLte, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }: {
active?: boolean;
completedAtGt?: string;
completedAtGte?: string;
Expand All @@ -150,6 +150,10 @@ export const UseBackfillServiceListBackfillsUiKeyFn = ({ active, completedAtGt,
createdAtLt?: string;
createdAtLte?: string;
dagId?: string;
durationGt?: number;
durationGte?: number;
durationLt?: number;
durationLte?: number;
fromDateGt?: string;
fromDateGte?: string;
fromDateLt?: string;
Expand All @@ -166,7 +170,7 @@ export const UseBackfillServiceListBackfillsUiKeyFn = ({ active, completedAtGt,
toDateGte?: string;
toDateLt?: string;
toDateLte?: string;
} = {}, queryKey?: Array<unknown>) => [useBackfillServiceListBackfillsUiKey, ...(queryKey ?? [{ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }])];
} = {}, queryKey?: Array<unknown>) => [useBackfillServiceListBackfillsUiKey, ...(queryKey ?? [{ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, durationGt, durationGte, durationLt, durationLte, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }])];
export type ConnectionServiceGetConnectionDefaultResponse = Awaited<ReturnType<typeof ConnectionService.getConnection>>;
export type ConnectionServiceGetConnectionQueryResult<TData = ConnectionServiceGetConnectionDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
export const useConnectionServiceGetConnectionKey = "ConnectionServiceGetConnection";
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -279,14 +279,18 @@ export const ensureUseBackfillServiceListBackfillDagRunsData = (queryClient: Que
* @param data.maxActiveRunsGt
* @param data.maxActiveRunsLte
* @param data.maxActiveRunsLt
* @param data.durationGte
* @param data.durationGt
* @param data.durationLte
* @param data.durationLt
* @param data.reprocessBehavior
* @param data.orderBy Attributes to order by, multi criteria sort is supported. Prefix with `-` for descending order. Supported attributes: `id`
* @param data.dagId
* @param data.active
* @returns BackfillCollectionResponse Successful Response
* @throws ApiError
*/
export const ensureUseBackfillServiceListBackfillsUiData = (queryClient: QueryClient, { active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }: {
export const ensureUseBackfillServiceListBackfillsUiData = (queryClient: QueryClient, { active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, durationGt, durationGte, durationLt, durationLte, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }: {
active?: boolean;
completedAtGt?: string;
completedAtGte?: string;
Expand All @@ -297,6 +301,10 @@ export const ensureUseBackfillServiceListBackfillsUiData = (queryClient: QueryCl
createdAtLt?: string;
createdAtLte?: string;
dagId?: string;
durationGt?: number;
durationGte?: number;
durationLt?: number;
durationLte?: number;
fromDateGt?: string;
fromDateGte?: string;
fromDateLt?: string;
Expand All @@ -313,7 +321,7 @@ export const ensureUseBackfillServiceListBackfillsUiData = (queryClient: QueryCl
toDateGte?: string;
toDateLt?: string;
toDateLte?: string;
} = {}) => queryClient.ensureQueryData({ queryKey: Common.UseBackfillServiceListBackfillsUiKeyFn({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }), queryFn: () => BackfillService.listBackfillsUi({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }) });
} = {}) => queryClient.ensureQueryData({ queryKey: Common.UseBackfillServiceListBackfillsUiKeyFn({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, durationGt, durationGte, durationLt, durationLte, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }), queryFn: () => BackfillService.listBackfillsUi({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, durationGt, durationGte, durationLt, durationLte, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }) });
/**
* Get Connection
* Get a connection entry.
Expand Down
12 changes: 10 additions & 2 deletions airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -279,14 +279,18 @@ export const prefetchUseBackfillServiceListBackfillDagRuns = (queryClient: Query
* @param data.maxActiveRunsGt
* @param data.maxActiveRunsLte
* @param data.maxActiveRunsLt
* @param data.durationGte
* @param data.durationGt
* @param data.durationLte
* @param data.durationLt
* @param data.reprocessBehavior
* @param data.orderBy Attributes to order by, multi criteria sort is supported. Prefix with `-` for descending order. Supported attributes: `id`
* @param data.dagId
* @param data.active
* @returns BackfillCollectionResponse Successful Response
* @throws ApiError
*/
export const prefetchUseBackfillServiceListBackfillsUi = (queryClient: QueryClient, { active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }: {
export const prefetchUseBackfillServiceListBackfillsUi = (queryClient: QueryClient, { active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, durationGt, durationGte, durationLt, durationLte, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }: {
active?: boolean;
completedAtGt?: string;
completedAtGte?: string;
Expand All @@ -297,6 +301,10 @@ export const prefetchUseBackfillServiceListBackfillsUi = (queryClient: QueryClie
createdAtLt?: string;
createdAtLte?: string;
dagId?: string;
durationGt?: number;
durationGte?: number;
durationLt?: number;
durationLte?: number;
fromDateGt?: string;
fromDateGte?: string;
fromDateLt?: string;
Expand All @@ -313,7 +321,7 @@ export const prefetchUseBackfillServiceListBackfillsUi = (queryClient: QueryClie
toDateGte?: string;
toDateLt?: string;
toDateLte?: string;
} = {}) => queryClient.prefetchQuery({ queryKey: Common.UseBackfillServiceListBackfillsUiKeyFn({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }), queryFn: () => BackfillService.listBackfillsUi({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }) });
} = {}) => queryClient.prefetchQuery({ queryKey: Common.UseBackfillServiceListBackfillsUiKeyFn({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, durationGt, durationGte, durationLt, durationLte, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }), queryFn: () => BackfillService.listBackfillsUi({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, durationGt, durationGte, durationLt, durationLte, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }) });
/**
* Get Connection
* Get a connection entry.
Expand Down
12 changes: 10 additions & 2 deletions airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
Original file line number Diff line number Diff line change
Expand Up @@ -279,14 +279,18 @@ export const useBackfillServiceListBackfillDagRuns = <TData = Common.BackfillSer
* @param data.maxActiveRunsGt
* @param data.maxActiveRunsLte
* @param data.maxActiveRunsLt
* @param data.durationGte
* @param data.durationGt
* @param data.durationLte
* @param data.durationLt
* @param data.reprocessBehavior
* @param data.orderBy Attributes to order by, multi criteria sort is supported. Prefix with `-` for descending order. Supported attributes: `id`
* @param data.dagId
* @param data.active
* @returns BackfillCollectionResponse Successful Response
* @throws ApiError
*/
export const useBackfillServiceListBackfillsUi = <TData = Common.BackfillServiceListBackfillsUiDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }: {
export const useBackfillServiceListBackfillsUi = <TData = Common.BackfillServiceListBackfillsUiDefaultResponse, TError = unknown, TQueryKey extends Array<unknown> = unknown[]>({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, durationGt, durationGte, durationLt, durationLte, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }: {
active?: boolean;
completedAtGt?: string;
completedAtGte?: string;
Expand All @@ -297,6 +301,10 @@ export const useBackfillServiceListBackfillsUi = <TData = Common.BackfillService
createdAtLt?: string;
createdAtLte?: string;
dagId?: string;
durationGt?: number;
durationGte?: number;
durationLt?: number;
durationLte?: number;
fromDateGt?: string;
fromDateGte?: string;
fromDateLt?: string;
Expand All @@ -313,7 +321,7 @@ export const useBackfillServiceListBackfillsUi = <TData = Common.BackfillService
toDateGte?: string;
toDateLt?: string;
toDateLte?: string;
} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey: Common.UseBackfillServiceListBackfillsUiKeyFn({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }, queryKey), queryFn: () => BackfillService.listBackfillsUi({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }) as TData, ...options });
} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey: Common.UseBackfillServiceListBackfillsUiKeyFn({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, durationGt, durationGte, durationLt, durationLte, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }, queryKey), queryFn: () => BackfillService.listBackfillsUi({ active, completedAtGt, completedAtGte, completedAtLt, completedAtLte, createdAtGt, createdAtGte, createdAtLt, createdAtLte, dagId, durationGt, durationGte, durationLt, durationLte, fromDateGt, fromDateGte, fromDateLt, fromDateLte, limit, maxActiveRunsGt, maxActiveRunsGte, maxActiveRunsLt, maxActiveRunsLte, offset, orderBy, reprocessBehavior, toDateGt, toDateGte, toDateLt, toDateLte }) as TData, ...options });
/**
* Get Connection
* Get a connection entry.
Expand Down
Loading
Loading