You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ds...@apache.org on 2024/11/04 17:06:38 UTC
(airflow) branch main updated: AIP-84 Migrate backfill API to fast api (#43496)
This is an automated email from the ASF dual-hosted git repository.
dstandish pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 30147ffd7d8 AIP-84 Migrate backfill API to fast api (#43496)
30147ffd7d8 is described below
commit 30147ffd7d8730cad018d851cc4d29a47a2673a0
Author: Daniel Standish <15...@users.noreply.github.com>
AuthorDate: Mon Nov 4 09:06:31 2024 -0800
AIP-84 Migrate backfill API to fast api (#43496)
---
.../api_connexion/endpoints/backfill_endpoint.py | 183 -----
airflow/api_connexion/openapi/v1.yaml | 149 ----
airflow/api_connexion/schemas/backfill_schema.py | 100 ---
.../api_fastapi/core_api/openapi/v1-generated.yaml | 870 ++++++++++++++-------
.../api_fastapi/core_api/routes/public/__init__.py | 3 +
.../core_api/routes/public/backfills.py | 187 +++++
.../api_fastapi/core_api/serializers/backfills.py | 59 ++
airflow/models/backfill.py | 35 +-
airflow/ui/openapi-gen/queries/common.ts | 191 +++--
airflow/ui/openapi-gen/queries/prefetch.ts | 182 +++--
airflow/ui/openapi-gen/queries/queries.ts | 500 ++++++++----
airflow/ui/openapi-gen/queries/suspense.ts | 235 ++++--
airflow/ui/openapi-gen/requests/schemas.gen.ts | 51 ++
airflow/ui/openapi-gen/requests/services.gen.ts | 434 ++++++----
airflow/ui/openapi-gen/requests/types.gen.ts | 421 +++++++---
airflow/www/static/js/types/api-generated.ts | 181 -----
.../endpoints/test_backfill_endpoint.py | 442 -----------
.../api_connexion/schemas/test_backfill_schema.py | 55 --
.../core_api/routes/public/test_backfills.py | 309 ++++++++
tests/models/test_backfill.py | 36 -
20 files changed, 2597 insertions(+), 2026 deletions(-)
diff --git a/airflow/api_connexion/endpoints/backfill_endpoint.py b/airflow/api_connexion/endpoints/backfill_endpoint.py
deleted file mode 100644
index b53d461dc5d..00000000000
--- a/airflow/api_connexion/endpoints/backfill_endpoint.py
+++ /dev/null
@@ -1,183 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-from __future__ import annotations
-
-import logging
-from functools import wraps
-from typing import TYPE_CHECKING
-
-from flask import request
-from marshmallow import ValidationError
-from sqlalchemy import select
-
-from airflow.api_connexion import security
-from airflow.api_connexion.exceptions import BadRequest, Conflict, NotFound
-from airflow.api_connexion.schemas.backfill_schema import (
- BackfillCollection,
- backfill_collection_schema,
- backfill_schema,
-)
-from airflow.models.backfill import (
- AlreadyRunningBackfill,
- Backfill,
- ReprocessBehavior,
- _cancel_backfill,
- _create_backfill,
-)
-from airflow.utils.session import NEW_SESSION, provide_session
-from airflow.www.decorators import action_logging
-
-if TYPE_CHECKING:
- from datetime import datetime
-
- from sqlalchemy.orm import Session
-
- from airflow.api_connexion.types import APIResponse
-
-log = logging.getLogger(__name__)
-
-RESOURCE_EVENT_PREFIX = "dag"
-
-
-def backfill_to_dag(func):
- """
- Enrich the request with dag_id.
-
- :meta private:
- """
-
- @wraps(func)
- def wrapper(*, backfill_id, session, **kwargs):
- backfill = session.get(Backfill, backfill_id)
- if not backfill:
- raise NotFound("Backfill not found")
- return func(dag_id=backfill.dag_id, backfill_id=backfill_id, session=session, **kwargs)
-
- return wrapper
-
-
-@security.requires_access_dag("GET")
-@action_logging
-@provide_session
-def list_backfills(dag_id, session):
- backfills = session.scalars(select(Backfill).where(Backfill.dag_id == dag_id)).all()
- obj = BackfillCollection(
- backfills=backfills,
- total_entries=len(backfills),
- )
- return backfill_collection_schema.dump(obj)
-
-
-@provide_session
-@backfill_to_dag
-@security.requires_access_dag("PUT")
-@action_logging
-def pause_backfill(*, backfill_id, session, **kwargs):
- br = session.get(Backfill, backfill_id)
- if br.completed_at:
- raise Conflict("Backfill is already completed.")
- if br.is_paused is False:
- br.is_paused = True
- session.commit()
- return backfill_schema.dump(br)
-
-
-@provide_session
-@backfill_to_dag
-@security.requires_access_dag("PUT")
-@action_logging
-def unpause_backfill(*, backfill_id, session, **kwargs):
- br = session.get(Backfill, backfill_id)
- if br.completed_at:
- raise Conflict("Backfill is already completed.")
- if br.is_paused:
- br.is_paused = False
- session.commit()
- return backfill_schema.dump(br)
-
-
-@provide_session
-@backfill_to_dag
-@security.requires_access_dag("GET")
-@action_logging
-def get_backfill(*, backfill_id: int, session: Session = NEW_SESSION, **kwargs):
- backfill = session.get(Backfill, backfill_id)
- if backfill:
- return backfill_schema.dump(backfill)
- raise NotFound("Backfill not found")
-
-
-def backfill_obj_to_kwargs(f):
- """
- Convert the request body (containing backfill object json) to kwargs.
-
- The main point here is to be compatible with the ``requires_access_dag`` decorator,
- which takes dag_id kwarg and doesn't support json request body.
- """
-
- @wraps(f)
- def inner():
- body = request.json
- try:
- obj = backfill_schema.load(body)
- except ValidationError as err:
- raise BadRequest(detail=str(err.messages))
- return f(**obj)
-
- return inner
-
-
-@backfill_obj_to_kwargs
-@security.requires_access_dag("PUT")
-@action_logging
-def create_backfill(
- dag_id: str,
- from_date: datetime,
- to_date: datetime,
- max_active_runs: int = 10,
- reverse: bool = False,
- dag_run_conf: dict | None = None,
- reprocess_behavior: ReprocessBehavior | None = None,
-) -> APIResponse:
- try:
- backfill_obj = _create_backfill(
- dag_id=dag_id,
- from_date=from_date,
- to_date=to_date,
- max_active_runs=max_active_runs,
- reverse=reverse,
- dag_run_conf=dag_run_conf,
- reprocess_behavior=reprocess_behavior,
- )
- return backfill_schema.dump(backfill_obj)
- except AlreadyRunningBackfill:
- raise Conflict(f"There is already a running backfill for dag {dag_id}")
-
-
-@provide_session
-@backfill_to_dag
-@security.requires_access_dag("PUT")
-@action_logging
-def cancel_backfill(
- *,
- backfill_id,
- session: Session = NEW_SESSION, # used by backfill_to_dag decorator
- **kwargs,
-):
- br = _cancel_backfill(backfill_id=backfill_id)
- return backfill_schema.dump(br)
diff --git a/airflow/api_connexion/openapi/v1.yaml b/airflow/api_connexion/openapi/v1.yaml
index 0ec0a3cdcb6..c884c159541 100644
--- a/airflow/api_connexion/openapi/v1.yaml
+++ b/airflow/api_connexion/openapi/v1.yaml
@@ -245,155 +245,6 @@ servers:
description: Apache Airflow Stable API.
paths:
- # Database entities
- /backfills:
- get:
- summary: List backfills
- x-openapi-router-controller: airflow.api_connexion.endpoints.backfill_endpoint
- operationId: list_backfills
- tags: [Backfill]
- parameters:
- - name: dag_id
- in: query
- schema:
- type: string
- required: true
- description: |
- List backfills for this dag.
- responses:
- "200":
- description: Success.
- content:
- application/json:
- schema:
- $ref: "#/components/schemas/BackfillCollection"
- "401":
- $ref: "#/components/responses/Unauthenticated"
- "403":
- $ref: "#/components/responses/PermissionDenied"
-
- post:
- summary: Create a backfill job.
- x-openapi-router-controller: airflow.api_connexion.endpoints.backfill_endpoint
- operationId: create_backfill
- tags: [Backfill]
- requestBody:
- required: true
- content:
- application/json:
- schema:
- $ref: "#/components/schemas/Backfill"
- responses:
- "200":
- description: Success.
- content:
- application/json:
- schema:
- $ref: "#/components/schemas/Backfill"
- "400":
- $ref: "#/components/responses/BadRequest"
- "401":
- $ref: "#/components/responses/Unauthenticated"
- "403":
- $ref: "#/components/responses/PermissionDenied"
-
- /backfills/{backfill_id}:
- parameters:
- - $ref: "#/components/parameters/BackfillIdPath"
- get:
- summary: Get a backfill
- x-openapi-router-controller: airflow.api_connexion.endpoints.backfill_endpoint
- operationId: get_backfill
- tags: [Backfill]
- responses:
- "200":
- description: Success.
- content:
- application/json:
- schema:
- $ref: "#/components/schemas/Backfill"
- "401":
- $ref: "#/components/responses/Unauthenticated"
- "403":
- $ref: "#/components/responses/PermissionDenied"
- "404":
- $ref: "#/components/responses/NotFound"
-
- /backfills/{backfill_id}/pause:
- parameters:
- - $ref: "#/components/parameters/BackfillIdPath"
- post:
- summary: Pause a backfill
- x-openapi-router-controller: airflow.api_connexion.endpoints.backfill_endpoint
- operationId: pause_backfill
- tags: [Backfill]
- responses:
- "200":
- description: Success.
- content:
- application/json:
- schema:
- $ref: "#/components/schemas/Backfill"
- "401":
- $ref: "#/components/responses/Unauthenticated"
- "403":
- $ref: "#/components/responses/PermissionDenied"
- "404":
- $ref: "#/components/responses/NotFound"
- "409":
- $ref: "#/components/responses/Conflict"
-
- /backfills/{backfill_id}/unpause:
- parameters:
- - $ref: "#/components/parameters/BackfillIdPath"
- post:
- summary: Pause a backfill
- x-openapi-router-controller: airflow.api_connexion.endpoints.backfill_endpoint
- operationId: unpause_backfill
- tags: [Backfill]
- responses:
- "200":
- description: Success.
- content:
- application/json:
- schema:
- $ref: "#/components/schemas/Backfill"
- "401":
- $ref: "#/components/responses/Unauthenticated"
- "403":
- $ref: "#/components/responses/PermissionDenied"
- "404":
- $ref: "#/components/responses/NotFound"
- "409":
- $ref: "#/components/responses/Conflict"
-
- /backfills/{backfill_id}/cancel:
- parameters:
- - $ref: "#/components/parameters/BackfillIdPath"
- post:
- summary: Cancel a backfill
- description: |
- When a backfill is cancelled, all queued dag runs will be marked as failed.
- Running dag runs will be allowed to continue.
- x-openapi-router-controller: airflow.api_connexion.endpoints.backfill_endpoint
- operationId: cancel_backfill
- tags: [Backfill]
- responses:
- "200":
- description: Success.
- content:
- application/json:
- schema:
- $ref: "#/components/schemas/Backfill"
- "401":
- $ref: "#/components/responses/Unauthenticated"
- "403":
- $ref: "#/components/responses/PermissionDenied"
- "404":
- $ref: "#/components/responses/NotFound"
- "409":
- $ref: "#/components/responses/Conflict"
-
# Database entities
/connections:
get:
diff --git a/airflow/api_connexion/schemas/backfill_schema.py b/airflow/api_connexion/schemas/backfill_schema.py
deleted file mode 100644
index 9f719846915..00000000000
--- a/airflow/api_connexion/schemas/backfill_schema.py
+++ /dev/null
@@ -1,100 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-from __future__ import annotations
-
-import typing
-from typing import NamedTuple
-
-from marshmallow import Schema, fields, utils, validate
-from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
-
-from airflow.models.backfill import Backfill, BackfillDagRun, ReprocessBehavior
-
-
-class ReprocessBehaviorField(fields.String):
- """Schema for ReprocessBehavior enum."""
-
- def __init__(self, **metadata):
- super().__init__(**metadata)
- self.validators = [validate.OneOf(ReprocessBehavior), *self.validators]
-
- def _serialize(self, value, attr, obj, **kwargs) -> str | None:
- if value is None:
- return None
- return utils.ensure_text_type(ReprocessBehavior(value).value)
-
- def _deserialize(self, value, attr, data, **kwargs) -> typing.Any:
- deser = super()._deserialize(value, attr, data, **kwargs)
- if not deser:
- return None
- return ReprocessBehavior(deser)
-
-
-class BackfillSchema(SQLAlchemySchema):
- """Backfill Schema."""
-
- class Meta:
- """Meta."""
-
- model = Backfill
-
- id = auto_field(dump_only=True)
- dag_id = auto_field()
- from_date = auto_field()
- to_date = auto_field()
- dag_run_conf = fields.Dict(allow_none=True)
- reverse = fields.Boolean()
- is_paused = auto_field()
- reprocess_behavior = ReprocessBehaviorField()
- max_active_runs = auto_field()
- created_at = auto_field()
- completed_at = auto_field()
- updated_at = auto_field()
-
-
-class BackfillDagRunSchema(SQLAlchemySchema):
- """Trigger Schema."""
-
- class Meta:
- """Meta."""
-
- model = BackfillDagRun
-
- id = auto_field(dump_only=True)
- backfill_id = auto_field(dump_only=True)
- dag_run_id = auto_field(dump_only=True)
- sort_ordinal = auto_field(dump_only=True)
-
-
-class BackfillCollection(NamedTuple):
- """List of Backfills with meta."""
-
- backfills: list[Backfill]
- total_entries: int
-
-
-class BackfillCollectionSchema(Schema):
- """Backfill Collection Schema."""
-
- backfills = fields.List(fields.Nested(BackfillSchema))
- total_entries = fields.Int()
-
-
-backfill_schema = BackfillSchema()
-backfill_dag_run_schema = BackfillDagRunSchema()
-backfill_collection_schema = BackfillCollectionSchema()
diff --git a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
index 19a4875e704..a82e34a3f5e 100644
--- a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
+++ b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
@@ -170,67 +170,46 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
- /public/connections/{connection_id}:
- delete:
+ /public/backfills/:
+ get:
tags:
- - Connection
- summary: Delete Connection
- description: Delete a connection entry.
- operationId: delete_connection
+ - Backfill
+ summary: List Backfills
+ operationId: list_backfills
parameters:
- - name: connection_id
- in: path
+ - name: dag_id
+ in: query
required: true
schema:
type: string
- title: Connection Id
- responses:
- '204':
- description: Successful Response
- '401':
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/HTTPExceptionResponse'
- description: Unauthorized
- '403':
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/HTTPExceptionResponse'
- description: Forbidden
- '404':
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/HTTPExceptionResponse'
- description: Not Found
- '422':
- description: Validation Error
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/HTTPValidationError'
- get:
- tags:
- - Connection
- summary: Get Connection
- description: Get a connection entry.
- operationId: get_connection
- parameters:
- - name: connection_id
- in: path
- required: true
+ title: Dag Id
+ - name: limit
+ in: query
+ required: false
+ schema:
+ type: integer
+ default: 100
+ title: Limit
+ - name: offset
+ in: query
+ required: false
+ schema:
+ type: integer
+ default: 0
+ title: Offset
+ - name: order_by
+ in: query
+ required: false
schema:
type: string
- title: Connection Id
+ default: id
+ title: Order By
responses:
'200':
description: Successful Response
content:
application/json:
- schema:
- $ref: '#/components/schemas/ConnectionResponse'
+ schema: {}
'401':
content:
application/json:
@@ -243,54 +222,29 @@ paths:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Forbidden
- '404':
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/HTTPExceptionResponse'
- description: Not Found
'422':
description: Validation Error
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
- /public/connections/:
- get:
+ post:
tags:
- - Connection
- summary: Get Connections
- description: Get all connection entries.
- operationId: get_connections
- parameters:
- - name: limit
- in: query
- required: false
- schema:
- type: integer
- default: 100
- title: Limit
- - name: offset
- in: query
- required: false
- schema:
- type: integer
- default: 0
- title: Offset
- - name: order_by
- in: query
- required: false
- schema:
- type: string
- default: id
- title: Order By
+ - Backfill
+ summary: Create Backfill
+ operationId: create_backfill
+ requestBody:
+ required: true
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/BackfillPostBody'
responses:
'200':
description: Successful Response
content:
application/json:
- schema:
- $ref: '#/components/schemas/ConnectionCollectionResponse'
+ schema: {}
'401':
content:
application/json:
@@ -309,38 +263,37 @@ paths:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Not Found
+ '409':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Conflict
'422':
description: Validation Error
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
- /public/dags/{dag_id}/dagRuns/{dag_run_id}:
+ /public/backfills/{backfill_id}:
get:
tags:
- - DagRun
- summary: Get Dag Run
- operationId: get_dag_run
+ - Backfill
+ summary: Get Backfill
+ operationId: get_backfill
parameters:
- - name: dag_id
- in: path
- required: true
- schema:
- type: string
- title: Dag Id
- - name: dag_run_id
+ - name: backfill_id
in: path
required: true
schema:
type: string
- title: Dag Run Id
+ title: Backfill Id
responses:
'200':
description: Successful Response
content:
application/json:
- schema:
- $ref: '#/components/schemas/DAGRunResponse'
+ schema: {}
'401':
content:
application/json:
@@ -365,34 +318,24 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
- delete:
+ /public/backfills/{backfill_id}/pause:
+ put:
tags:
- - DagRun
- summary: Delete Dag Run
- description: Delete a DAG Run entry.
- operationId: delete_dag_run
+ - Backfill
+ summary: Pause Backfill
+ operationId: pause_backfill
parameters:
- - name: dag_id
- in: path
- required: true
- schema:
- type: string
- title: Dag Id
- - name: dag_run_id
+ - name: backfill_id
in: path
required: true
schema:
- type: string
- title: Dag Run Id
+ title: Backfill Id
responses:
- '204':
+ '200':
description: Successful Response
- '400':
content:
application/json:
- schema:
- $ref: '#/components/schemas/HTTPExceptionResponse'
- description: Bad Request
+ schema: {}
'401':
content:
application/json:
@@ -411,60 +354,36 @@ paths:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Not Found
+ '409':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Conflict
'422':
description: Validation Error
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
- patch:
+ /public/backfills/{backfill_id}/unpause:
+ put:
tags:
- - DagRun
- summary: Patch Dag Run State
- description: Modify a DAG Run.
- operationId: patch_dag_run_state
+ - Backfill
+ summary: Unpause Backfill
+ operationId: unpause_backfill
parameters:
- - name: dag_id
- in: path
- required: true
- schema:
- type: string
- title: Dag Id
- - name: dag_run_id
+ - name: backfill_id
in: path
required: true
schema:
- type: string
- title: Dag Run Id
- - name: update_mask
- in: query
- required: false
- schema:
- anyOf:
- - type: array
- items:
- type: string
- - type: 'null'
- title: Update Mask
- requestBody:
- required: true
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/DAGRunPatchBody'
+ title: Backfill Id
responses:
'200':
description: Successful Response
content:
application/json:
- schema:
- $ref: '#/components/schemas/DAGRunResponse'
- '400':
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/HTTPExceptionResponse'
- description: Bad Request
+ schema: {}
'401':
content:
application/json:
@@ -483,50 +402,36 @@ paths:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Not Found
+ '409':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Conflict
'422':
description: Validation Error
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
- /public/dagSources/{file_token}:
- get:
+ /public/backfills/{backfill_id}/cancel:
+ put:
tags:
- - DagSource
- summary: Get Dag Source
- description: Get source code using file token.
- operationId: get_dag_source
+ - Backfill
+ summary: Cancel Backfill
+ operationId: cancel_backfill
parameters:
- - name: file_token
+ - name: backfill_id
in: path
required: true
schema:
- type: string
- title: File Token
- - name: accept
- in: header
- required: false
- schema:
- type: string
- default: '*/*'
- title: Accept
+ title: Backfill Id
responses:
'200':
description: Successful Response
content:
application/json:
- schema:
- $ref: '#/components/schemas/DAGSourceResponse'
- text/plain:
- schema:
- type: string
- example: dag code
- '400':
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/HTTPExceptionResponse'
- description: Bad Request
+ schema: {}
'401':
content:
application/json:
@@ -545,12 +450,12 @@ paths:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Not Found
- '406':
+ '409':
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
- description: Not Acceptable
+ description: Conflict
'422':
description: Validation Error
content:
@@ -734,23 +639,387 @@ paths:
title: Last Dag Run State
requestBody:
required: true
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/DAGPatchBody'
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/DAGPatchBody'
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/DAGCollectionResponse'
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Bad Request
+ '401':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unauthorized
+ '403':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Forbidden
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
+ '422':
+ description: Validation Error
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPValidationError'
+ /public/dags/tags:
+ get:
+ tags:
+ - DAG
+ summary: Get Dag Tags
+ description: Get all DAG tags.
+ operationId: get_dag_tags
+ parameters:
+ - name: limit
+ in: query
+ required: false
+ schema:
+ type: integer
+ default: 100
+ title: Limit
+ - name: offset
+ in: query
+ required: false
+ schema:
+ type: integer
+ default: 0
+ title: Offset
+ - name: order_by
+ in: query
+ required: false
+ schema:
+ type: string
+ default: name
+ title: Order By
+ - name: tag_name_pattern
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Tag Name Pattern
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/DAGTagCollectionResponse'
+ '401':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unauthorized
+ '403':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Forbidden
+ '422':
+ description: Validation Error
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPValidationError'
+ /public/dags/{dag_id}:
+ get:
+ tags:
+ - DAG
+ summary: Get Dag
+ description: Get basic information about a DAG.
+ operationId: get_dag
+ parameters:
+ - name: dag_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Dag Id
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/DAGResponse'
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Bad Request
+ '401':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unauthorized
+ '403':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Forbidden
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
+ '422':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unprocessable Entity
+ patch:
+ tags:
+ - DAG
+ summary: Patch Dag
+ description: Patch the specific DAG.
+ operationId: patch_dag
+ parameters:
+ - name: dag_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Dag Id
+ - name: update_mask
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: array
+ items:
+ type: string
+ - type: 'null'
+ title: Update Mask
+ requestBody:
+ required: true
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/DAGPatchBody'
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/DAGResponse'
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Bad Request
+ '401':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unauthorized
+ '403':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Forbidden
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
+ '422':
+ description: Validation Error
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPValidationError'
+ delete:
+ tags:
+ - DAG
+ summary: Delete Dag
+ description: Delete the specific DAG.
+ operationId: delete_dag
+ parameters:
+ - name: dag_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Dag Id
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema: {}
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Bad Request
+ '401':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unauthorized
+ '403':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Forbidden
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
+ '422':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unprocessable Entity
+ /public/dags/{dag_id}/details:
+ get:
+ tags:
+ - DAG
+ summary: Get Dag Details
+ description: Get details of DAG.
+ operationId: get_dag_details
+ parameters:
+ - name: dag_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Dag Id
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/DAGDetailsResponse'
+ '400':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Bad Request
+ '401':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unauthorized
+ '403':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Forbidden
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
+ '422':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unprocessable Entity
+ /public/connections/{connection_id}:
+ delete:
+ tags:
+ - Connection
+ summary: Delete Connection
+ description: Delete a connection entry.
+ operationId: delete_connection
+ parameters:
+ - name: connection_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Connection Id
responses:
- '200':
+ '204':
description: Successful Response
+ '401':
content:
application/json:
schema:
- $ref: '#/components/schemas/DAGCollectionResponse'
- '400':
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Unauthorized
+ '403':
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
- description: Bad Request
+ description: Forbidden
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
+ '422':
+ description: Validation Error
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPValidationError'
+ get:
+ tags:
+ - Connection
+ summary: Get Connection
+ description: Get a connection entry.
+ operationId: get_connection
+ parameters:
+ - name: connection_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Connection Id
+ responses:
+ '200':
+ description: Successful Response
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/ConnectionResponse'
'401':
content:
application/json:
@@ -775,13 +1044,13 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
- /public/dags/tags:
+ /public/connections/:
get:
tags:
- - DAG
- summary: Get Dag Tags
- description: Get all DAG tags.
- operationId: get_dag_tags
+ - Connection
+ summary: Get Connections
+ description: Get all connection entries.
+ operationId: get_connections
parameters:
- name: limit
in: query
@@ -802,23 +1071,15 @@ paths:
required: false
schema:
type: string
- default: name
+ default: id
title: Order By
- - name: tag_name_pattern
- in: query
- required: false
- schema:
- anyOf:
- - type: string
- - type: 'null'
- title: Tag Name Pattern
responses:
'200':
description: Successful Response
content:
application/json:
schema:
- $ref: '#/components/schemas/DAGTagCollectionResponse'
+ $ref: '#/components/schemas/ConnectionCollectionResponse'
'401':
content:
application/json:
@@ -831,19 +1092,24 @@ paths:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Forbidden
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
'422':
description: Validation Error
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
- /public/dags/{dag_id}:
+ /public/dags/{dag_id}/dagRuns/{dag_run_id}:
get:
tags:
- - DAG
- summary: Get Dag
- description: Get basic information about a DAG.
- operationId: get_dag
+ - DagRun
+ summary: Get Dag Run
+ operationId: get_dag_run
parameters:
- name: dag_id
in: path
@@ -851,19 +1117,19 @@ paths:
schema:
type: string
title: Dag Id
+ - name: dag_run_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Dag Run Id
responses:
'200':
description: Successful Response
content:
application/json:
schema:
- $ref: '#/components/schemas/DAGResponse'
- '400':
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/HTTPExceptionResponse'
- description: Bad Request
+ $ref: '#/components/schemas/DAGRunResponse'
'401':
content:
application/json:
@@ -883,17 +1149,17 @@ paths:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Not Found
'422':
+ description: Validation Error
content:
application/json:
schema:
- $ref: '#/components/schemas/HTTPExceptionResponse'
- description: Unprocessable Entity
- patch:
+ $ref: '#/components/schemas/HTTPValidationError'
+ delete:
tags:
- - DAG
- summary: Patch Dag
- description: Patch the specific DAG.
- operationId: patch_dag
+ - DagRun
+ summary: Delete Dag Run
+ description: Delete a DAG Run entry.
+ operationId: delete_dag_run
parameters:
- name: dag_id
in: path
@@ -901,29 +1167,15 @@ paths:
schema:
type: string
title: Dag Id
- - name: update_mask
- in: query
- required: false
- schema:
- anyOf:
- - type: array
- items:
- type: string
- - type: 'null'
- title: Update Mask
- requestBody:
+ - name: dag_run_id
+ in: path
required: true
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/DAGPatchBody'
+ schema:
+ type: string
+ title: Dag Run Id
responses:
- '200':
+ '204':
description: Successful Response
- content:
- application/json:
- schema:
- $ref: '#/components/schemas/DAGResponse'
'400':
content:
application/json:
@@ -954,12 +1206,12 @@ paths:
application/json:
schema:
$ref: '#/components/schemas/HTTPValidationError'
- delete:
+ patch:
tags:
- - DAG
- summary: Delete Dag
- description: Delete the specific DAG.
- operationId: delete_dag
+ - DagRun
+ summary: Patch Dag Run State
+ description: Modify a DAG Run.
+ operationId: patch_dag_run_state
parameters:
- name: dag_id
in: path
@@ -967,12 +1219,35 @@ paths:
schema:
type: string
title: Dag Id
+ - name: dag_run_id
+ in: path
+ required: true
+ schema:
+ type: string
+ title: Dag Run Id
+ - name: update_mask
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: array
+ items:
+ type: string
+ - type: 'null'
+ title: Update Mask
+ requestBody:
+ required: true
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/DAGRunPatchBody'
responses:
'200':
description: Successful Response
content:
application/json:
- schema: {}
+ schema:
+ $ref: '#/components/schemas/DAGRunResponse'
'400':
content:
application/json:
@@ -998,32 +1273,43 @@ paths:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Not Found
'422':
+ description: Validation Error
content:
application/json:
schema:
- $ref: '#/components/schemas/HTTPExceptionResponse'
- description: Unprocessable Entity
- /public/dags/{dag_id}/details:
+ $ref: '#/components/schemas/HTTPValidationError'
+ /public/dagSources/{file_token}:
get:
tags:
- - DAG
- summary: Get Dag Details
- description: Get details of DAG.
- operationId: get_dag_details
+ - DagSource
+ summary: Get Dag Source
+ description: Get source code using file token.
+ operationId: get_dag_source
parameters:
- - name: dag_id
+ - name: file_token
in: path
required: true
schema:
type: string
- title: Dag Id
+ title: File Token
+ - name: accept
+ in: header
+ required: false
+ schema:
+ type: string
+ default: '*/*'
+ title: Accept
responses:
'200':
description: Successful Response
content:
application/json:
schema:
- $ref: '#/components/schemas/DAGDetailsResponse'
+ $ref: '#/components/schemas/DAGSourceResponse'
+ text/plain:
+ schema:
+ type: string
+ example: dag code
'400':
content:
application/json:
@@ -1048,12 +1334,18 @@ paths:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Not Found
- '422':
+ '406':
content:
application/json:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
- description: Unprocessable Entity
+ description: Not Acceptable
+ '422':
+ description: Validation Error
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPValidationError'
/public/eventLogs/{event_log_id}:
get:
tags:
@@ -1929,6 +2221,41 @@ components:
type: object
title: AppBuilderViewResponse
description: Serializer for AppBuilder View responses.
+ BackfillPostBody:
+ properties:
+ dag_id:
+ type: string
+ title: Dag Id
+ from_date:
+ type: string
+ format: date-time
+ title: From Date
+ to_date:
+ type: string
+ format: date-time
+ title: To Date
+ run_backwards:
+ type: boolean
+ title: Run Backwards
+ default: false
+ dag_run_conf:
+ type: object
+ title: Dag Run Conf
+ default: {}
+ reprocess_behavior:
+ $ref: '#/components/schemas/ReprocessBehavior'
+ default: none
+ max_active_runs:
+ type: integer
+ title: Max Active Runs
+ default: 10
+ type: object
+ required:
+ - dag_id
+ - from_date
+ - to_date
+ title: BackfillPostBody
+ description: Object used for create backfill request.
BaseInfoSchema:
properties:
status:
@@ -3309,6 +3636,17 @@ components:
- version
title: ProviderResponse
description: Provider serializer for responses.
+ ReprocessBehavior:
+ type: string
+ enum:
+ - failed
+ - completed
+ - none
+ title: ReprocessBehavior
+ description: 'Internal enum for setting reprocess behavior in a backfill.
+
+
+ :meta private:'
SchedulerInfoSchema:
properties:
status:
diff --git a/airflow/api_fastapi/core_api/routes/public/__init__.py b/airflow/api_fastapi/core_api/routes/public/__init__.py
index a153952287b..a443f5a28ae 100644
--- a/airflow/api_fastapi/core_api/routes/public/__init__.py
+++ b/airflow/api_fastapi/core_api/routes/public/__init__.py
@@ -18,6 +18,7 @@
from __future__ import annotations
from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.routes.public.backfills import backfills_router
from airflow.api_fastapi.core_api.routes.public.connections import connections_router
from airflow.api_fastapi.core_api.routes.public.dag_run import dag_run_router
from airflow.api_fastapi.core_api.routes.public.dag_sources import dag_sources_router
@@ -35,6 +36,8 @@ from airflow.api_fastapi.core_api.routes.public.version import version_router
public_router = AirflowRouter(prefix="/public")
+public_router.include_router(backfills_router)
+public_router.include_router(dags_router)
public_router.include_router(connections_router)
public_router.include_router(dag_run_router)
public_router.include_router(dag_sources_router)
diff --git a/airflow/api_fastapi/core_api/routes/public/backfills.py b/airflow/api_fastapi/core_api/routes/public/backfills.py
new file mode 100644
index 00000000000..f6fe531d006
--- /dev/null
+++ b/airflow/api_fastapi/core_api/routes/public/backfills.py
@@ -0,0 +1,187 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from fastapi import Depends, HTTPException
+from sqlalchemy import select, update
+from sqlalchemy.orm import Session
+from typing_extensions import Annotated
+
+from airflow.api_fastapi.common.db.common import get_session, paginated_select
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.api_fastapi.core_api.serializers.backfills import (
+ BackfillCollectionResponse,
+ BackfillPostBody,
+ BackfillResponse,
+)
+from airflow.models import DagRun
+from airflow.models.backfill import (
+ AlreadyRunningBackfill,
+ Backfill,
+ BackfillDagRun,
+ _create_backfill,
+)
+from airflow.utils import timezone
+from airflow.utils.state import DagRunState
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="/",
+ responses=create_openapi_http_exception_doc([401, 403]),
+)
+async def list_backfills(
+ dag_id: str,
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[Session, Depends(get_session)],
+):
+ select_stmt, total_entries = paginated_select(
+ select(Backfill).where(Backfill.dag_id == dag_id),
+ [],
+ order_by=order_by,
+ offset=offset,
+ limit=limit,
+ session=session,
+ )
+ backfills = session.scalars(select_stmt).all()
+
+ return BackfillCollectionResponse(
+ backfills=[BackfillResponse.model_validate(x, from_attributes=True) for x in backfills],
+ total_entries=len(backfills),
+ )
+
+
+@backfills_router.get(
+ path="/{backfill_id}",
+ responses=create_openapi_http_exception_doc([401, 403, 404]),
+)
+async def get_backfill(
+ backfill_id: str,
+ session: Annotated[Session, Depends(get_session)],
+):
+ backfill = session.get(Backfill, backfill_id)
+ if backfill:
+ return BackfillResponse.model_validate(backfill, from_attributes=True)
+ raise HTTPException(404, "Backfill not found")
+
+
+@backfills_router.put(
+ path="/{backfill_id}/pause",
+ responses=create_openapi_http_exception_doc([401, 403, 404, 409]),
+)
+async def pause_backfill(*, backfill_id, session: Annotated[Session, Depends(get_session)]):
+ b = session.get(Backfill, backfill_id)
+ if not b:
+ raise HTTPException(404, f"Could not find backfill with id {backfill_id}")
+ if b.completed_at:
+ raise HTTPException(409, "Backfill is already completed.")
+ if b.is_paused is False:
+ b.is_paused = True
+ session.commit()
+ return BackfillResponse.model_validate(b, from_attributes=True)
+
+
+@backfills_router.put(
+ path="/{backfill_id}/unpause",
+ responses=create_openapi_http_exception_doc([401, 403, 404, 409]),
+)
+async def unpause_backfill(*, backfill_id, session: Annotated[Session, Depends(get_session)]):
+ b = session.get(Backfill, backfill_id)
+ if not b:
+ raise HTTPException(404, f"Could not find backfill with id {backfill_id}")
+ if b.completed_at:
+ raise HTTPException(409, "Backfill is already completed.")
+ if b.is_paused:
+ b.is_paused = False
+ return BackfillResponse.model_validate(b, from_attributes=True)
+
+
+@backfills_router.put(
+ path="/{backfill_id}/cancel",
+ responses=create_openapi_http_exception_doc([401, 403, 404, 409]),
+)
+async def cancel_backfill(*, backfill_id, session: Annotated[Session, Depends(get_session)]):
+ b: Backfill = session.get(Backfill, backfill_id)
+ if not b:
+ raise HTTPException(404, f"Could not find backfill with id {backfill_id}")
+ if b.completed_at is not None:
+ raise HTTPException(409, "Backfill is already completed.")
+
+ # first, pause, and commit immediately to ensure no other dag runs are started
+ if not b.is_paused:
+ b.is_paused = True
+ session.commit() # ensure no new runs started
+
+ query = (
+ update(DagRun)
+ .where(
+ DagRun.id.in_(
+ select(
+ BackfillDagRun.dag_run_id,
+ ).where(
+ BackfillDagRun.backfill_id == b.id,
+ ),
+ ),
+ DagRun.state == DagRunState.QUEUED,
+ )
+ .values(state=DagRunState.FAILED)
+ .execution_options(synchronize_session=False)
+ )
+ session.execute(query)
+ session.commit() # this will fail all the queued dag runs in this backfill
+
+ # this is in separate transaction just to avoid potential conflicts
+ session.refresh(b)
+ b.completed_at = timezone.utcnow()
+ return BackfillResponse.model_validate(b, from_attributes=True)
+
+
+@backfills_router.post(
+ path="/",
+ responses=create_openapi_http_exception_doc([401, 403, 404, 409]),
+)
+async def create_backfill(
+ backfill_request: BackfillPostBody,
+):
+ from_date = timezone.coerce_datetime(backfill_request.from_date)
+ to_date = timezone.coerce_datetime(backfill_request.to_date)
+ try:
+ backfill_obj = _create_backfill(
+ dag_id=backfill_request.dag_id,
+ from_date=from_date,
+ to_date=to_date,
+ max_active_runs=backfill_request.max_active_runs,
+ reverse=backfill_request.run_backwards,
+ dag_run_conf=backfill_request.dag_run_conf,
+ reprocess_behavior=backfill_request.reprocess_behavior,
+ )
+ return BackfillResponse.model_validate(backfill_obj, from_attributes=True)
+ except AlreadyRunningBackfill:
+ raise HTTPException(
+ status_code=409,
+ detail=f"There is already a running backfill for dag {backfill_request.dag_id}",
+ )
diff --git a/airflow/api_fastapi/core_api/serializers/backfills.py b/airflow/api_fastapi/core_api/serializers/backfills.py
new file mode 100644
index 00000000000..69d6a98ccfd
--- /dev/null
+++ b/airflow/api_fastapi/core_api/serializers/backfills.py
@@ -0,0 +1,59 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from airflow.models.backfill import ReprocessBehavior
+
+
+class BackfillPostBody(BaseModel):
+ """Object used for create backfill request."""
+
+ dag_id: str
+ from_date: datetime
+ to_date: datetime
+ run_backwards: bool = False
+ dag_run_conf: dict = {}
+ reprocess_behavior: ReprocessBehavior = ReprocessBehavior.NONE
+ max_active_runs: int = 10
+
+
+class BackfillResponse(BaseModel):
+ """Base serializer for Backfill."""
+
+ id: int
+ dag_id: str
+ from_date: datetime
+ to_date: datetime
+ dag_run_conf: dict
+ is_paused: bool
+ reprocess_behavior: ReprocessBehavior
+ max_active_runs: int
+ created_at: datetime
+ completed_at: datetime | None
+ updated_at: datetime
+
+
+class BackfillCollectionResponse(BaseModel):
+ """Backfill Collection serializer for responses."""
+
+ backfills: list[BackfillResponse]
+ total_entries: int
diff --git a/airflow/models/backfill.py b/airflow/models/backfill.py
index 72df6a1e0eb..648b35c5bde 100644
--- a/airflow/models/backfill.py
+++ b/airflow/models/backfill.py
@@ -36,12 +36,11 @@ from sqlalchemy import (
desc,
func,
select,
- update,
)
from sqlalchemy.orm import relationship, validates
from sqlalchemy_jsonfield import JSONField
-from airflow.api_connexion.exceptions import Conflict, NotFound
+from airflow.api_connexion.exceptions import NotFound
from airflow.exceptions import AirflowException
from airflow.models.base import Base, StringID
from airflow.settings import json
@@ -87,7 +86,7 @@ class Backfill(Base):
dag_id = Column(StringID(), nullable=False)
from_date = Column(UtcDateTime, nullable=False)
to_date = Column(UtcDateTime, nullable=False)
- dag_run_conf = Column(JSONField(json=json), nullable=True)
+ dag_run_conf = Column(JSONField(json=json), nullable=False, default={})
is_paused = Column(Boolean, default=False)
"""
Controls whether new dag runs will be created for this backfill.
@@ -333,33 +332,3 @@ def _create_backfill(
info,
)
return br
-
-
-def _cancel_backfill(backfill_id) -> Backfill:
- with create_session() as session:
- b: Backfill = session.get(Backfill, backfill_id)
- if b.completed_at is not None:
- raise Conflict("Backfill is already completed.")
-
- b.completed_at = timezone.utcnow()
-
- # first, pause
- if not b.is_paused:
- b.is_paused = True
-
- session.commit()
-
- from airflow.models import DagRun
-
- # now, let's mark all queued dag runs as failed
- query = (
- update(DagRun)
- .where(
- DagRun.id.in_(select(BackfillDagRun.dag_run_id).where(BackfillDagRun.backfill_id == b.id)),
- DagRun.state == DagRunState.QUEUED,
- )
- .values(state=DagRunState.FAILED)
- .execution_options(synchronize_session=False)
- )
- session.execute(query)
- return b
diff --git a/airflow/ui/openapi-gen/queries/common.ts b/airflow/ui/openapi-gen/queries/common.ts
index f715b41c3f4..75a343ce74c 100644
--- a/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow/ui/openapi-gen/queries/common.ts
@@ -3,6 +3,7 @@ import { UseQueryResult } from "@tanstack/react-query";
import {
AssetService,
+ BackfillService,
ConnectionService,
DagRunService,
DagService,
@@ -109,90 +110,48 @@ export const UseDagsServiceRecentDagRunsKeyFn = (
},
]),
];
-export type ConnectionServiceGetConnectionDefaultResponse = Awaited<
- ReturnType<typeof ConnectionService.getConnection>
+export type BackfillServiceListBackfillsDefaultResponse = Awaited<
+ ReturnType<typeof BackfillService.listBackfills>
>;
-export type ConnectionServiceGetConnectionQueryResult<
- TData = ConnectionServiceGetConnectionDefaultResponse,
+export type BackfillServiceListBackfillsQueryResult<
+ TData = BackfillServiceListBackfillsDefaultResponse,
TError = unknown,
> = UseQueryResult<TData, TError>;
-export const useConnectionServiceGetConnectionKey =
- "ConnectionServiceGetConnection";
-export const UseConnectionServiceGetConnectionKeyFn = (
- {
- connectionId,
- }: {
- connectionId: string;
- },
- queryKey?: Array<unknown>,
-) => [
- useConnectionServiceGetConnectionKey,
- ...(queryKey ?? [{ connectionId }]),
-];
-export type ConnectionServiceGetConnectionsDefaultResponse = Awaited<
- ReturnType<typeof ConnectionService.getConnections>
->;
-export type ConnectionServiceGetConnectionsQueryResult<
- TData = ConnectionServiceGetConnectionsDefaultResponse,
- TError = unknown,
-> = UseQueryResult<TData, TError>;
-export const useConnectionServiceGetConnectionsKey =
- "ConnectionServiceGetConnections";
-export const UseConnectionServiceGetConnectionsKeyFn = (
+export const useBackfillServiceListBackfillsKey =
+ "BackfillServiceListBackfills";
+export const UseBackfillServiceListBackfillsKeyFn = (
{
+ dagId,
limit,
offset,
orderBy,
}: {
+ dagId: string;
limit?: number;
offset?: number;
orderBy?: string;
- } = {},
+ },
queryKey?: Array<unknown>,
) => [
- useConnectionServiceGetConnectionsKey,
- ...(queryKey ?? [{ limit, offset, orderBy }]),
+ useBackfillServiceListBackfillsKey,
+ ...(queryKey ?? [{ dagId, limit, offset, orderBy }]),
];
-export type DagRunServiceGetDagRunDefaultResponse = Awaited<
- ReturnType<typeof DagRunService.getDagRun>
->;
-export type DagRunServiceGetDagRunQueryResult<
- TData = DagRunServiceGetDagRunDefaultResponse,
- TError = unknown,
-> = UseQueryResult<TData, TError>;
-export const useDagRunServiceGetDagRunKey = "DagRunServiceGetDagRun";
-export const UseDagRunServiceGetDagRunKeyFn = (
- {
- dagId,
- dagRunId,
- }: {
- dagId: string;
- dagRunId: string;
- },
- queryKey?: Array<unknown>,
-) => [useDagRunServiceGetDagRunKey, ...(queryKey ?? [{ dagId, dagRunId }])];
-export type DagSourceServiceGetDagSourceDefaultResponse = Awaited<
- ReturnType<typeof DagSourceService.getDagSource>
+export type BackfillServiceGetBackfillDefaultResponse = Awaited<
+ ReturnType<typeof BackfillService.getBackfill>
>;
-export type DagSourceServiceGetDagSourceQueryResult<
- TData = DagSourceServiceGetDagSourceDefaultResponse,
+export type BackfillServiceGetBackfillQueryResult<
+ TData = BackfillServiceGetBackfillDefaultResponse,
TError = unknown,
> = UseQueryResult<TData, TError>;
-export const useDagSourceServiceGetDagSourceKey =
- "DagSourceServiceGetDagSource";
-export const UseDagSourceServiceGetDagSourceKeyFn = (
+export const useBackfillServiceGetBackfillKey = "BackfillServiceGetBackfill";
+export const UseBackfillServiceGetBackfillKeyFn = (
{
- accept,
- fileToken,
+ backfillId,
}: {
- accept?: string;
- fileToken: string;
+ backfillId: string;
},
queryKey?: Array<unknown>,
-) => [
- useDagSourceServiceGetDagSourceKey,
- ...(queryKey ?? [{ accept, fileToken }]),
-];
+) => [useBackfillServiceGetBackfillKey, ...(queryKey ?? [{ backfillId }])];
export type DagServiceGetDagsDefaultResponse = Awaited<
ReturnType<typeof DagService.getDags>
>;
@@ -300,6 +259,90 @@ export const UseDagServiceGetDagDetailsKeyFn = (
},
queryKey?: Array<unknown>,
) => [useDagServiceGetDagDetailsKey, ...(queryKey ?? [{ dagId }])];
+export type ConnectionServiceGetConnectionDefaultResponse = Awaited<
+ ReturnType<typeof ConnectionService.getConnection>
+>;
+export type ConnectionServiceGetConnectionQueryResult<
+ TData = ConnectionServiceGetConnectionDefaultResponse,
+ TError = unknown,
+> = UseQueryResult<TData, TError>;
+export const useConnectionServiceGetConnectionKey =
+ "ConnectionServiceGetConnection";
+export const UseConnectionServiceGetConnectionKeyFn = (
+ {
+ connectionId,
+ }: {
+ connectionId: string;
+ },
+ queryKey?: Array<unknown>,
+) => [
+ useConnectionServiceGetConnectionKey,
+ ...(queryKey ?? [{ connectionId }]),
+];
+export type ConnectionServiceGetConnectionsDefaultResponse = Awaited<
+ ReturnType<typeof ConnectionService.getConnections>
+>;
+export type ConnectionServiceGetConnectionsQueryResult<
+ TData = ConnectionServiceGetConnectionsDefaultResponse,
+ TError = unknown,
+> = UseQueryResult<TData, TError>;
+export const useConnectionServiceGetConnectionsKey =
+ "ConnectionServiceGetConnections";
+export const UseConnectionServiceGetConnectionsKeyFn = (
+ {
+ limit,
+ offset,
+ orderBy,
+ }: {
+ limit?: number;
+ offset?: number;
+ orderBy?: string;
+ } = {},
+ queryKey?: Array<unknown>,
+) => [
+ useConnectionServiceGetConnectionsKey,
+ ...(queryKey ?? [{ limit, offset, orderBy }]),
+];
+export type DagRunServiceGetDagRunDefaultResponse = Awaited<
+ ReturnType<typeof DagRunService.getDagRun>
+>;
+export type DagRunServiceGetDagRunQueryResult<
+ TData = DagRunServiceGetDagRunDefaultResponse,
+ TError = unknown,
+> = UseQueryResult<TData, TError>;
+export const useDagRunServiceGetDagRunKey = "DagRunServiceGetDagRun";
+export const UseDagRunServiceGetDagRunKeyFn = (
+ {
+ dagId,
+ dagRunId,
+ }: {
+ dagId: string;
+ dagRunId: string;
+ },
+ queryKey?: Array<unknown>,
+) => [useDagRunServiceGetDagRunKey, ...(queryKey ?? [{ dagId, dagRunId }])];
+export type DagSourceServiceGetDagSourceDefaultResponse = Awaited<
+ ReturnType<typeof DagSourceService.getDagSource>
+>;
+export type DagSourceServiceGetDagSourceQueryResult<
+ TData = DagSourceServiceGetDagSourceDefaultResponse,
+ TError = unknown,
+> = UseQueryResult<TData, TError>;
+export const useDagSourceServiceGetDagSourceKey =
+ "DagSourceServiceGetDagSource";
+export const UseDagSourceServiceGetDagSourceKeyFn = (
+ {
+ accept,
+ fileToken,
+ }: {
+ accept?: string;
+ fileToken: string;
+ },
+ queryKey?: Array<unknown>,
+) => [
+ useDagSourceServiceGetDagSourceKey,
+ ...(queryKey ?? [{ accept, fileToken }]),
+];
export type EventLogServiceGetEventLogDefaultResponse = Awaited<
ReturnType<typeof EventLogService.getEventLog>
>;
@@ -529,14 +572,23 @@ export const UseVersionServiceGetVersionKeyFn = (queryKey?: Array<unknown>) => [
useVersionServiceGetVersionKey,
...(queryKey ?? []),
];
+export type BackfillServiceCreateBackfillMutationResult = Awaited<
+ ReturnType<typeof BackfillService.createBackfill>
+>;
export type PoolServicePostPoolMutationResult = Awaited<
ReturnType<typeof PoolService.postPool>
>;
export type VariableServicePostVariableMutationResult = Awaited<
ReturnType<typeof VariableService.postVariable>
>;
-export type DagRunServicePatchDagRunStateMutationResult = Awaited<
- ReturnType<typeof DagRunService.patchDagRunState>
+export type BackfillServicePauseBackfillMutationResult = Awaited<
+ ReturnType<typeof BackfillService.pauseBackfill>
+>;
+export type BackfillServiceUnpauseBackfillMutationResult = Awaited<
+ ReturnType<typeof BackfillService.unpauseBackfill>
+>;
+export type BackfillServiceCancelBackfillMutationResult = Awaited<
+ ReturnType<typeof BackfillService.cancelBackfill>
>;
export type DagServicePatchDagsMutationResult = Awaited<
ReturnType<typeof DagService.patchDags>
@@ -544,21 +596,24 @@ export type DagServicePatchDagsMutationResult = Awaited<
export type DagServicePatchDagMutationResult = Awaited<
ReturnType<typeof DagService.patchDag>
>;
+export type DagRunServicePatchDagRunStateMutationResult = Awaited<
+ ReturnType<typeof DagRunService.patchDagRunState>
+>;
export type PoolServicePatchPoolMutationResult = Awaited<
ReturnType<typeof PoolService.patchPool>
>;
export type VariableServicePatchVariableMutationResult = Awaited<
ReturnType<typeof VariableService.patchVariable>
>;
+export type DagServiceDeleteDagMutationResult = Awaited<
+ ReturnType<typeof DagService.deleteDag>
+>;
export type ConnectionServiceDeleteConnectionMutationResult = Awaited<
ReturnType<typeof ConnectionService.deleteConnection>
>;
export type DagRunServiceDeleteDagRunMutationResult = Awaited<
ReturnType<typeof DagRunService.deleteDagRun>
>;
-export type DagServiceDeleteDagMutationResult = Awaited<
- ReturnType<typeof DagService.deleteDag>
->;
export type PoolServiceDeletePoolMutationResult = Awaited<
ReturnType<typeof PoolService.deletePool>
>;
diff --git a/airflow/ui/openapi-gen/queries/prefetch.ts b/airflow/ui/openapi-gen/queries/prefetch.ts
index a0f3a75eb22..63e8d4b4313 100644
--- a/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -3,6 +3,7 @@ import { type QueryClient } from "@tanstack/react-query";
import {
AssetService,
+ BackfillService,
ConnectionService,
DagRunService,
DagService,
@@ -138,102 +139,57 @@ export const prefetchUseDagsServiceRecentDagRuns = (
}),
});
/**
- * Get Connection
- * Get a connection entry.
- * @param data The data for the request.
- * @param data.connectionId
- * @returns ConnectionResponse Successful Response
- * @throws ApiError
- */
-export const prefetchUseConnectionServiceGetConnection = (
- queryClient: QueryClient,
- {
- connectionId,
- }: {
- connectionId: string;
- },
-) =>
- queryClient.prefetchQuery({
- queryKey: Common.UseConnectionServiceGetConnectionKeyFn({ connectionId }),
- queryFn: () => ConnectionService.getConnection({ connectionId }),
- });
-/**
- * Get Connections
- * Get all connection entries.
+ * List Backfills
* @param data The data for the request.
+ * @param data.dagId
* @param data.limit
* @param data.offset
* @param data.orderBy
- * @returns ConnectionCollectionResponse Successful Response
+ * @returns unknown Successful Response
* @throws ApiError
*/
-export const prefetchUseConnectionServiceGetConnections = (
+export const prefetchUseBackfillServiceListBackfills = (
queryClient: QueryClient,
{
+ dagId,
limit,
offset,
orderBy,
}: {
+ dagId: string;
limit?: number;
offset?: number;
orderBy?: string;
- } = {},
+ },
) =>
queryClient.prefetchQuery({
- queryKey: Common.UseConnectionServiceGetConnectionsKeyFn({
+ queryKey: Common.UseBackfillServiceListBackfillsKeyFn({
+ dagId,
limit,
offset,
orderBy,
}),
- queryFn: () => ConnectionService.getConnections({ limit, offset, orderBy }),
- });
-/**
- * Get Dag Run
- * @param data The data for the request.
- * @param data.dagId
- * @param data.dagRunId
- * @returns DAGRunResponse Successful Response
- * @throws ApiError
- */
-export const prefetchUseDagRunServiceGetDagRun = (
- queryClient: QueryClient,
- {
- dagId,
- dagRunId,
- }: {
- dagId: string;
- dagRunId: string;
- },
-) =>
- queryClient.prefetchQuery({
- queryKey: Common.UseDagRunServiceGetDagRunKeyFn({ dagId, dagRunId }),
- queryFn: () => DagRunService.getDagRun({ dagId, dagRunId }),
+ queryFn: () =>
+ BackfillService.listBackfills({ dagId, limit, offset, orderBy }),
});
/**
- * Get Dag Source
- * Get source code using file token.
+ * Get Backfill
* @param data The data for the request.
- * @param data.fileToken
- * @param data.accept
- * @returns DAGSourceResponse Successful Response
+ * @param data.backfillId
+ * @returns unknown Successful Response
* @throws ApiError
*/
-export const prefetchUseDagSourceServiceGetDagSource = (
+export const prefetchUseBackfillServiceGetBackfill = (
queryClient: QueryClient,
{
- accept,
- fileToken,
+ backfillId,
}: {
- accept?: string;
- fileToken: string;
+ backfillId: string;
},
) =>
queryClient.prefetchQuery({
- queryKey: Common.UseDagSourceServiceGetDagSourceKeyFn({
- accept,
- fileToken,
- }),
- queryFn: () => DagSourceService.getDagSource({ accept, fileToken }),
+ queryKey: Common.UseBackfillServiceGetBackfillKeyFn({ backfillId }),
+ queryFn: () => BackfillService.getBackfill({ backfillId }),
});
/**
* Get Dags
@@ -380,6 +336,104 @@ export const prefetchUseDagServiceGetDagDetails = (
queryKey: Common.UseDagServiceGetDagDetailsKeyFn({ dagId }),
queryFn: () => DagService.getDagDetails({ dagId }),
});
+/**
+ * Get Connection
+ * Get a connection entry.
+ * @param data The data for the request.
+ * @param data.connectionId
+ * @returns ConnectionResponse Successful Response
+ * @throws ApiError
+ */
+export const prefetchUseConnectionServiceGetConnection = (
+ queryClient: QueryClient,
+ {
+ connectionId,
+ }: {
+ connectionId: string;
+ },
+) =>
+ queryClient.prefetchQuery({
+ queryKey: Common.UseConnectionServiceGetConnectionKeyFn({ connectionId }),
+ queryFn: () => ConnectionService.getConnection({ connectionId }),
+ });
+/**
+ * Get Connections
+ * Get all connection entries.
+ * @param data The data for the request.
+ * @param data.limit
+ * @param data.offset
+ * @param data.orderBy
+ * @returns ConnectionCollectionResponse Successful Response
+ * @throws ApiError
+ */
+export const prefetchUseConnectionServiceGetConnections = (
+ queryClient: QueryClient,
+ {
+ limit,
+ offset,
+ orderBy,
+ }: {
+ limit?: number;
+ offset?: number;
+ orderBy?: string;
+ } = {},
+) =>
+ queryClient.prefetchQuery({
+ queryKey: Common.UseConnectionServiceGetConnectionsKeyFn({
+ limit,
+ offset,
+ orderBy,
+ }),
+ queryFn: () => ConnectionService.getConnections({ limit, offset, orderBy }),
+ });
+/**
+ * Get Dag Run
+ * @param data The data for the request.
+ * @param data.dagId
+ * @param data.dagRunId
+ * @returns DAGRunResponse Successful Response
+ * @throws ApiError
+ */
+export const prefetchUseDagRunServiceGetDagRun = (
+ queryClient: QueryClient,
+ {
+ dagId,
+ dagRunId,
+ }: {
+ dagId: string;
+ dagRunId: string;
+ },
+) =>
+ queryClient.prefetchQuery({
+ queryKey: Common.UseDagRunServiceGetDagRunKeyFn({ dagId, dagRunId }),
+ queryFn: () => DagRunService.getDagRun({ dagId, dagRunId }),
+ });
+/**
+ * Get Dag Source
+ * Get source code using file token.
+ * @param data The data for the request.
+ * @param data.fileToken
+ * @param data.accept
+ * @returns DAGSourceResponse Successful Response
+ * @throws ApiError
+ */
+export const prefetchUseDagSourceServiceGetDagSource = (
+ queryClient: QueryClient,
+ {
+ accept,
+ fileToken,
+ }: {
+ accept?: string;
+ fileToken: string;
+ },
+) =>
+ queryClient.prefetchQuery({
+ queryKey: Common.UseDagSourceServiceGetDagSourceKeyFn({
+ accept,
+ fileToken,
+ }),
+ queryFn: () => DagSourceService.getDagSource({ accept, fileToken }),
+ });
/**
* Get Event Log
* @param data The data for the request.
diff --git a/airflow/ui/openapi-gen/queries/queries.ts b/airflow/ui/openapi-gen/queries/queries.ts
index 8ffcee3defb..5f8f649372d 100644
--- a/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow/ui/openapi-gen/queries/queries.ts
@@ -8,6 +8,7 @@ import {
import {
AssetService,
+ BackfillService,
ConnectionService,
DagRunService,
DagService,
@@ -25,6 +26,7 @@ import {
VersionService,
} from "../requests/services.gen";
import {
+ BackfillPostBody,
DAGPatchBody,
DAGRunPatchBody,
DagRunState,
@@ -173,132 +175,69 @@ export const useDagsServiceRecentDagRuns = <
...options,
});
/**
- * Get Connection
- * Get a connection entry.
- * @param data The data for the request.
- * @param data.connectionId
- * @returns ConnectionResponse Successful Response
- * @throws ApiError
- */
-export const useConnectionServiceGetConnection = <
- TData = Common.ConnectionServiceGetConnectionDefaultResponse,
- TError = unknown,
- TQueryKey extends Array<unknown> = unknown[],
->(
- {
- connectionId,
- }: {
- connectionId: string;
- },
- queryKey?: TQueryKey,
- options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
-) =>
- useQuery<TData, TError>({
- queryKey: Common.UseConnectionServiceGetConnectionKeyFn(
- { connectionId },
- queryKey,
- ),
- queryFn: () => ConnectionService.getConnection({ connectionId }) as TData,
- ...options,
- });
-/**
- * Get Connections
- * Get all connection entries.
+ * List Backfills
* @param data The data for the request.
+ * @param data.dagId
* @param data.limit
* @param data.offset
* @param data.orderBy
- * @returns ConnectionCollectionResponse Successful Response
+ * @returns unknown Successful Response
* @throws ApiError
*/
-export const useConnectionServiceGetConnections = <
- TData = Common.ConnectionServiceGetConnectionsDefaultResponse,
+export const useBackfillServiceListBackfills = <
+ TData = Common.BackfillServiceListBackfillsDefaultResponse,
TError = unknown,
TQueryKey extends Array<unknown> = unknown[],
>(
{
+ dagId,
limit,
offset,
orderBy,
}: {
+ dagId: string;
limit?: number;
offset?: number;
orderBy?: string;
- } = {},
- queryKey?: TQueryKey,
- options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
-) =>
- useQuery<TData, TError>({
- queryKey: Common.UseConnectionServiceGetConnectionsKeyFn(
- { limit, offset, orderBy },
- queryKey,
- ),
- queryFn: () =>
- ConnectionService.getConnections({ limit, offset, orderBy }) as TData,
- ...options,
- });
-/**
- * Get Dag Run
- * @param data The data for the request.
- * @param data.dagId
- * @param data.dagRunId
- * @returns DAGRunResponse Successful Response
- * @throws ApiError
- */
-export const useDagRunServiceGetDagRun = <
- TData = Common.DagRunServiceGetDagRunDefaultResponse,
- TError = unknown,
- TQueryKey extends Array<unknown> = unknown[],
->(
- {
- dagId,
- dagRunId,
- }: {
- dagId: string;
- dagRunId: string;
},
queryKey?: TQueryKey,
options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
) =>
useQuery<TData, TError>({
- queryKey: Common.UseDagRunServiceGetDagRunKeyFn(
- { dagId, dagRunId },
+ queryKey: Common.UseBackfillServiceListBackfillsKeyFn(
+ { dagId, limit, offset, orderBy },
queryKey,
),
- queryFn: () => DagRunService.getDagRun({ dagId, dagRunId }) as TData,
+ queryFn: () =>
+ BackfillService.listBackfills({ dagId, limit, offset, orderBy }) as TData,
...options,
});
/**
- * Get Dag Source
- * Get source code using file token.
+ * Get Backfill
* @param data The data for the request.
- * @param data.fileToken
- * @param data.accept
- * @returns DAGSourceResponse Successful Response
+ * @param data.backfillId
+ * @returns unknown Successful Response
* @throws ApiError
*/
-export const useDagSourceServiceGetDagSource = <
- TData = Common.DagSourceServiceGetDagSourceDefaultResponse,
+export const useBackfillServiceGetBackfill = <
+ TData = Common.BackfillServiceGetBackfillDefaultResponse,
TError = unknown,
TQueryKey extends Array<unknown> = unknown[],
>(
{
- accept,
- fileToken,
+ backfillId,
}: {
- accept?: string;
- fileToken: string;
+ backfillId: string;
},
queryKey?: TQueryKey,
options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
) =>
useQuery<TData, TError>({
- queryKey: Common.UseDagSourceServiceGetDagSourceKeyFn(
- { accept, fileToken },
+ queryKey: Common.UseBackfillServiceGetBackfillKeyFn(
+ { backfillId },
queryKey,
),
- queryFn: () =>
- DagSourceService.getDagSource({ accept, fileToken }) as TData,
+ queryFn: () => BackfillService.getBackfill({ backfillId }) as TData,
...options,
});
/**
@@ -476,6 +415,135 @@ export const useDagServiceGetDagDetails = <
queryFn: () => DagService.getDagDetails({ dagId }) as TData,
...options,
});
+/**
+ * Get Connection
+ * Get a connection entry.
+ * @param data The data for the request.
+ * @param data.connectionId
+ * @returns ConnectionResponse Successful Response
+ * @throws ApiError
+ */
+export const useConnectionServiceGetConnection = <
+ TData = Common.ConnectionServiceGetConnectionDefaultResponse,
+ TError = unknown,
+ TQueryKey extends Array<unknown> = unknown[],
+>(
+ {
+ connectionId,
+ }: {
+ connectionId: string;
+ },
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useQuery<TData, TError>({
+ queryKey: Common.UseConnectionServiceGetConnectionKeyFn(
+ { connectionId },
+ queryKey,
+ ),
+ queryFn: () => ConnectionService.getConnection({ connectionId }) as TData,
+ ...options,
+ });
+/**
+ * Get Connections
+ * Get all connection entries.
+ * @param data The data for the request.
+ * @param data.limit
+ * @param data.offset
+ * @param data.orderBy
+ * @returns ConnectionCollectionResponse Successful Response
+ * @throws ApiError
+ */
+export const useConnectionServiceGetConnections = <
+ TData = Common.ConnectionServiceGetConnectionsDefaultResponse,
+ TError = unknown,
+ TQueryKey extends Array<unknown> = unknown[],
+>(
+ {
+ limit,
+ offset,
+ orderBy,
+ }: {
+ limit?: number;
+ offset?: number;
+ orderBy?: string;
+ } = {},
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useQuery<TData, TError>({
+ queryKey: Common.UseConnectionServiceGetConnectionsKeyFn(
+ { limit, offset, orderBy },
+ queryKey,
+ ),
+ queryFn: () =>
+ ConnectionService.getConnections({ limit, offset, orderBy }) as TData,
+ ...options,
+ });
+/**
+ * Get Dag Run
+ * @param data The data for the request.
+ * @param data.dagId
+ * @param data.dagRunId
+ * @returns DAGRunResponse Successful Response
+ * @throws ApiError
+ */
+export const useDagRunServiceGetDagRun = <
+ TData = Common.DagRunServiceGetDagRunDefaultResponse,
+ TError = unknown,
+ TQueryKey extends Array<unknown> = unknown[],
+>(
+ {
+ dagId,
+ dagRunId,
+ }: {
+ dagId: string;
+ dagRunId: string;
+ },
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useQuery<TData, TError>({
+ queryKey: Common.UseDagRunServiceGetDagRunKeyFn(
+ { dagId, dagRunId },
+ queryKey,
+ ),
+ queryFn: () => DagRunService.getDagRun({ dagId, dagRunId }) as TData,
+ ...options,
+ });
+/**
+ * Get Dag Source
+ * Get source code using file token.
+ * @param data The data for the request.
+ * @param data.fileToken
+ * @param data.accept
+ * @returns DAGSourceResponse Successful Response
+ * @throws ApiError
+ */
+export const useDagSourceServiceGetDagSource = <
+ TData = Common.DagSourceServiceGetDagSourceDefaultResponse,
+ TError = unknown,
+ TQueryKey extends Array<unknown> = unknown[],
+>(
+ {
+ accept,
+ fileToken,
+ }: {
+ accept?: string;
+ fileToken: string;
+ },
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useQuery<TData, TError>({
+ queryKey: Common.UseDagSourceServiceGetDagSourceKeyFn(
+ { accept, fileToken },
+ queryKey,
+ ),
+ queryFn: () =>
+ DagSourceService.getDagSource({ accept, fileToken }) as TData,
+ ...options,
+ });
/**
* Get Event Log
* @param data The data for the request.
@@ -858,6 +926,44 @@ export const useVersionServiceGetVersion = <
queryFn: () => VersionService.getVersion() as TData,
...options,
});
+/**
+ * Create Backfill
+ * @param data The data for the request.
+ * @param data.requestBody
+ * @returns unknown Successful Response
+ * @throws ApiError
+ */
+export const useBackfillServiceCreateBackfill = <
+ TData = Common.BackfillServiceCreateBackfillMutationResult,
+ TError = unknown,
+ TContext = unknown,
+>(
+ options?: Omit<
+ UseMutationOptions<
+ TData,
+ TError,
+ {
+ requestBody: BackfillPostBody;
+ },
+ TContext
+ >,
+ "mutationFn"
+ >,
+) =>
+ useMutation<
+ TData,
+ TError,
+ {
+ requestBody: BackfillPostBody;
+ },
+ TContext
+ >({
+ mutationFn: ({ requestBody }) =>
+ BackfillService.createBackfill({
+ requestBody,
+ }) as unknown as Promise<TData>,
+ ...options,
+ });
/**
* Post Pool
* Create a Pool.
@@ -935,18 +1041,14 @@ export const useVariableServicePostVariable = <
...options,
});
/**
- * Patch Dag Run State
- * Modify a DAG Run.
+ * Pause Backfill
* @param data The data for the request.
- * @param data.dagId
- * @param data.dagRunId
- * @param data.requestBody
- * @param data.updateMask
- * @returns DAGRunResponse Successful Response
+ * @param data.backfillId
+ * @returns unknown Successful Response
* @throws ApiError
*/
-export const useDagRunServicePatchDagRunState = <
- TData = Common.DagRunServicePatchDagRunStateMutationResult,
+export const useBackfillServicePauseBackfill = <
+ TData = Common.BackfillServicePauseBackfillMutationResult,
TError = unknown,
TContext = unknown,
>(
@@ -955,10 +1057,7 @@ export const useDagRunServicePatchDagRunState = <
TData,
TError,
{
- dagId: string;
- dagRunId: string;
- requestBody: DAGRunPatchBody;
- updateMask?: string[];
+ backfillId: unknown;
},
TContext
>,
@@ -969,19 +1068,89 @@ export const useDagRunServicePatchDagRunState = <
TData,
TError,
{
- dagId: string;
- dagRunId: string;
- requestBody: DAGRunPatchBody;
- updateMask?: string[];
+ backfillId: unknown;
},
TContext
>({
- mutationFn: ({ dagId, dagRunId, requestBody, updateMask }) =>
- DagRunService.patchDagRunState({
- dagId,
- dagRunId,
- requestBody,
- updateMask,
+ mutationFn: ({ backfillId }) =>
+ BackfillService.pauseBackfill({
+ backfillId,
+ }) as unknown as Promise<TData>,
+ ...options,
+ });
+/**
+ * Unpause Backfill
+ * @param data The data for the request.
+ * @param data.backfillId
+ * @returns unknown Successful Response
+ * @throws ApiError
+ */
+export const useBackfillServiceUnpauseBackfill = <
+ TData = Common.BackfillServiceUnpauseBackfillMutationResult,
+ TError = unknown,
+ TContext = unknown,
+>(
+ options?: Omit<
+ UseMutationOptions<
+ TData,
+ TError,
+ {
+ backfillId: unknown;
+ },
+ TContext
+ >,
+ "mutationFn"
+ >,
+) =>
+ useMutation<
+ TData,
+ TError,
+ {
+ backfillId: unknown;
+ },
+ TContext
+ >({
+ mutationFn: ({ backfillId }) =>
+ BackfillService.unpauseBackfill({
+ backfillId,
+ }) as unknown as Promise<TData>,
+ ...options,
+ });
+/**
+ * Cancel Backfill
+ * @param data The data for the request.
+ * @param data.backfillId
+ * @returns unknown Successful Response
+ * @throws ApiError
+ */
+export const useBackfillServiceCancelBackfill = <
+ TData = Common.BackfillServiceCancelBackfillMutationResult,
+ TError = unknown,
+ TContext = unknown,
+>(
+ options?: Omit<
+ UseMutationOptions<
+ TData,
+ TError,
+ {
+ backfillId: unknown;
+ },
+ TContext
+ >,
+ "mutationFn"
+ >,
+) =>
+ useMutation<
+ TData,
+ TError,
+ {
+ backfillId: unknown;
+ },
+ TContext
+ >({
+ mutationFn: ({ backfillId }) =>
+ BackfillService.cancelBackfill({
+ backfillId,
}) as unknown as Promise<TData>,
...options,
});
@@ -1118,6 +1287,57 @@ export const useDagServicePatchDag = <
}) as unknown as Promise<TData>,
...options,
});
+/**
+ * Patch Dag Run State
+ * Modify a DAG Run.
+ * @param data The data for the request.
+ * @param data.dagId
+ * @param data.dagRunId
+ * @param data.requestBody
+ * @param data.updateMask
+ * @returns DAGRunResponse Successful Response
+ * @throws ApiError
+ */
+export const useDagRunServicePatchDagRunState = <
+ TData = Common.DagRunServicePatchDagRunStateMutationResult,
+ TError = unknown,
+ TContext = unknown,
+>(
+ options?: Omit<
+ UseMutationOptions<
+ TData,
+ TError,
+ {
+ dagId: string;
+ dagRunId: string;
+ requestBody: DAGRunPatchBody;
+ updateMask?: string[];
+ },
+ TContext
+ >,
+ "mutationFn"
+ >,
+) =>
+ useMutation<
+ TData,
+ TError,
+ {
+ dagId: string;
+ dagRunId: string;
+ requestBody: DAGRunPatchBody;
+ updateMask?: string[];
+ },
+ TContext
+ >({
+ mutationFn: ({ dagId, dagRunId, requestBody, updateMask }) =>
+ DagRunService.patchDagRunState({
+ dagId,
+ dagRunId,
+ requestBody,
+ updateMask,
+ }) as unknown as Promise<TData>,
+ ...options,
+ });
/**
* Patch Pool
* Update a Pool.
@@ -1213,15 +1433,15 @@ export const useVariableServicePatchVariable = <
...options,
});
/**
- * Delete Connection
- * Delete a connection entry.
+ * Delete Dag
+ * Delete the specific DAG.
* @param data The data for the request.
- * @param data.connectionId
- * @returns void Successful Response
+ * @param data.dagId
+ * @returns unknown Successful Response
* @throws ApiError
*/
-export const useConnectionServiceDeleteConnection = <
- TData = Common.ConnectionServiceDeleteConnectionMutationResult,
+export const useDagServiceDeleteDag = <
+ TData = Common.DagServiceDeleteDagMutationResult,
TError = unknown,
TContext = unknown,
>(
@@ -1230,7 +1450,7 @@ export const useConnectionServiceDeleteConnection = <
TData,
TError,
{
- connectionId: string;
+ dagId: string;
},
TContext
>,
@@ -1241,27 +1461,24 @@ export const useConnectionServiceDeleteConnection = <
TData,
TError,
{
- connectionId: string;
+ dagId: string;
},
TContext
>({
- mutationFn: ({ connectionId }) =>
- ConnectionService.deleteConnection({
- connectionId,
- }) as unknown as Promise<TData>,
+ mutationFn: ({ dagId }) =>
+ DagService.deleteDag({ dagId }) as unknown as Promise<TData>,
...options,
});
/**
- * Delete Dag Run
- * Delete a DAG Run entry.
+ * Delete Connection
+ * Delete a connection entry.
* @param data The data for the request.
- * @param data.dagId
- * @param data.dagRunId
+ * @param data.connectionId
* @returns void Successful Response
* @throws ApiError
*/
-export const useDagRunServiceDeleteDagRun = <
- TData = Common.DagRunServiceDeleteDagRunMutationResult,
+export const useConnectionServiceDeleteConnection = <
+ TData = Common.ConnectionServiceDeleteConnectionMutationResult,
TError = unknown,
TContext = unknown,
>(
@@ -1270,8 +1487,7 @@ export const useDagRunServiceDeleteDagRun = <
TData,
TError,
{
- dagId: string;
- dagRunId: string;
+ connectionId: string;
},
TContext
>,
@@ -1282,28 +1498,27 @@ export const useDagRunServiceDeleteDagRun = <
TData,
TError,
{
- dagId: string;
- dagRunId: string;
+ connectionId: string;
},
TContext
>({
- mutationFn: ({ dagId, dagRunId }) =>
- DagRunService.deleteDagRun({
- dagId,
- dagRunId,
+ mutationFn: ({ connectionId }) =>
+ ConnectionService.deleteConnection({
+ connectionId,
}) as unknown as Promise<TData>,
...options,
});
/**
- * Delete Dag
- * Delete the specific DAG.
+ * Delete Dag Run
+ * Delete a DAG Run entry.
* @param data The data for the request.
* @param data.dagId
- * @returns unknown Successful Response
+ * @param data.dagRunId
+ * @returns void Successful Response
* @throws ApiError
*/
-export const useDagServiceDeleteDag = <
- TData = Common.DagServiceDeleteDagMutationResult,
+export const useDagRunServiceDeleteDagRun = <
+ TData = Common.DagRunServiceDeleteDagRunMutationResult,
TError = unknown,
TContext = unknown,
>(
@@ -1313,6 +1528,7 @@ export const useDagServiceDeleteDag = <
TError,
{
dagId: string;
+ dagRunId: string;
},
TContext
>,
@@ -1324,11 +1540,15 @@ export const useDagServiceDeleteDag = <
TError,
{
dagId: string;
+ dagRunId: string;
},
TContext
>({
- mutationFn: ({ dagId }) =>
- DagService.deleteDag({ dagId }) as unknown as Promise<TData>,
+ mutationFn: ({ dagId, dagRunId }) =>
+ DagRunService.deleteDagRun({
+ dagId,
+ dagRunId,
+ }) as unknown as Promise<TData>,
...options,
});
/**
diff --git a/airflow/ui/openapi-gen/queries/suspense.ts b/airflow/ui/openapi-gen/queries/suspense.ts
index 6ceed83349d..1222b7f5536 100644
--- a/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow/ui/openapi-gen/queries/suspense.ts
@@ -3,6 +3,7 @@ import { UseQueryOptions, useSuspenseQuery } from "@tanstack/react-query";
import {
AssetService,
+ BackfillService,
ConnectionService,
DagRunService,
DagService,
@@ -160,132 +161,69 @@ export const useDagsServiceRecentDagRunsSuspense = <
...options,
});
/**
- * Get Connection
- * Get a connection entry.
- * @param data The data for the request.
- * @param data.connectionId
- * @returns ConnectionResponse Successful Response
- * @throws ApiError
- */
-export const useConnectionServiceGetConnectionSuspense = <
- TData = Common.ConnectionServiceGetConnectionDefaultResponse,
- TError = unknown,
- TQueryKey extends Array<unknown> = unknown[],
->(
- {
- connectionId,
- }: {
- connectionId: string;
- },
- queryKey?: TQueryKey,
- options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
-) =>
- useSuspenseQuery<TData, TError>({
- queryKey: Common.UseConnectionServiceGetConnectionKeyFn(
- { connectionId },
- queryKey,
- ),
- queryFn: () => ConnectionService.getConnection({ connectionId }) as TData,
- ...options,
- });
-/**
- * Get Connections
- * Get all connection entries.
+ * List Backfills
* @param data The data for the request.
+ * @param data.dagId
* @param data.limit
* @param data.offset
* @param data.orderBy
- * @returns ConnectionCollectionResponse Successful Response
+ * @returns unknown Successful Response
* @throws ApiError
*/
-export const useConnectionServiceGetConnectionsSuspense = <
- TData = Common.ConnectionServiceGetConnectionsDefaultResponse,
+export const useBackfillServiceListBackfillsSuspense = <
+ TData = Common.BackfillServiceListBackfillsDefaultResponse,
TError = unknown,
TQueryKey extends Array<unknown> = unknown[],
>(
{
+ dagId,
limit,
offset,
orderBy,
}: {
+ dagId: string;
limit?: number;
offset?: number;
orderBy?: string;
- } = {},
- queryKey?: TQueryKey,
- options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
-) =>
- useSuspenseQuery<TData, TError>({
- queryKey: Common.UseConnectionServiceGetConnectionsKeyFn(
- { limit, offset, orderBy },
- queryKey,
- ),
- queryFn: () =>
- ConnectionService.getConnections({ limit, offset, orderBy }) as TData,
- ...options,
- });
-/**
- * Get Dag Run
- * @param data The data for the request.
- * @param data.dagId
- * @param data.dagRunId
- * @returns DAGRunResponse Successful Response
- * @throws ApiError
- */
-export const useDagRunServiceGetDagRunSuspense = <
- TData = Common.DagRunServiceGetDagRunDefaultResponse,
- TError = unknown,
- TQueryKey extends Array<unknown> = unknown[],
->(
- {
- dagId,
- dagRunId,
- }: {
- dagId: string;
- dagRunId: string;
},
queryKey?: TQueryKey,
options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
) =>
useSuspenseQuery<TData, TError>({
- queryKey: Common.UseDagRunServiceGetDagRunKeyFn(
- { dagId, dagRunId },
+ queryKey: Common.UseBackfillServiceListBackfillsKeyFn(
+ { dagId, limit, offset, orderBy },
queryKey,
),
- queryFn: () => DagRunService.getDagRun({ dagId, dagRunId }) as TData,
+ queryFn: () =>
+ BackfillService.listBackfills({ dagId, limit, offset, orderBy }) as TData,
...options,
});
/**
- * Get Dag Source
- * Get source code using file token.
+ * Get Backfill
* @param data The data for the request.
- * @param data.fileToken
- * @param data.accept
- * @returns DAGSourceResponse Successful Response
+ * @param data.backfillId
+ * @returns unknown Successful Response
* @throws ApiError
*/
-export const useDagSourceServiceGetDagSourceSuspense = <
- TData = Common.DagSourceServiceGetDagSourceDefaultResponse,
+export const useBackfillServiceGetBackfillSuspense = <
+ TData = Common.BackfillServiceGetBackfillDefaultResponse,
TError = unknown,
TQueryKey extends Array<unknown> = unknown[],
>(
{
- accept,
- fileToken,
+ backfillId,
}: {
- accept?: string;
- fileToken: string;
+ backfillId: string;
},
queryKey?: TQueryKey,
options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
) =>
useSuspenseQuery<TData, TError>({
- queryKey: Common.UseDagSourceServiceGetDagSourceKeyFn(
- { accept, fileToken },
+ queryKey: Common.UseBackfillServiceGetBackfillKeyFn(
+ { backfillId },
queryKey,
),
- queryFn: () =>
- DagSourceService.getDagSource({ accept, fileToken }) as TData,
+ queryFn: () => BackfillService.getBackfill({ backfillId }) as TData,
...options,
});
/**
@@ -463,6 +401,135 @@ export const useDagServiceGetDagDetailsSuspense = <
queryFn: () => DagService.getDagDetails({ dagId }) as TData,
...options,
});
+/**
+ * Get Connection
+ * Get a connection entry.
+ * @param data The data for the request.
+ * @param data.connectionId
+ * @returns ConnectionResponse Successful Response
+ * @throws ApiError
+ */
+export const useConnectionServiceGetConnectionSuspense = <
+ TData = Common.ConnectionServiceGetConnectionDefaultResponse,
+ TError = unknown,
+ TQueryKey extends Array<unknown> = unknown[],
+>(
+ {
+ connectionId,
+ }: {
+ connectionId: string;
+ },
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useSuspenseQuery<TData, TError>({
+ queryKey: Common.UseConnectionServiceGetConnectionKeyFn(
+ { connectionId },
+ queryKey,
+ ),
+ queryFn: () => ConnectionService.getConnection({ connectionId }) as TData,
+ ...options,
+ });
+/**
+ * Get Connections
+ * Get all connection entries.
+ * @param data The data for the request.
+ * @param data.limit
+ * @param data.offset
+ * @param data.orderBy
+ * @returns ConnectionCollectionResponse Successful Response
+ * @throws ApiError
+ */
+export const useConnectionServiceGetConnectionsSuspense = <
+ TData = Common.ConnectionServiceGetConnectionsDefaultResponse,
+ TError = unknown,
+ TQueryKey extends Array<unknown> = unknown[],
+>(
+ {
+ limit,
+ offset,
+ orderBy,
+ }: {
+ limit?: number;
+ offset?: number;
+ orderBy?: string;
+ } = {},
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useSuspenseQuery<TData, TError>({
+ queryKey: Common.UseConnectionServiceGetConnectionsKeyFn(
+ { limit, offset, orderBy },
+ queryKey,
+ ),
+ queryFn: () =>
+ ConnectionService.getConnections({ limit, offset, orderBy }) as TData,
+ ...options,
+ });
+/**
+ * Get Dag Run
+ * @param data The data for the request.
+ * @param data.dagId
+ * @param data.dagRunId
+ * @returns DAGRunResponse Successful Response
+ * @throws ApiError
+ */
+export const useDagRunServiceGetDagRunSuspense = <
+ TData = Common.DagRunServiceGetDagRunDefaultResponse,
+ TError = unknown,
+ TQueryKey extends Array<unknown> = unknown[],
+>(
+ {
+ dagId,
+ dagRunId,
+ }: {
+ dagId: string;
+ dagRunId: string;
+ },
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useSuspenseQuery<TData, TError>({
+ queryKey: Common.UseDagRunServiceGetDagRunKeyFn(
+ { dagId, dagRunId },
+ queryKey,
+ ),
+ queryFn: () => DagRunService.getDagRun({ dagId, dagRunId }) as TData,
+ ...options,
+ });
+/**
+ * Get Dag Source
+ * Get source code using file token.
+ * @param data The data for the request.
+ * @param data.fileToken
+ * @param data.accept
+ * @returns DAGSourceResponse Successful Response
+ * @throws ApiError
+ */
+export const useDagSourceServiceGetDagSourceSuspense = <
+ TData = Common.DagSourceServiceGetDagSourceDefaultResponse,
+ TError = unknown,
+ TQueryKey extends Array<unknown> = unknown[],
+>(
+ {
+ accept,
+ fileToken,
+ }: {
+ accept?: string;
+ fileToken: string;
+ },
+ queryKey?: TQueryKey,
+ options?: Omit<UseQueryOptions<TData, TError>, "queryKey" | "queryFn">,
+) =>
+ useSuspenseQuery<TData, TError>({
+ queryKey: Common.UseDagSourceServiceGetDagSourceKeyFn(
+ { accept, fileToken },
+ queryKey,
+ ),
+ queryFn: () =>
+ DagSourceService.getDagSource({ accept, fileToken }) as TData,
+ ...options,
+ });
/**
* Get Event Log
* @param data The data for the request.
diff --git a/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 3f74dc46a8a..5624358636c 100644
--- a/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -89,6 +89,48 @@ export const $AppBuilderViewResponse = {
description: "Serializer for AppBuilder View responses.",
} as const;
+export const $BackfillPostBody = {
+ properties: {
+ dag_id: {
+ type: "string",
+ title: "Dag Id",
+ },
+ from_date: {
+ type: "string",
+ format: "date-time",
+ title: "From Date",
+ },
+ to_date: {
+ type: "string",
+ format: "date-time",
+ title: "To Date",
+ },
+ run_backwards: {
+ type: "boolean",
+ title: "Run Backwards",
+ default: false,
+ },
+ dag_run_conf: {
+ type: "object",
+ title: "Dag Run Conf",
+ default: {},
+ },
+ reprocess_behavior: {
+ $ref: "#/components/schemas/ReprocessBehavior",
+ default: "none",
+ },
+ max_active_runs: {
+ type: "integer",
+ title: "Max Active Runs",
+ default: 10,
+ },
+ },
+ type: "object",
+ required: ["dag_id", "from_date", "to_date"],
+ title: "BackfillPostBody",
+ description: "Object used for create backfill request.",
+} as const;
+
export const $BaseInfoSchema = {
properties: {
status: {
@@ -2216,6 +2258,15 @@ export const $ProviderResponse = {
description: "Provider serializer for responses.",
} as const;
+export const $ReprocessBehavior = {
+ type: "string",
+ enum: ["failed", "completed", "none"],
+ title: "ReprocessBehavior",
+ description: `Internal enum for setting reprocess behavior in a backfill.
+
+:meta private:`,
+} as const;
+
export const $SchedulerInfoSchema = {
properties: {
status: {
diff --git a/airflow/ui/openapi-gen/requests/services.gen.ts b/airflow/ui/openapi-gen/requests/services.gen.ts
index 6c42a500e97..e82d0c5146f 100644
--- a/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -9,20 +9,18 @@ import type {
HistoricalMetricsResponse,
RecentDagRunsData,
RecentDagRunsResponse,
- DeleteConnectionData,
- DeleteConnectionResponse,
- GetConnectionData,
- GetConnectionResponse,
- GetConnectionsData,
- GetConnectionsResponse,
- GetDagRunData,
- GetDagRunResponse,
- DeleteDagRunData,
- DeleteDagRunResponse,
- PatchDagRunStateData,
- PatchDagRunStateResponse,
- GetDagSourceData,
- GetDagSourceResponse,
+ ListBackfillsData,
+ ListBackfillsResponse,
+ CreateBackfillData,
+ CreateBackfillResponse,
+ GetBackfillData,
+ GetBackfillResponse,
+ PauseBackfillData,
+ PauseBackfillResponse,
+ UnpauseBackfillData,
+ UnpauseBackfillResponse,
+ CancelBackfillData,
+ CancelBackfillResponse,
GetDagsData,
GetDagsResponse,
PatchDagsData,
@@ -37,6 +35,20 @@ import type {
DeleteDagResponse,
GetDagDetailsData,
GetDagDetailsResponse,
+ DeleteConnectionData,
+ DeleteConnectionResponse,
+ GetConnectionData,
+ GetConnectionResponse,
+ GetConnectionsData,
+ GetConnectionsResponse,
+ GetDagRunData,
+ GetDagRunResponse,
+ DeleteDagRunData,
+ DeleteDagRunResponse,
+ PatchDagRunStateData,
+ PatchDagRunStateResponse,
+ GetDagSourceData,
+ GetDagSourceResponse,
GetEventLogData,
GetEventLogResponse,
GetHealthResponse,
@@ -168,108 +180,77 @@ export class DagsService {
}
}
-export class ConnectionService {
- /**
- * Delete Connection
- * Delete a connection entry.
- * @param data The data for the request.
- * @param data.connectionId
- * @returns void Successful Response
- * @throws ApiError
- */
- public static deleteConnection(
- data: DeleteConnectionData,
- ): CancelablePromise<DeleteConnectionResponse> {
- return __request(OpenAPI, {
- method: "DELETE",
- url: "/public/connections/{connection_id}",
- path: {
- connection_id: data.connectionId,
- },
- errors: {
- 401: "Unauthorized",
- 403: "Forbidden",
- 404: "Not Found",
- 422: "Validation Error",
- },
- });
- }
-
+export class BackfillService {
/**
- * Get Connection
- * Get a connection entry.
+ * List Backfills
* @param data The data for the request.
- * @param data.connectionId
- * @returns ConnectionResponse Successful Response
+ * @param data.dagId
+ * @param data.limit
+ * @param data.offset
+ * @param data.orderBy
+ * @returns unknown Successful Response
* @throws ApiError
*/
- public static getConnection(
- data: GetConnectionData,
- ): CancelablePromise<GetConnectionResponse> {
+ public static listBackfills(
+ data: ListBackfillsData,
+ ): CancelablePromise<ListBackfillsResponse> {
return __request(OpenAPI, {
method: "GET",
- url: "/public/connections/{connection_id}",
- path: {
- connection_id: data.connectionId,
+ url: "/public/backfills/",
+ query: {
+ dag_id: data.dagId,
+ limit: data.limit,
+ offset: data.offset,
+ order_by: data.orderBy,
},
errors: {
401: "Unauthorized",
403: "Forbidden",
- 404: "Not Found",
422: "Validation Error",
},
});
}
/**
- * Get Connections
- * Get all connection entries.
+ * Create Backfill
* @param data The data for the request.
- * @param data.limit
- * @param data.offset
- * @param data.orderBy
- * @returns ConnectionCollectionResponse Successful Response
+ * @param data.requestBody
+ * @returns unknown Successful Response
* @throws ApiError
*/
- public static getConnections(
- data: GetConnectionsData = {},
- ): CancelablePromise<GetConnectionsResponse> {
+ public static createBackfill(
+ data: CreateBackfillData,
+ ): CancelablePromise<CreateBackfillResponse> {
return __request(OpenAPI, {
- method: "GET",
- url: "/public/connections/",
- query: {
- limit: data.limit,
- offset: data.offset,
- order_by: data.orderBy,
- },
+ method: "POST",
+ url: "/public/backfills/",
+ body: data.requestBody,
+ mediaType: "application/json",
errors: {
401: "Unauthorized",
403: "Forbidden",
404: "Not Found",
+ 409: "Conflict",
422: "Validation Error",
},
});
}
-}
-export class DagRunService {
/**
- * Get Dag Run
+ * Get Backfill
* @param data The data for the request.
- * @param data.dagId
- * @param data.dagRunId
- * @returns DAGRunResponse Successful Response
+ * @param data.backfillId
+ * @returns unknown Successful Response
* @throws ApiError
*/
- public static getDagRun(
- data: GetDagRunData,
- ): CancelablePromise<GetDagRunResponse> {
+ public static getBackfill(
+ data: GetBackfillData,
+ ): CancelablePromise<GetBackfillResponse> {
return __request(OpenAPI, {
method: "GET",
- url: "/public/dags/{dag_id}/dagRuns/{dag_run_id}",
+ url: "/public/backfills/{backfill_id}",
path: {
- dag_id: data.dagId,
- dag_run_id: data.dagRunId,
+ backfill_id: data.backfillId,
},
errors: {
401: "Unauthorized",
@@ -281,99 +262,78 @@ export class DagRunService {
}
/**
- * Delete Dag Run
- * Delete a DAG Run entry.
+ * Pause Backfill
* @param data The data for the request.
- * @param data.dagId
- * @param data.dagRunId
- * @returns void Successful Response
+ * @param data.backfillId
+ * @returns unknown Successful Response
* @throws ApiError
*/
- public static deleteDagRun(
- data: DeleteDagRunData,
- ): CancelablePromise<DeleteDagRunResponse> {
+ public static pauseBackfill(
+ data: PauseBackfillData,
+ ): CancelablePromise<PauseBackfillResponse> {
return __request(OpenAPI, {
- method: "DELETE",
- url: "/public/dags/{dag_id}/dagRuns/{dag_run_id}",
+ method: "PUT",
+ url: "/public/backfills/{backfill_id}/pause",
path: {
- dag_id: data.dagId,
- dag_run_id: data.dagRunId,
+ backfill_id: data.backfillId,
},
errors: {
- 400: "Bad Request",
401: "Unauthorized",
403: "Forbidden",
404: "Not Found",
+ 409: "Conflict",
422: "Validation Error",
},
});
}
/**
- * Patch Dag Run State
- * Modify a DAG Run.
+ * Unpause Backfill
* @param data The data for the request.
- * @param data.dagId
- * @param data.dagRunId
- * @param data.requestBody
- * @param data.updateMask
- * @returns DAGRunResponse Successful Response
+ * @param data.backfillId
+ * @returns unknown Successful Response
* @throws ApiError
*/
- public static patchDagRunState(
- data: PatchDagRunStateData,
- ): CancelablePromise<PatchDagRunStateResponse> {
+ public static unpauseBackfill(
+ data: UnpauseBackfillData,
+ ): CancelablePromise<UnpauseBackfillResponse> {
return __request(OpenAPI, {
- method: "PATCH",
- url: "/public/dags/{dag_id}/dagRuns/{dag_run_id}",
+ method: "PUT",
+ url: "/public/backfills/{backfill_id}/unpause",
path: {
- dag_id: data.dagId,
- dag_run_id: data.dagRunId,
- },
- query: {
- update_mask: data.updateMask,
+ backfill_id: data.backfillId,
},
- body: data.requestBody,
- mediaType: "application/json",
errors: {
- 400: "Bad Request",
401: "Unauthorized",
403: "Forbidden",
404: "Not Found",
+ 409: "Conflict",
422: "Validation Error",
},
});
}
-}
-export class DagSourceService {
/**
- * Get Dag Source
- * Get source code using file token.
+ * Cancel Backfill
* @param data The data for the request.
- * @param data.fileToken
- * @param data.accept
- * @returns DAGSourceResponse Successful Response
+ * @param data.backfillId
+ * @returns unknown Successful Response
* @throws ApiError
*/
- public static getDagSource(
- data: GetDagSourceData,
- ): CancelablePromise<GetDagSourceResponse> {
+ public static cancelBackfill(
+ data: CancelBackfillData,
+ ): CancelablePromise<CancelBackfillResponse> {
return __request(OpenAPI, {
- method: "GET",
- url: "/public/dagSources/{file_token}",
+ method: "PUT",
+ url: "/public/backfills/{backfill_id}/cancel",
path: {
- file_token: data.fileToken,
- },
- headers: {
- accept: data.accept,
+ backfill_id: data.backfillId,
},
errors: {
- 400: "Bad Request",
401: "Unauthorized",
403: "Forbidden",
404: "Not Found",
- 406: "Not Acceptable",
+ 409: "Conflict",
422: "Validation Error",
},
});
@@ -613,6 +573,218 @@ export class DagService {
}
}
+export class ConnectionService {
+ /**
+ * Delete Connection
+ * Delete a connection entry.
+ * @param data The data for the request.
+ * @param data.connectionId
+ * @returns void Successful Response
+ * @throws ApiError
+ */
+ public static deleteConnection(
+ data: DeleteConnectionData,
+ ): CancelablePromise<DeleteConnectionResponse> {
+ return __request(OpenAPI, {
+ method: "DELETE",
+ url: "/public/connections/{connection_id}",
+ path: {
+ connection_id: data.connectionId,
+ },
+ errors: {
+ 401: "Unauthorized",
+ 403: "Forbidden",
+ 404: "Not Found",
+ 422: "Validation Error",
+ },
+ });
+ }
+
+ /**
+ * Get Connection
+ * Get a connection entry.
+ * @param data The data for the request.
+ * @param data.connectionId
+ * @returns ConnectionResponse Successful Response
+ * @throws ApiError
+ */
+ public static getConnection(
+ data: GetConnectionData,
+ ): CancelablePromise<GetConnectionResponse> {
+ return __request(OpenAPI, {
+ method: "GET",
+ url: "/public/connections/{connection_id}",
+ path: {
+ connection_id: data.connectionId,
+ },
+ errors: {
+ 401: "Unauthorized",
+ 403: "Forbidden",
+ 404: "Not Found",
+ 422: "Validation Error",
+ },
+ });
+ }
+
+ /**
+ * Get Connections
+ * Get all connection entries.
+ * @param data The data for the request.
+ * @param data.limit
+ * @param data.offset
+ * @param data.orderBy
+ * @returns ConnectionCollectionResponse Successful Response
+ * @throws ApiError
+ */
+ public static getConnections(
+ data: GetConnectionsData = {},
+ ): CancelablePromise<GetConnectionsResponse> {
+ return __request(OpenAPI, {
+ method: "GET",
+ url: "/public/connections/",
+ query: {
+ limit: data.limit,
+ offset: data.offset,
+ order_by: data.orderBy,
+ },
+ errors: {
+ 401: "Unauthorized",
+ 403: "Forbidden",
+ 404: "Not Found",
+ 422: "Validation Error",
+ },
+ });
+ }
+}
+
+export class DagRunService {
+ /**
+ * Get Dag Run
+ * @param data The data for the request.
+ * @param data.dagId
+ * @param data.dagRunId
+ * @returns DAGRunResponse Successful Response
+ * @throws ApiError
+ */
+ public static getDagRun(
+ data: GetDagRunData,
+ ): CancelablePromise<GetDagRunResponse> {
+ return __request(OpenAPI, {
+ method: "GET",
+ url: "/public/dags/{dag_id}/dagRuns/{dag_run_id}",
+ path: {
+ dag_id: data.dagId,
+ dag_run_id: data.dagRunId,
+ },
+ errors: {
+ 401: "Unauthorized",
+ 403: "Forbidden",
+ 404: "Not Found",
+ 422: "Validation Error",
+ },
+ });
+ }
+
+ /**
+ * Delete Dag Run
+ * Delete a DAG Run entry.
+ * @param data The data for the request.
+ * @param data.dagId
+ * @param data.dagRunId
+ * @returns void Successful Response
+ * @throws ApiError
+ */
+ public static deleteDagRun(
+ data: DeleteDagRunData,
+ ): CancelablePromise<DeleteDagRunResponse> {
+ return __request(OpenAPI, {
+ method: "DELETE",
+ url: "/public/dags/{dag_id}/dagRuns/{dag_run_id}",
+ path: {
+ dag_id: data.dagId,
+ dag_run_id: data.dagRunId,
+ },
+ errors: {
+ 400: "Bad Request",
+ 401: "Unauthorized",
+ 403: "Forbidden",
+ 404: "Not Found",
+ 422: "Validation Error",
+ },
+ });
+ }
+
+ /**
+ * Patch Dag Run State
+ * Modify a DAG Run.
+ * @param data The data for the request.
+ * @param data.dagId
+ * @param data.dagRunId
+ * @param data.requestBody
+ * @param data.updateMask
+ * @returns DAGRunResponse Successful Response
+ * @throws ApiError
+ */
+ public static patchDagRunState(
+ data: PatchDagRunStateData,
+ ): CancelablePromise<PatchDagRunStateResponse> {
+ return __request(OpenAPI, {
+ method: "PATCH",
+ url: "/public/dags/{dag_id}/dagRuns/{dag_run_id}",
+ path: {
+ dag_id: data.dagId,
+ dag_run_id: data.dagRunId,
+ },
+ query: {
+ update_mask: data.updateMask,
+ },
+ body: data.requestBody,
+ mediaType: "application/json",
+ errors: {
+ 400: "Bad Request",
+ 401: "Unauthorized",
+ 403: "Forbidden",
+ 404: "Not Found",
+ 422: "Validation Error",
+ },
+ });
+ }
+}
+
+export class DagSourceService {
+ /**
+ * Get Dag Source
+ * Get source code using file token.
+ * @param data The data for the request.
+ * @param data.fileToken
+ * @param data.accept
+ * @returns DAGSourceResponse Successful Response
+ * @throws ApiError
+ */
+ public static getDagSource(
+ data: GetDagSourceData,
+ ): CancelablePromise<GetDagSourceResponse> {
+ return __request(OpenAPI, {
+ method: "GET",
+ url: "/public/dagSources/{file_token}",
+ path: {
+ file_token: data.fileToken,
+ },
+ headers: {
+ accept: data.accept,
+ },
+ errors: {
+ 400: "Bad Request",
+ 401: "Unauthorized",
+ 403: "Forbidden",
+ 404: "Not Found",
+ 406: "Not Acceptable",
+ 422: "Validation Error",
+ },
+ });
+ }
+}
+
export class EventLogService {
/**
* Get Event Log
diff --git a/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow/ui/openapi-gen/requests/types.gen.ts
index 4a06652e380..2c11dd03cc0 100644
--- a/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -21,6 +21,21 @@ export type AppBuilderViewResponse = {
[key: string]: unknown;
};
+/**
+ * Object used for create backfill request.
+ */
+export type BackfillPostBody = {
+ dag_id: string;
+ from_date: string;
+ to_date: string;
+ run_backwards?: boolean;
+ dag_run_conf?: {
+ [key: string]: unknown;
+ };
+ reprocess_behavior?: ReprocessBehavior;
+ max_active_runs?: number;
+};
+
/**
* Base status field for metadatabase and scheduler.
*/
@@ -502,6 +517,13 @@ export type ProviderResponse = {
version: string;
};
+/**
+ * Internal enum for setting reprocess behavior in a backfill.
+ *
+ * :meta private:
+ */
+export type ReprocessBehavior = "failed" | "completed" | "none";
+
/**
* Schema for Scheduler info.
*/
@@ -674,55 +696,44 @@ export type RecentDagRunsData = {
export type RecentDagRunsResponse = DAGWithLatestDagRunsCollectionResponse;
-export type DeleteConnectionData = {
- connectionId: string;
-};
-
-export type DeleteConnectionResponse = void;
-
-export type GetConnectionData = {
- connectionId: string;
-};
-
-export type GetConnectionResponse = ConnectionResponse;
-
-export type GetConnectionsData = {
+export type ListBackfillsData = {
+ dagId: string;
limit?: number;
offset?: number;
orderBy?: string;
};
-export type GetConnectionsResponse = ConnectionCollectionResponse;
+export type ListBackfillsResponse = unknown;
-export type GetDagRunData = {
- dagId: string;
- dagRunId: string;
+export type CreateBackfillData = {
+ requestBody: BackfillPostBody;
};
-export type GetDagRunResponse = DAGRunResponse;
+export type CreateBackfillResponse = unknown;
-export type DeleteDagRunData = {
- dagId: string;
- dagRunId: string;
+export type GetBackfillData = {
+ backfillId: string;
};
-export type DeleteDagRunResponse = void;
+export type GetBackfillResponse = unknown;
-export type PatchDagRunStateData = {
- dagId: string;
- dagRunId: string;
- requestBody: DAGRunPatchBody;
- updateMask?: Array<string> | null;
+export type PauseBackfillData = {
+ backfillId: unknown;
};
-export type PatchDagRunStateResponse = DAGRunResponse;
+export type PauseBackfillResponse = unknown;
-export type GetDagSourceData = {
- accept?: string;
- fileToken: string;
+export type UnpauseBackfillData = {
+ backfillId: unknown;
};
-export type GetDagSourceResponse = DAGSourceResponse;
+export type UnpauseBackfillResponse = unknown;
+
+export type CancelBackfillData = {
+ backfillId: unknown;
+};
+
+export type CancelBackfillResponse = unknown;
export type GetDagsData = {
dagDisplayNamePattern?: string | null;
@@ -789,6 +800,56 @@ export type GetDagDetailsData = {
export type GetDagDetailsResponse = DAGDetailsResponse;
+export type DeleteConnectionData = {
+ connectionId: string;
+};
+
+export type DeleteConnectionResponse = void;
+
+export type GetConnectionData = {
+ connectionId: string;
+};
+
+export type GetConnectionResponse = ConnectionResponse;
+
+export type GetConnectionsData = {
+ limit?: number;
+ offset?: number;
+ orderBy?: string;
+};
+
+export type GetConnectionsResponse = ConnectionCollectionResponse;
+
+export type GetDagRunData = {
+ dagId: string;
+ dagRunId: string;
+};
+
+export type GetDagRunResponse = DAGRunResponse;
+
+export type DeleteDagRunData = {
+ dagId: string;
+ dagRunId: string;
+};
+
+export type DeleteDagRunResponse = void;
+
+export type PatchDagRunStateData = {
+ dagId: string;
+ dagRunId: string;
+ requestBody: DAGRunPatchBody;
+ updateMask?: Array<string> | null;
+};
+
+export type PatchDagRunStateResponse = DAGRunResponse;
+
+export type GetDagSourceData = {
+ accept?: string;
+ fileToken: string;
+};
+
+export type GetDagSourceResponse = DAGSourceResponse;
+
export type GetEventLogData = {
eventLogId: number;
};
@@ -960,14 +1021,14 @@ export type $OpenApiTs = {
};
};
};
- "/public/connections/{connection_id}": {
- delete: {
- req: DeleteConnectionData;
+ "/public/backfills/": {
+ get: {
+ req: ListBackfillsData;
res: {
/**
* Successful Response
*/
- 204: void;
+ 200: unknown;
/**
* Unauthorized
*/
@@ -976,23 +1037,19 @@ export type $OpenApiTs = {
* Forbidden
*/
403: HTTPExceptionResponse;
- /**
- * Not Found
- */
- 404: HTTPExceptionResponse;
/**
* Validation Error
*/
422: HTTPValidationError;
};
};
- get: {
- req: GetConnectionData;
+ post: {
+ req: CreateBackfillData;
res: {
/**
* Successful Response
*/
- 200: ConnectionResponse;
+ 200: unknown;
/**
* Unauthorized
*/
@@ -1005,6 +1062,10 @@ export type $OpenApiTs = {
* Not Found
*/
404: HTTPExceptionResponse;
+ /**
+ * Conflict
+ */
+ 409: HTTPExceptionResponse;
/**
* Validation Error
*/
@@ -1012,14 +1073,14 @@ export type $OpenApiTs = {
};
};
};
- "/public/connections/": {
+ "/public/backfills/{backfill_id}": {
get: {
- req: GetConnectionsData;
+ req: GetBackfillData;
res: {
/**
* Successful Response
*/
- 200: ConnectionCollectionResponse;
+ 200: unknown;
/**
* Unauthorized
*/
@@ -1039,14 +1100,14 @@ export type $OpenApiTs = {
};
};
};
- "/public/dags/{dag_id}/dagRuns/{dag_run_id}": {
- get: {
- req: GetDagRunData;
+ "/public/backfills/{backfill_id}/pause": {
+ put: {
+ req: PauseBackfillData;
res: {
/**
* Successful Response
*/
- 200: DAGRunResponse;
+ 200: unknown;
/**
* Unauthorized
*/
@@ -1060,51 +1121,24 @@ export type $OpenApiTs = {
*/
404: HTTPExceptionResponse;
/**
- * Validation Error
+ * Conflict
*/
- 422: HTTPValidationError;
- };
- };
- delete: {
- req: DeleteDagRunData;
- res: {
- /**
- * Successful Response
- */
- 204: void;
- /**
- * Bad Request
- */
- 400: HTTPExceptionResponse;
- /**
- * Unauthorized
- */
- 401: HTTPExceptionResponse;
- /**
- * Forbidden
- */
- 403: HTTPExceptionResponse;
- /**
- * Not Found
- */
- 404: HTTPExceptionResponse;
+ 409: HTTPExceptionResponse;
/**
* Validation Error
*/
422: HTTPValidationError;
};
};
- patch: {
- req: PatchDagRunStateData;
+ };
+ "/public/backfills/{backfill_id}/unpause": {
+ put: {
+ req: UnpauseBackfillData;
res: {
/**
* Successful Response
*/
- 200: DAGRunResponse;
- /**
- * Bad Request
- */
- 400: HTTPExceptionResponse;
+ 200: unknown;
/**
* Unauthorized
*/
@@ -1117,6 +1151,10 @@ export type $OpenApiTs = {
* Not Found
*/
404: HTTPExceptionResponse;
+ /**
+ * Conflict
+ */
+ 409: HTTPExceptionResponse;
/**
* Validation Error
*/
@@ -1124,18 +1162,14 @@ export type $OpenApiTs = {
};
};
};
- "/public/dagSources/{file_token}": {
- get: {
- req: GetDagSourceData;
+ "/public/backfills/{backfill_id}/cancel": {
+ put: {
+ req: CancelBackfillData;
res: {
/**
* Successful Response
*/
- 200: DAGSourceResponse;
- /**
- * Bad Request
- */
- 400: HTTPExceptionResponse;
+ 200: unknown;
/**
* Unauthorized
*/
@@ -1149,9 +1183,9 @@ export type $OpenApiTs = {
*/
404: HTTPExceptionResponse;
/**
- * Not Acceptable
+ * Conflict
*/
- 406: HTTPExceptionResponse;
+ 409: HTTPExceptionResponse;
/**
* Validation Error
*/
@@ -1346,6 +1380,205 @@ export type $OpenApiTs = {
};
};
};
+ "/public/connections/{connection_id}": {
+ delete: {
+ req: DeleteConnectionData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 204: void;
+ /**
+ * Unauthorized
+ */
+ 401: HTTPExceptionResponse;
+ /**
+ * Forbidden
+ */
+ 403: HTTPExceptionResponse;
+ /**
+ * Not Found
+ */
+ 404: HTTPExceptionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ get: {
+ req: GetConnectionData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 200: ConnectionResponse;
+ /**
+ * Unauthorized
+ */
+ 401: HTTPExceptionResponse;
+ /**
+ * Forbidden
+ */
+ 403: HTTPExceptionResponse;
+ /**
+ * Not Found
+ */
+ 404: HTTPExceptionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ };
+ "/public/connections/": {
+ get: {
+ req: GetConnectionsData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 200: ConnectionCollectionResponse;
+ /**
+ * Unauthorized
+ */
+ 401: HTTPExceptionResponse;
+ /**
+ * Forbidden
+ */
+ 403: HTTPExceptionResponse;
+ /**
+ * Not Found
+ */
+ 404: HTTPExceptionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ };
+ "/public/dags/{dag_id}/dagRuns/{dag_run_id}": {
+ get: {
+ req: GetDagRunData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 200: DAGRunResponse;
+ /**
+ * Unauthorized
+ */
+ 401: HTTPExceptionResponse;
+ /**
+ * Forbidden
+ */
+ 403: HTTPExceptionResponse;
+ /**
+ * Not Found
+ */
+ 404: HTTPExceptionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ delete: {
+ req: DeleteDagRunData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 204: void;
+ /**
+ * Bad Request
+ */
+ 400: HTTPExceptionResponse;
+ /**
+ * Unauthorized
+ */
+ 401: HTTPExceptionResponse;
+ /**
+ * Forbidden
+ */
+ 403: HTTPExceptionResponse;
+ /**
+ * Not Found
+ */
+ 404: HTTPExceptionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ patch: {
+ req: PatchDagRunStateData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 200: DAGRunResponse;
+ /**
+ * Bad Request
+ */
+ 400: HTTPExceptionResponse;
+ /**
+ * Unauthorized
+ */
+ 401: HTTPExceptionResponse;
+ /**
+ * Forbidden
+ */
+ 403: HTTPExceptionResponse;
+ /**
+ * Not Found
+ */
+ 404: HTTPExceptionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ };
+ "/public/dagSources/{file_token}": {
+ get: {
+ req: GetDagSourceData;
+ res: {
+ /**
+ * Successful Response
+ */
+ 200: DAGSourceResponse;
+ /**
+ * Bad Request
+ */
+ 400: HTTPExceptionResponse;
+ /**
+ * Unauthorized
+ */
+ 401: HTTPExceptionResponse;
+ /**
+ * Forbidden
+ */
+ 403: HTTPExceptionResponse;
+ /**
+ * Not Found
+ */
+ 404: HTTPExceptionResponse;
+ /**
+ * Not Acceptable
+ */
+ 406: HTTPExceptionResponse;
+ /**
+ * Validation Error
+ */
+ 422: HTTPValidationError;
+ };
+ };
+ };
"/public/eventLogs/{event_log_id}": {
get: {
req: GetEventLogData;
diff --git a/airflow/www/static/js/types/api-generated.ts b/airflow/www/static/js/types/api-generated.ts
index 616f6fbfa29..7526c340b29 100644
--- a/airflow/www/static/js/types/api-generated.ts
+++ b/airflow/www/static/js/types/api-generated.ts
@@ -6,50 +6,6 @@ import type { CamelCasedPropertiesDeep } from "type-fest";
*/
export interface paths {
- "/backfills": {
- get: operations["list_backfills"];
- post: operations["create_backfill"];
- };
- "/backfills/{backfill_id}": {
- get: operations["get_backfill"];
- parameters: {
- path: {
- /** The integer id identifying the backfill entity. */
- backfill_id: components["parameters"]["BackfillIdPath"];
- };
- };
- };
- "/backfills/{backfill_id}/pause": {
- post: operations["pause_backfill"];
- parameters: {
- path: {
- /** The integer id identifying the backfill entity. */
- backfill_id: components["parameters"]["BackfillIdPath"];
- };
- };
- };
- "/backfills/{backfill_id}/unpause": {
- post: operations["unpause_backfill"];
- parameters: {
- path: {
- /** The integer id identifying the backfill entity. */
- backfill_id: components["parameters"]["BackfillIdPath"];
- };
- };
- };
- "/backfills/{backfill_id}/cancel": {
- /**
- * When a backfill is cancelled, all queued dag runs will be marked as failed.
- * Running dag runs will be allowed to continue.
- */
- post: operations["cancel_backfill"];
- parameters: {
- path: {
- /** The integer id identifying the backfill entity. */
- backfill_id: components["parameters"]["BackfillIdPath"];
- };
- };
- };
"/connections": {
get: operations["get_connections"];
post: operations["post_connection"];
@@ -2787,125 +2743,6 @@ export interface components {
}
export interface operations {
- list_backfills: {
- parameters: {
- query: {
- /** List backfills for this dag. */
- dag_id: string;
- };
- };
- responses: {
- /** Success. */
- 200: {
- content: {
- "application/json": components["schemas"]["BackfillCollection"];
- };
- };
- 401: components["responses"]["Unauthenticated"];
- 403: components["responses"]["PermissionDenied"];
- };
- };
- create_backfill: {
- responses: {
- /** Success. */
- 200: {
- content: {
- "application/json": components["schemas"]["Backfill"];
- };
- };
- 400: components["responses"]["BadRequest"];
- 401: components["responses"]["Unauthenticated"];
- 403: components["responses"]["PermissionDenied"];
- };
- requestBody: {
- content: {
- "application/json": components["schemas"]["Backfill"];
- };
- };
- };
- get_backfill: {
- parameters: {
- path: {
- /** The integer id identifying the backfill entity. */
- backfill_id: components["parameters"]["BackfillIdPath"];
- };
- };
- responses: {
- /** Success. */
- 200: {
- content: {
- "application/json": components["schemas"]["Backfill"];
- };
- };
- 401: components["responses"]["Unauthenticated"];
- 403: components["responses"]["PermissionDenied"];
- 404: components["responses"]["NotFound"];
- };
- };
- pause_backfill: {
- parameters: {
- path: {
- /** The integer id identifying the backfill entity. */
- backfill_id: components["parameters"]["BackfillIdPath"];
- };
- };
- responses: {
- /** Success. */
- 200: {
- content: {
- "application/json": components["schemas"]["Backfill"];
- };
- };
- 401: components["responses"]["Unauthenticated"];
- 403: components["responses"]["PermissionDenied"];
- 404: components["responses"]["NotFound"];
- 409: components["responses"]["Conflict"];
- };
- };
- unpause_backfill: {
- parameters: {
- path: {
- /** The integer id identifying the backfill entity. */
- backfill_id: components["parameters"]["BackfillIdPath"];
- };
- };
- responses: {
- /** Success. */
- 200: {
- content: {
- "application/json": components["schemas"]["Backfill"];
- };
- };
- 401: components["responses"]["Unauthenticated"];
- 403: components["responses"]["PermissionDenied"];
- 404: components["responses"]["NotFound"];
- 409: components["responses"]["Conflict"];
- };
- };
- /**
- * When a backfill is cancelled, all queued dag runs will be marked as failed.
- * Running dag runs will be allowed to continue.
- */
- cancel_backfill: {
- parameters: {
- path: {
- /** The integer id identifying the backfill entity. */
- backfill_id: components["parameters"]["BackfillIdPath"];
- };
- };
- responses: {
- /** Success. */
- 200: {
- content: {
- "application/json": components["schemas"]["Backfill"];
- };
- };
- 401: components["responses"]["Unauthenticated"];
- 403: components["responses"]["PermissionDenied"];
- 404: components["responses"]["NotFound"];
- 409: components["responses"]["Conflict"];
- };
- };
get_connections: {
parameters: {
query: {
@@ -5642,24 +5479,6 @@ export type HealthStatus = CamelCasedPropertiesDeep<
export type Operations = operations;
/* Types for operation variables */
-export type ListBackfillsVariables = CamelCasedPropertiesDeep<
- operations["list_backfills"]["parameters"]["query"]
->;
-export type CreateBackfillVariables = CamelCasedPropertiesDeep<
- operations["create_backfill"]["requestBody"]["content"]["application/json"]
->;
-export type GetBackfillVariables = CamelCasedPropertiesDeep<
- operations["get_backfill"]["parameters"]["path"]
->;
-export type PauseBackfillVariables = CamelCasedPropertiesDeep<
- operations["pause_backfill"]["parameters"]["path"]
->;
-export type UnpauseBackfillVariables = CamelCasedPropertiesDeep<
- operations["unpause_backfill"]["parameters"]["path"]
->;
-export type CancelBackfillVariables = CamelCasedPropertiesDeep<
- operations["cancel_backfill"]["parameters"]["path"]
->;
export type GetConnectionsVariables = CamelCasedPropertiesDeep<
operations["get_connections"]["parameters"]["query"]
>;
diff --git a/tests/api_connexion/endpoints/test_backfill_endpoint.py b/tests/api_connexion/endpoints/test_backfill_endpoint.py
deleted file mode 100644
index 14a5078c539..00000000000
--- a/tests/api_connexion/endpoints/test_backfill_endpoint.py
+++ /dev/null
@@ -1,442 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-from __future__ import annotations
-
-import os
-from datetime import datetime
-from unittest import mock
-
-import pendulum
-import pytest
-
-from airflow.models import DagBag, DagModel
-from airflow.models.backfill import Backfill, ReprocessBehavior
-from airflow.models.dag import DAG
-from airflow.operators.empty import EmptyOperator
-from airflow.utils import timezone
-from airflow.utils.session import provide_session
-
-from tests_common.test_utils.api_connexion_utils import create_user, delete_user
-from tests_common.test_utils.db import (
- clear_db_backfills,
- clear_db_dags,
- clear_db_runs,
- clear_db_serialized_dags,
-)
-
-pytestmark = [pytest.mark.db_test, pytest.mark.need_serialized_dag]
-
-
-DAG_ID = "test_dag"
-TASK_ID = "op1"
-DAG2_ID = "test_dag2"
-DAG3_ID = "test_dag3"
-UTC_JSON_REPR = "UTC" if pendulum.__version__.startswith("3") else "Timezone('UTC')"
-
-
-def _clean_db():
- clear_db_backfills()
- clear_db_runs()
- clear_db_dags()
- clear_db_serialized_dags()
-
-
-@pytest.fixture(autouse=True)
-def clean_db():
- _clean_db()
- yield
- _clean_db()
-
-
-@pytest.fixture(scope="module")
-def configured_app(minimal_app_for_api):
- app = minimal_app_for_api
-
- create_user(
- app,
- username="test",
- role_name="admin",
- )
- create_user(app, username="test_no_permissions", role_name=None)
-
- with DAG(
- DAG_ID,
- schedule=None,
- start_date=datetime(2020, 6, 15),
- doc_md="details",
- params={"foo": 1},
- tags=["example"],
- ) as dag:
- EmptyOperator(task_id=TASK_ID)
-
- with DAG(DAG2_ID, schedule=None, start_date=datetime(2020, 6, 15)) as dag2: # no doc_md
- EmptyOperator(task_id=TASK_ID)
-
- with DAG(DAG3_ID, schedule=None) as dag3: # DAG start_date set to None
- EmptyOperator(task_id=TASK_ID, start_date=datetime(2019, 6, 12))
-
- dag_bag = DagBag(os.devnull, include_examples=False)
- dag_bag.dags = {dag.dag_id: dag, dag2.dag_id: dag2, dag3.dag_id: dag3}
-
- app.dag_bag = dag_bag
-
- yield app
-
- delete_user(app, username="test")
- delete_user(app, username="test_no_permissions")
-
-
-class TestBackfillEndpoint:
- @pytest.fixture(autouse=True)
- def setup_attrs(self, configured_app) -> None:
- self.app = configured_app
- self.client = self.app.test_client() # type:ignore
- self.dag_id = DAG_ID
- self.dag2_id = DAG2_ID
- self.dag3_id = DAG3_ID
-
- @provide_session
- def _create_dag_models(self, *, count=1, dag_id_prefix="TEST_DAG", is_paused=False, session=None):
- dags = []
- for num in range(1, count + 1):
- dag_model = DagModel(
- dag_id=f"{dag_id_prefix}_{num}",
- fileloc=f"/tmp/dag_{num}.py",
- is_active=True,
- timetable_summary="0 0 * * *",
- is_paused=is_paused,
- )
- session.add(dag_model)
- dags.append(dag_model)
- return dags
-
- @provide_session
- def _create_deactivated_dag(self, session=None):
- dag_model = DagModel(
- dag_id="TEST_DAG_DELETED_1",
- fileloc="/tmp/dag_del_1.py",
- schedule_interval="2 2 * * *",
- is_active=False,
- )
- session.add(dag_model)
-
-
-class TestListBackfills(TestBackfillEndpoint):
- def test_should_respond_200(self, session):
- (dag,) = self._create_dag_models()
- from_date = timezone.utcnow()
- to_date = timezone.utcnow()
- b = Backfill(dag_id=dag.dag_id, from_date=from_date, to_date=to_date)
- session.add(b)
- session.commit()
- response = self.client.get(
- f"/api/v1/backfills?dag_id={dag.dag_id}",
- environ_overrides={"REMOTE_USER": "test"},
- )
- assert response.status_code == 200
- assert response.json == {
- "backfills": [
- {
- "completed_at": mock.ANY,
- "created_at": mock.ANY,
- "dag_id": "TEST_DAG_1",
- "dag_run_conf": None,
- "from_date": from_date.isoformat(),
- "id": b.id,
- "is_paused": False,
- "reprocess_behavior": "none",
- "max_active_runs": 10,
- "to_date": to_date.isoformat(),
- "updated_at": mock.ANY,
- }
- ],
- "total_entries": 1,
- }
-
- @pytest.mark.parametrize(
- "user, expected",
- [
- ("test_no_permissions", 403),
- ("test", 200),
- (None, 401),
- ],
- )
- def test_should_respond_200_with_granular_dag_access(self, user, expected, session):
- (dag,) = self._create_dag_models()
- from_date = timezone.utcnow()
- to_date = timezone.utcnow()
- b = Backfill(
- dag_id=dag.dag_id,
- from_date=from_date,
- to_date=to_date,
- )
-
- session.add(b)
- session.commit()
- kwargs = {}
- if user:
- kwargs.update(environ_overrides={"REMOTE_USER": user})
- response = self.client.get("/api/v1/backfills?dag_id=TEST_DAG_1", **kwargs)
- assert response.status_code == expected
-
-
-class TestGetBackfill(TestBackfillEndpoint):
- def test_should_respond_200(self, session):
- (dag,) = self._create_dag_models()
- from_date = timezone.utcnow()
- to_date = timezone.utcnow()
- backfill = Backfill(dag_id=dag.dag_id, from_date=from_date, to_date=to_date)
- session.add(backfill)
- session.commit()
- response = self.client.get(
- f"/api/v1/backfills/{backfill.id}",
- environ_overrides={"REMOTE_USER": "test"},
- )
- assert response.status_code == 200
- assert response.json == {
- "completed_at": mock.ANY,
- "created_at": mock.ANY,
- "dag_id": "TEST_DAG_1",
- "dag_run_conf": None,
- "from_date": from_date.isoformat(),
- "id": backfill.id,
- "is_paused": False,
- "reprocess_behavior": "none",
- "max_active_runs": 10,
- "to_date": to_date.isoformat(),
- "updated_at": mock.ANY,
- }
-
- def test_no_exist(self, session):
- response = self.client.get(
- f"/api/v1/backfills/{23198409834208}",
- environ_overrides={"REMOTE_USER": "test"},
- )
- assert response.status_code == 404
- assert response.json.get("title") == "Backfill not found"
-
- @pytest.mark.parametrize(
- "user, expected",
- [
- ("test_no_permissions", 403),
- ("test", 200),
- (None, 401),
- ],
- )
- def test_should_respond_200_with_granular_dag_access(self, user, expected, session):
- (dag,) = self._create_dag_models()
- from_date = timezone.utcnow()
- to_date = timezone.utcnow()
- backfill = Backfill(
- dag_id=dag.dag_id,
- from_date=from_date,
- to_date=to_date,
- )
- session.add(backfill)
- session.commit()
- kwargs = {}
- if user:
- kwargs.update(environ_overrides={"REMOTE_USER": user})
- response = self.client.get(f"/api/v1/backfills/{backfill.id}", **kwargs)
- assert response.status_code == expected
-
-
-class TestCreateBackfill(TestBackfillEndpoint):
- @pytest.mark.parametrize(
- "user, expected",
- [
- ("test_no_permissions", 403),
- ("test", 200),
- (None, 401),
- ],
- )
- @pytest.mark.parametrize(
- "repro_act, repro_exp",
- [
- (None, ReprocessBehavior.NONE),
- ("none", ReprocessBehavior.NONE),
- ("failed", ReprocessBehavior.FAILED),
- ("completed", ReprocessBehavior.COMPLETED),
- ],
- )
- def test_create_backfill(self, repro_act, repro_exp, user, expected, session, dag_maker):
- if repro_act is not None and expected > 300:
- pytest.skip("this combination not needed")
- with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
- EmptyOperator(task_id="mytask")
- session.query(DagModel).all()
- from_date = pendulum.parse("2024-01-01")
- from_date_iso = from_date.isoformat()
- to_date = pendulum.parse("2024-02-01")
- to_date_iso = to_date.isoformat()
- max_active_runs = 5
- data = {
- "dag_id": dag.dag_id,
- "from_date": f"{from_date_iso}",
- "to_date": f"{to_date_iso}",
- "max_active_runs": max_active_runs,
- "reverse": False,
- "dag_run_conf": {"param1": "val1", "param2": True},
- }
- if repro_act is not None:
- data["reprocess_behavior"] = repro_act
- kwargs = {}
- if user:
- kwargs.update(environ_overrides={"REMOTE_USER": user})
-
- response = self.client.post(
- "/api/v1/backfills",
- json=data,
- **kwargs,
- )
- assert response.status_code == expected
- if expected < 300:
- assert response.json == {
- "completed_at": mock.ANY,
- "created_at": mock.ANY,
- "dag_id": "TEST_DAG_1",
- "dag_run_conf": {"param1": "val1", "param2": True},
- "from_date": from_date_iso,
- "id": mock.ANY,
- "is_paused": False,
- "reprocess_behavior": repro_exp,
- "max_active_runs": 5,
- "to_date": to_date_iso,
- "updated_at": mock.ANY,
- }
-
-
-class TestPauseBackfill(TestBackfillEndpoint):
- def test_should_respond_200(self, session):
- (dag,) = self._create_dag_models()
- from_date = timezone.utcnow()
- to_date = timezone.utcnow()
- backfill = Backfill(dag_id=dag.dag_id, from_date=from_date, to_date=to_date)
- session.add(backfill)
- session.commit()
- response = self.client.post(
- f"/api/v1/backfills/{backfill.id}/pause",
- environ_overrides={"REMOTE_USER": "test"},
- )
- assert response.status_code == 200
- assert response.json == {
- "completed_at": mock.ANY,
- "created_at": mock.ANY,
- "dag_id": "TEST_DAG_1",
- "dag_run_conf": None,
- "from_date": from_date.isoformat(),
- "id": backfill.id,
- "is_paused": True,
- "reprocess_behavior": "none",
- "max_active_runs": 10,
- "to_date": to_date.isoformat(),
- "updated_at": mock.ANY,
- }
-
- @pytest.mark.parametrize(
- "user, expected",
- [
- ("test_no_permissions", 403),
- ("test", 200),
- (None, 401),
- ],
- )
- def test_should_respond_200_with_granular_dag_access(self, user, expected, session):
- (dag,) = self._create_dag_models()
- from_date = timezone.utcnow()
- to_date = timezone.utcnow()
- backfill = Backfill(
- dag_id=dag.dag_id,
- from_date=from_date,
- to_date=to_date,
- )
- session.add(backfill)
- session.commit()
- kwargs = {}
- if user:
- kwargs.update(environ_overrides={"REMOTE_USER": user})
- response = self.client.post(f"/api/v1/backfills/{backfill.id}/pause", **kwargs)
- assert response.status_code == expected
-
-
-class TestCancelBackfill(TestBackfillEndpoint):
- def test_should_respond_200(self, session):
- (dag,) = self._create_dag_models()
- from_date = timezone.utcnow()
- to_date = timezone.utcnow()
- backfill = Backfill(dag_id=dag.dag_id, from_date=from_date, to_date=to_date)
- session.add(backfill)
- session.commit()
- response = self.client.post(
- f"/api/v1/backfills/{backfill.id}/cancel",
- environ_overrides={"REMOTE_USER": "test"},
- )
- assert response.status_code == 200
- assert response.json == {
- "completed_at": mock.ANY,
- "created_at": mock.ANY,
- "dag_id": "TEST_DAG_1",
- "dag_run_conf": None,
- "from_date": from_date.isoformat(),
- "id": backfill.id,
- "is_paused": True,
- "reprocess_behavior": "none",
- "max_active_runs": 10,
- "to_date": to_date.isoformat(),
- "updated_at": mock.ANY,
- }
- assert pendulum.parse(response.json["completed_at"])
- # now it is marked as completed
- assert pendulum.parse(response.json["completed_at"])
-
- # get conflict when canceling already-canceled backfill
- response = self.client.post(
- f"/api/v1/backfills/{backfill.id}/cancel", environ_overrides={"REMOTE_USER": "test"}
- )
- assert response.status_code == 409
-
- @pytest.mark.parametrize(
- "user, expected",
- [
- ("test_no_permissions", 403),
- ("test", 200),
- (None, 401),
- ],
- )
- def test_should_respond_200_with_granular_dag_access(self, user, expected, session):
- (dag,) = self._create_dag_models()
- from_date = timezone.utcnow()
- to_date = timezone.utcnow()
- backfill = Backfill(
- dag_id=dag.dag_id,
- from_date=from_date,
- to_date=to_date,
- )
- session.add(backfill)
- session.commit()
- kwargs = {}
- if user:
- kwargs.update(environ_overrides={"REMOTE_USER": user})
- response = self.client.post(f"/api/v1/backfills/{backfill.id}/cancel", **kwargs)
- assert response.status_code == expected
- if response.status_code < 300:
- # now it is marked as completed
- assert pendulum.parse(response.json["completed_at"])
-
- # get conflict when canceling already-canceled backfill
- response = self.client.post(f"/api/v1/backfills/{backfill.id}/cancel", **kwargs)
- assert response.status_code == 409
diff --git a/tests/api_connexion/schemas/test_backfill_schema.py b/tests/api_connexion/schemas/test_backfill_schema.py
deleted file mode 100644
index b9c4e91a59f..00000000000
--- a/tests/api_connexion/schemas/test_backfill_schema.py
+++ /dev/null
@@ -1,55 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-from __future__ import annotations
-
-from airflow.api_connexion.schemas.backfill_schema import BackfillCollection, backfill_collection_schema
-from airflow.models.backfill import Backfill
-from airflow.utils import timezone
-
-
-class TestBackfillSchema:
- def test_serialize_direct(self):
- now = timezone.utcnow()
- now_iso = now.isoformat()
- b1 = Backfill(
- dag_id="hi",
- created_at=now,
- completed_at=now,
- from_date=now,
- to_date=now,
- updated_at=now,
- )
- bc = BackfillCollection(backfills=[b1], total_entries=1)
- out = backfill_collection_schema.dump(bc)
- assert out == {
- "backfills": [
- {
- "completed_at": now_iso,
- "created_at": now_iso,
- "dag_id": "hi",
- "dag_run_conf": None,
- "from_date": now_iso,
- "id": None,
- "is_paused": None,
- "reprocess_behavior": None,
- "max_active_runs": None,
- "to_date": now_iso,
- "updated_at": now_iso,
- }
- ],
- "total_entries": 1,
- }
diff --git a/tests/api_fastapi/core_api/routes/public/test_backfills.py b/tests/api_fastapi/core_api/routes/public/test_backfills.py
new file mode 100644
index 00000000000..1c64b10848f
--- /dev/null
+++ b/tests/api_fastapi/core_api/routes/public/test_backfills.py
@@ -0,0 +1,309 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import os
+from datetime import datetime
+from unittest import mock
+
+import pendulum
+import pytest
+from sqlalchemy import select
+
+from airflow.models import DagBag, DagModel, DagRun
+from airflow.models.backfill import Backfill, BackfillDagRun, ReprocessBehavior, _create_backfill
+from airflow.models.dag import DAG
+from airflow.operators.empty import EmptyOperator
+from airflow.providers.standard.operators.python import PythonOperator
+from airflow.utils import timezone
+from airflow.utils.session import provide_session
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils.db import (
+ clear_db_backfills,
+ clear_db_dags,
+ clear_db_runs,
+ clear_db_serialized_dags,
+)
+
+pytestmark = [pytest.mark.db_test, pytest.mark.need_serialized_dag]
+
+
+DAG_ID = "test_dag"
+TASK_ID = "op1"
+DAG2_ID = "test_dag2"
+DAG3_ID = "test_dag3"
+
+
+def _clean_db():
+ clear_db_backfills()
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
+
+@pytest.fixture(autouse=True)
+def clean_db():
+ _clean_db()
+ yield
+ _clean_db()
+
+
+def make_dags():
+ with DAG(
+ DAG_ID,
+ schedule=None,
+ start_date=datetime(2020, 6, 15),
+ doc_md="details",
+ params={"foo": 1},
+ tags=["example"],
+ ) as dag:
+ EmptyOperator(task_id=TASK_ID)
+
+ with DAG(DAG2_ID, schedule=None, start_date=datetime(2020, 6, 15)) as dag2: # no doc_md
+ EmptyOperator(task_id=TASK_ID)
+
+ with DAG(DAG3_ID, schedule=None) as dag3: # DAG start_date set to None
+ EmptyOperator(task_id=TASK_ID, start_date=datetime(2019, 6, 12))
+
+ dag_bag = DagBag(os.devnull, include_examples=False)
+ dag_bag.dags = {dag.dag_id: dag, dag2.dag_id: dag2, dag3.dag_id: dag3}
+
+
+def to_iso(val):
+ return pendulum.instance(val).to_iso8601_string()
+
+
+class TestBackfillEndpoint:
+ @provide_session
+ def _create_dag_models(self, *, count=1, dag_id_prefix="TEST_DAG", is_paused=False, session=None):
+ dags = []
+ for num in range(1, count + 1):
+ dag_model = DagModel(
+ dag_id=f"{dag_id_prefix}_{num}",
+ fileloc=f"/tmp/dag_{num}.py",
+ is_active=True,
+ timetable_summary="0 0 * * *",
+ is_paused=is_paused,
+ )
+ session.add(dag_model)
+ dags.append(dag_model)
+ return dags
+
+
+class TestListBackfills(TestBackfillEndpoint):
+ def test_list_backfill(self, test_client, session):
+ (dag,) = self._create_dag_models()
+ from_date = timezone.utcnow()
+ to_date = timezone.utcnow()
+ b = Backfill(dag_id=dag.dag_id, from_date=from_date, to_date=to_date)
+ session.add(b)
+ session.commit()
+ response = test_client.get(f"/public/backfills?dag_id={dag.dag_id}")
+ assert response.status_code == 200
+ assert response.json() == {
+ "backfills": [
+ {
+ "completed_at": mock.ANY,
+ "created_at": mock.ANY,
+ "dag_id": "TEST_DAG_1",
+ "dag_run_conf": {},
+ "from_date": to_iso(from_date),
+ "id": b.id,
+ "is_paused": False,
+ "reprocess_behavior": "none",
+ "max_active_runs": 10,
+ "to_date": to_iso(to_date),
+ "updated_at": mock.ANY,
+ }
+ ],
+ "total_entries": 1,
+ }
+
+
+class TestGetBackfill(TestBackfillEndpoint):
+ def test_get_backfill(self, session, test_client):
+ (dag,) = self._create_dag_models()
+ from_date = timezone.utcnow()
+ to_date = timezone.utcnow()
+ backfill = Backfill(dag_id=dag.dag_id, from_date=from_date, to_date=to_date)
+ session.add(backfill)
+ session.commit()
+ response = test_client.get(f"/public/backfills/{backfill.id}")
+ assert response.status_code == 200
+ assert response.json() == {
+ "completed_at": mock.ANY,
+ "created_at": mock.ANY,
+ "dag_id": "TEST_DAG_1",
+ "dag_run_conf": {},
+ "from_date": to_iso(from_date),
+ "id": backfill.id,
+ "is_paused": False,
+ "reprocess_behavior": "none",
+ "max_active_runs": 10,
+ "to_date": to_iso(to_date),
+ "updated_at": mock.ANY,
+ }
+
+ def test_no_exist(self, session, test_client):
+ response = test_client.get(f"/public/backfills/{231984098}")
+ assert response.status_code == 404
+ assert response.json().get("detail") == "Backfill not found"
+
+
+class TestCreateBackfill(TestBackfillEndpoint):
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp",
+ [
+ (None, ReprocessBehavior.NONE),
+ ("none", ReprocessBehavior.NONE),
+ ("failed", ReprocessBehavior.FAILED),
+ ("completed", ReprocessBehavior.COMPLETED),
+ ],
+ )
+ def test_create_backfill(self, repro_act, repro_exp, session, dag_maker, test_client):
+ with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask")
+ session.query(DagModel).all()
+ session.commit()
+ from_date = pendulum.parse("2024-01-01")
+ from_date_iso = to_iso(from_date)
+ to_date = pendulum.parse("2024-02-01")
+ to_date_iso = to_iso(to_date)
+ max_active_runs = 5
+ data = {
+ "dag_id": dag.dag_id,
+ "from_date": f"{from_date_iso}",
+ "to_date": f"{to_date_iso}",
+ "max_active_runs": max_active_runs,
+ "run_backwards": False,
+ "dag_run_conf": {"param1": "val1", "param2": True},
+ }
+ if repro_act is not None:
+ data["reprocess_behavior"] = repro_act
+ response = test_client.post(
+ url="/public/backfills",
+ json=data,
+ )
+ assert response.status_code == 200
+ assert response.json() == {
+ "completed_at": mock.ANY,
+ "created_at": mock.ANY,
+ "dag_id": "TEST_DAG_1",
+ "dag_run_conf": {"param1": "val1", "param2": True},
+ "from_date": from_date_iso,
+ "id": mock.ANY,
+ "is_paused": False,
+ "reprocess_behavior": repro_exp,
+ "max_active_runs": 5,
+ "to_date": to_date_iso,
+ "updated_at": mock.ANY,
+ }
+
+
+class TestCancelBackfill(TestBackfillEndpoint):
+ def test_cancel_backfill(self, session, test_client):
+ (dag,) = self._create_dag_models()
+ from_date = timezone.utcnow()
+ to_date = timezone.utcnow()
+ backfill = Backfill(dag_id=dag.dag_id, from_date=from_date, to_date=to_date)
+ session.add(backfill)
+ session.commit()
+ response = test_client.put(
+ f"/public/backfills/{backfill.id}/cancel",
+ )
+ assert response.status_code == 200
+ assert response.json() == {
+ "completed_at": mock.ANY,
+ "created_at": mock.ANY,
+ "dag_id": "TEST_DAG_1",
+ "dag_run_conf": {},
+ "from_date": to_iso(from_date),
+ "id": backfill.id,
+ "is_paused": True,
+ "reprocess_behavior": "none",
+ "max_active_runs": 10,
+ "to_date": to_iso(to_date),
+ "updated_at": mock.ANY,
+ }
+ assert pendulum.parse(response.json()["completed_at"])
+ # now it is marked as completed
+ assert pendulum.parse(response.json()["completed_at"])
+
+ # get conflict when canceling already-canceled backfill
+ response = test_client.put(f"/public/backfills/{backfill.id}/cancel")
+ assert response.status_code == 409
+
+ def test_cancel_backfill_end_states(self, dag_maker, session, test_client):
+ """
+ Queued runs should be marked *failed*.
+ Every other dag run should be left alone.
+ """
+ with dag_maker(schedule="@daily") as dag:
+ PythonOperator(task_id="hi", python_callable=print)
+ b = _create_backfill(
+ dag_id=dag.dag_id,
+ from_date=timezone.datetime(2021, 1, 1),
+ to_date=timezone.datetime(2021, 1, 5),
+ max_active_runs=2,
+ reverse=False,
+ dag_run_conf={},
+ )
+ query = (
+ select(DagRun)
+ .join(BackfillDagRun.dag_run)
+ .where(BackfillDagRun.backfill_id == b.id)
+ .order_by(BackfillDagRun.sort_ordinal)
+ )
+ dag_runs = session.scalars(query).all()
+ dates = [str(x.logical_date.date()) for x in dag_runs]
+ expected_dates = ["2021-01-01", "2021-01-02", "2021-01-03", "2021-01-04", "2021-01-05"]
+ assert dates == expected_dates
+ assert all(x.state == DagRunState.QUEUED for x in dag_runs)
+ dag_runs[0].state = "running"
+ session.commit()
+ response = test_client.put(f"/public/backfills/{b.id}/cancel")
+ assert response.status_code == 200
+ session.expunge_all()
+ dag_runs = session.scalars(query).all()
+ states = [x.state for x in dag_runs]
+ assert states == ["running", "failed", "failed", "failed", "failed"]
+
+
+class TestPauseBackfill(TestBackfillEndpoint):
+ def test_pause_backfill(self, session, test_client):
+ (dag,) = self._create_dag_models()
+ from_date = timezone.utcnow()
+ to_date = timezone.utcnow()
+ backfill = Backfill(dag_id=dag.dag_id, from_date=from_date, to_date=to_date)
+ session.add(backfill)
+ session.commit()
+ response = test_client.put(f"/public/backfills/{backfill.id}/pause")
+ assert response.status_code == 200
+ assert response.json() == {
+ "completed_at": mock.ANY,
+ "created_at": mock.ANY,
+ "dag_id": "TEST_DAG_1",
+ "dag_run_conf": {},
+ "from_date": to_iso(from_date),
+ "id": backfill.id,
+ "is_paused": True,
+ "reprocess_behavior": "none",
+ "max_active_runs": 10,
+ "to_date": to_iso(to_date),
+ "updated_at": mock.ANY,
+ }
diff --git a/tests/models/test_backfill.py b/tests/models/test_backfill.py
index a67590e91a5..a32db07808e 100644
--- a/tests/models/test_backfill.py
+++ b/tests/models/test_backfill.py
@@ -32,7 +32,6 @@ from airflow.models.backfill import (
BackfillDagRun,
BackfillDagRunExceptionReason,
ReprocessBehavior,
- _cancel_backfill,
_create_backfill,
)
from airflow.providers.standard.operators.python import PythonOperator
@@ -323,41 +322,6 @@ def test_active_dag_run(dag_maker, session):
)
-def test_cancel_backfill(dag_maker, session):
- """
- Queued runs should be marked *failed*.
- Every other dag run should be left alone.
- """
- with dag_maker(schedule="@daily") as dag:
- PythonOperator(task_id="hi", python_callable=print)
- b = _create_backfill(
- dag_id=dag.dag_id,
- from_date=timezone.datetime(2021, 1, 1),
- to_date=timezone.datetime(2021, 1, 5),
- max_active_runs=2,
- reverse=False,
- dag_run_conf={},
- )
- query = (
- select(DagRun)
- .join(BackfillDagRun.dag_run)
- .where(BackfillDagRun.backfill_id == b.id)
- .order_by(BackfillDagRun.sort_ordinal)
- )
- dag_runs = session.scalars(query).all()
- dates = [str(x.logical_date.date()) for x in dag_runs]
- expected_dates = ["2021-01-01", "2021-01-02", "2021-01-03", "2021-01-04", "2021-01-05"]
- assert dates == expected_dates
- assert all(x.state == DagRunState.QUEUED for x in dag_runs)
- dag_runs[0].state = "running"
- session.commit()
- _cancel_backfill(backfill_id=b.id)
- session.expunge_all()
- dag_runs = session.scalars(query).all()
- states = [x.state for x in dag_runs]
- assert states == ["running", "failed", "failed", "failed", "failed"]
-
-
def create_next_run(
*, is_backfill: bool, next_date: datetime, dag_id: str, dag_maker, reprocess=None, session: Session
):