You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2025/01/26 20:10:03 UTC
(airflow) branch main updated: Unify Data Models for Bulk Actions in Rest API (FastAPI) (#46033)
This is an automated email from the ASF dual-hosted git repository.
potiuk 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 1ef2b6c91d7 Unify Data Models for Bulk Actions in Rest API (FastAPI) (#46033)
1ef2b6c91d7 is described below
commit 1ef2b6c91d7c1bef4d8bc440b5efc80ec43cee17
Author: Bugra Ozturk <bu...@users.noreply.github.com>
AuthorDate: Sun Jan 26 21:09:57 2025 +0100
Unify Data Models for Bulk Actions in Rest API (FastAPI) (#46033)
* Unify datamodels and services for bulk endpoints
* Remove commented code pieces
---
airflow/api_fastapi/core_api/datamodels/common.py | 96 +-
.../api_fastapi/core_api/datamodels/connections.py | 80 --
airflow/api_fastapi/core_api/datamodels/pools.py | 90 +-
.../api_fastapi/core_api/datamodels/variables.py | 78 --
.../api_fastapi/core_api/openapi/v1-generated.yaml | 730 +++++++--------
.../core_api/routes/public/connections.py | 33 +-
.../api_fastapi/core_api/routes/public/pools.py | 34 +-
.../core_api/routes/public/variables.py | 30 +-
.../api_fastapi/core_api/services/public/common.py | 74 ++
.../core_api/services/public/connections.py | 250 +++---
.../api_fastapi/core_api/services/public/pools.py | 245 +++--
.../core_api/services/public/variables.py | 206 ++---
airflow/ui/openapi-gen/queries/queries.ts | 30 +-
airflow/ui/openapi-gen/requests/schemas.gen.ts | 991 +++++++++------------
airflow/ui/openapi-gen/requests/services.gen.ts | 6 +-
airflow/ui/openapi-gen/requests/types.gen.ts | 464 ++++------
.../src/pages/Variables/DeleteVariablesButton.tsx | 2 +-
.../ui/src/pages/Variables/ImportVariablesForm.tsx | 2 +-
pyproject.toml | 4 +
.../core_api/routes/public/test_connections.py | 26 +-
.../core_api/routes/public/test_pools.py | 84 +-
.../core_api/routes/public/test_variables.py | 82 +-
22 files changed, 1568 insertions(+), 2069 deletions(-)
diff --git a/airflow/api_fastapi/core_api/datamodels/common.py b/airflow/api_fastapi/core_api/datamodels/common.py
index 232c7215fb8..4af5356f2ed 100644
--- a/airflow/api_fastapi/core_api/datamodels/common.py
+++ b/airflow/api_fastapi/core_api/datamodels/common.py
@@ -23,11 +23,17 @@ Common Data Models for Airflow REST API.
from __future__ import annotations
import enum
+from typing import Annotated, Any, Generic, TypeVar, Union
-from airflow.api_fastapi.core_api.base import BaseModel
+from pydantic import Discriminator, Field, Tag
+from airflow.api_fastapi.core_api.base import BaseModel
# Common Bulk Data Models
+T = TypeVar("T")
+K = TypeVar("K")
+
+
class BulkAction(enum.Enum):
"""Bulk Action to be performed on the used model."""
@@ -45,14 +51,94 @@ class BulkActionOnExistence(enum.Enum):
class BulkActionNotOnExistence(enum.Enum):
- """Bulk Action to be taken if the entity does not exists."""
+ """Bulk Action to be taken if the entity does not exist."""
FAIL = "fail"
SKIP = "skip"
-# TODO: Unify All Bulk Operation Related Base Data Models
-class BulkBaseAction(BaseModel):
+class BulkBaseAction(BaseModel, Generic[T]):
"""Base class for bulk actions."""
- action: BulkAction
+ action: BulkAction = Field(..., description="The action to be performed on the entities.")
+
+
+class BulkCreateAction(BulkBaseAction[T]):
+ """Bulk Create entity serializer for request bodies."""
+
+ entities: list[T] = Field(..., description="A list of entities to be created.")
+ action_on_existence: BulkActionOnExistence = BulkActionOnExistence.FAIL
+
+
+class BulkUpdateAction(BulkBaseAction[T]):
+ """Bulk Update entity serializer for request bodies."""
+
+ entities: list[T] = Field(..., description="A list of entities to be updated.")
+ action_on_non_existence: BulkActionNotOnExistence = BulkActionNotOnExistence.FAIL
+
+
+class BulkDeleteAction(BulkBaseAction[T]):
+ """Bulk Delete entity serializer for request bodies."""
+
+ entities: list[str] = Field(..., description="A list of entity id/key to be deleted.")
+ action_on_non_existence: BulkActionNotOnExistence = BulkActionNotOnExistence.FAIL
+
+
+def _action_discriminator(action: Any) -> str:
+ return BulkAction(action["action"]).value
+
+
+class BulkBody(BaseModel, Generic[T]):
+ """Serializer for bulk entity operations."""
+
+ actions: list[
+ Annotated[
+ Union[
+ Annotated[BulkCreateAction[T], Tag(BulkAction.CREATE.value)],
+ Annotated[BulkUpdateAction[T], Tag(BulkAction.UPDATE.value)],
+ Annotated[BulkDeleteAction[T], Tag(BulkAction.DELETE.value)],
+ ],
+ Discriminator(_action_discriminator),
+ ]
+ ]
+
+
+class BulkActionResponse(BaseModel):
+ """
+ Serializer for individual bulk action responses.
+
+ Represents the outcome of a single bulk operation (create, update, or delete).
+ The response includes a list of successful keys and any errors encountered during the operation.
+ This structure helps users understand which key actions succeeded and which failed.
+ """
+
+ success: list[str] = Field(
+ default=[], description="A list of unique id/key representing successful operations."
+ )
+ errors: list[dict[str, Any]] = Field(
+ default=[],
+ description="A list of errors encountered during the operation, each containing details about the issue.",
+ )
+
+
+class BulkResponse(BaseModel):
+ """
+ Serializer for responses to bulk entity operations.
+
+ This represents the results of create, update, and delete actions performed on entity in bulk.
+ Each action (if requested) is represented as a field containing details about successful keys and any encountered errors.
+ Fields are populated in the response only if the respective action was part of the request, else are set None.
+ """
+
+ create: BulkActionResponse | None = Field(
+ default=None,
+ description="Details of the bulk create operation, including successful keys and errors.",
+ )
+ update: BulkActionResponse | None = Field(
+ default=None,
+ description="Details of the bulk update operation, including successful keys and errors.",
+ )
+ delete: BulkActionResponse | None = Field(
+ default=None,
+ description="Details of the bulk delete operation, including successful keys and errors.",
+ )
diff --git a/airflow/api_fastapi/core_api/datamodels/connections.py b/airflow/api_fastapi/core_api/datamodels/connections.py
index b6764fcff12..00d075bc4ac 100644
--- a/airflow/api_fastapi/core_api/datamodels/connections.py
+++ b/airflow/api_fastapi/core_api/datamodels/connections.py
@@ -18,18 +18,11 @@
from __future__ import annotations
import json
-from typing import Any
from pydantic import Field, field_validator
from pydantic_core.core_schema import ValidationInfo
from airflow.api_fastapi.core_api.base import BaseModel
-from airflow.api_fastapi.core_api.datamodels.common import (
- BulkAction,
- BulkActionNotOnExistence,
- BulkActionOnExistence,
- BulkBaseAction,
-)
from airflow.utils.log.secrets_masker import redact
@@ -95,76 +88,3 @@ class ConnectionBody(BaseModel):
port: int | None = Field(default=None)
password: str | None = Field(default=None)
extra: str | None = Field(default=None)
-
-
-class ConnectionBulkCreateAction(BulkBaseAction):
- """Bulk Create Variable serializer for request bodies."""
-
- action: BulkAction = BulkAction.CREATE
- connections: list[ConnectionBody] = Field(..., description="A list of connections to be created.")
- action_on_existence: BulkActionOnExistence = BulkActionOnExistence.FAIL
-
-
-class ConnectionBulkUpdateAction(BulkBaseAction):
- """Bulk Update Connection serializer for request bodies."""
-
- action: BulkAction = BulkAction.UPDATE
- connections: list[ConnectionBody] = Field(..., description="A list of connections to be updated.")
- action_on_non_existence: BulkActionNotOnExistence = BulkActionNotOnExistence.FAIL
-
-
-class ConnectionBulkDeleteAction(BulkBaseAction):
- """Bulk Delete Connection serializer for request bodies."""
-
- action: BulkAction = BulkAction.DELETE
- connection_ids: list[str] = Field(..., description="A list of connection IDs to be deleted.")
- action_on_non_existence: BulkActionNotOnExistence = BulkActionNotOnExistence.FAIL
-
-
-class ConnectionBulkBody(BaseModel):
- """Request body for bulk Connection operations (create, update, delete)."""
-
- actions: list[ConnectionBulkCreateAction | ConnectionBulkUpdateAction | ConnectionBulkDeleteAction] = (
- Field(..., description="A list of Connection actions to perform.")
- )
-
-
-class ConnectionBulkActionResponse(BaseModel):
- """
- Serializer for individual bulk action responses.
-
- Represents the outcome of a single bulk operation (create, update, or delete).
- The response includes a list of successful connection_ids and any errors encountered during the operation.
- This structure helps users understand which key actions succeeded and which failed.
- """
-
- success: list[str] = Field(
- default_factory=list, description="A list of connection_ids representing successful operations."
- )
- errors: list[dict[str, Any]] = Field(
- default_factory=list,
- description="A list of errors encountered during the operation, each containing details about the issue.",
- )
-
-
-class ConnectionBulkResponse(BaseModel):
- """
- Serializer for responses to bulk connection operations.
-
- This represents the results of create, update, and delete actions performed on connections in bulk.
- Each action (if requested) is represented as a field containing details about successful connection_ids and any encountered errors.
- Fields are populated in the response only if the respective action was part of the request, else are set None.
- """
-
- create: ConnectionBulkActionResponse | None = Field(
- default=None,
- description="Details of the bulk create operation, including successful connection_ids and errors.",
- )
- update: ConnectionBulkActionResponse | None = Field(
- default=None,
- description="Details of the bulk update operation, including successful connection_ids and errors.",
- )
- delete: ConnectionBulkActionResponse | None = Field(
- default=None,
- description="Details of the bulk delete operation, including successful connection_ids and errors.",
- )
diff --git a/airflow/api_fastapi/core_api/datamodels/pools.py b/airflow/api_fastapi/core_api/datamodels/pools.py
index 61dcf9f746b..0040c49a1ef 100644
--- a/airflow/api_fastapi/core_api/datamodels/pools.py
+++ b/airflow/api_fastapi/core_api/datamodels/pools.py
@@ -17,17 +17,11 @@
from __future__ import annotations
-from typing import Annotated, Any, Callable
+from typing import Annotated, Callable
from pydantic import BeforeValidator, ConfigDict, Field
from airflow.api_fastapi.core_api.base import BaseModel
-from airflow.api_fastapi.core_api.datamodels.common import (
- BulkAction,
- BulkActionNotOnExistence,
- BulkActionOnExistence,
- BulkBaseAction,
-)
def _call_function(function: Callable[[], int]) -> int:
@@ -77,89 +71,9 @@ class PoolPatchBody(BaseModel):
include_deferred: bool | None = None
-class PoolPostBody(BasePool):
+class PoolBody(BasePool):
"""Pool serializer for post bodies."""
pool: str = Field(alias="name", max_length=256)
description: str | None = None
include_deferred: bool = False
-
-
-class PoolPostBulkBody(BaseModel):
- """Pools serializer for post bodies."""
-
- pools: list[PoolPostBody]
- overwrite: bool | None = Field(default=False)
-
-
-class PoolBulkCreateAction(BulkBaseAction):
- """Bulk Create Pool serializer for request bodies."""
-
- action: BulkAction = BulkAction.CREATE
- pools: list[PoolPostBody] = Field(..., description="A list of pools to be created.")
- action_on_existence: BulkActionOnExistence = BulkActionOnExistence.FAIL
-
-
-class PoolBulkUpdateAction(BulkBaseAction):
- """Bulk Update Pool serializer for request bodies."""
-
- action: BulkAction = BulkAction.UPDATE
- pools: list[PoolPatchBody] = Field(..., description="A list of pools to be updated.")
- action_on_non_existence: BulkActionNotOnExistence = BulkActionNotOnExistence.FAIL
-
-
-class PoolBulkDeleteAction(BulkBaseAction):
- """Bulk Delete Pool serializer for request bodies."""
-
- action: BulkAction = BulkAction.DELETE
- pool_names: list[str] = Field(..., description="A list of pool names to be deleted.")
- action_on_non_existence: BulkActionNotOnExistence = BulkActionNotOnExistence.FAIL
-
-
-class PoolBulkBody(BaseModel):
- """Request body for bulk Pool operations (create, update, delete)."""
-
- actions: list[PoolBulkCreateAction | PoolBulkUpdateAction | PoolBulkDeleteAction] = Field(
- ..., description="A list of Pool actions to perform."
- )
-
-
-class PoolBulkActionResponse(BaseModel):
- """
- Serializer for individual bulk action responses.
-
- Represents the outcome of a single bulk operation (create, update, or delete).
- The response includes a list of successful pool names and any errors encountered during the operation.
- This structure helps users understand which key actions succeeded and which failed.
- """
-
- success: list[str] = Field(
- default_factory=list, description="A list of pool names representing successful operations."
- )
- errors: list[dict[str, Any]] = Field(
- default_factory=list,
- description="A list of errors encountered during the operation, each containing details about the issue.",
- )
-
-
-class PoolBulkResponse(BaseModel):
- """
- Serializer for responses to bulk pool operations.
-
- This represents the results of create, update, and delete actions performed on pools in bulk.
- Each action (if requested) is represented as a field containing details about successful pool names and any encountered errors.
- Fields are populated in the response only if the respective action was part of the request, else are set None.
- """
-
- create: PoolBulkActionResponse | None = Field(
- default=None,
- description="Details of the bulk create operation, including successful pool names and errors.",
- )
- update: PoolBulkActionResponse | None = Field(
- default=None,
- description="Details of the bulk update operation, including successful pool names and errors.",
- )
- delete: PoolBulkActionResponse | None = Field(
- default=None,
- description="Details of the bulk delete operation, including successful pool names and errors.",
- )
diff --git a/airflow/api_fastapi/core_api/datamodels/variables.py b/airflow/api_fastapi/core_api/datamodels/variables.py
index d63bf40ae79..8307809bc5f 100644
--- a/airflow/api_fastapi/core_api/datamodels/variables.py
+++ b/airflow/api_fastapi/core_api/datamodels/variables.py
@@ -18,17 +18,10 @@
from __future__ import annotations
import json
-from typing import Any
from pydantic import ConfigDict, Field, model_validator
from airflow.api_fastapi.core_api.base import BaseModel
-from airflow.api_fastapi.core_api.datamodels.common import (
- BulkAction,
- BulkActionNotOnExistence,
- BulkActionOnExistence,
- BulkBaseAction,
-)
from airflow.models.base import ID_LEN
from airflow.typing_compat import Self
from airflow.utils.log.secrets_masker import redact
@@ -80,74 +73,3 @@ class VariablesImportResponse(BaseModel):
created_variable_keys: list[str]
import_count: int
created_count: int
-
-
-class VariableBulkCreateAction(BulkBaseAction):
- """Bulk Create Variable serializer for request bodies."""
-
- action: BulkAction = BulkAction.CREATE
- variables: list[VariableBody] = Field(..., description="A list of variables to be created.")
- action_on_existence: BulkActionOnExistence = BulkActionOnExistence.FAIL
-
-
-class VariableBulkUpdateAction(BulkBaseAction):
- """Bulk Update Variable serializer for request bodies."""
-
- action: BulkAction = BulkAction.UPDATE
- variables: list[VariableBody] = Field(..., description="A list of variables to be updated.")
- action_on_non_existence: BulkActionNotOnExistence = BulkActionNotOnExistence.FAIL
-
-
-class VariableBulkDeleteAction(BulkBaseAction):
- """Bulk Delete Variable serializer for request bodies."""
-
- action: BulkAction = BulkAction.DELETE
- keys: list[str] = Field(..., description="A list of variable keys to be deleted.")
- action_on_non_existence: BulkActionNotOnExistence = BulkActionNotOnExistence.FAIL
-
-
-class VariableBulkBody(BaseModel):
- """Request body for bulk variable operations (create, update, delete)."""
-
- actions: list[VariableBulkCreateAction | VariableBulkUpdateAction | VariableBulkDeleteAction] = Field(
- ..., description="A list of variable actions to perform."
- )
-
-
-class VariableBulkActionResponse(BaseModel):
- """
- Serializer for individual bulk action responses.
-
- Represents the outcome of a single bulk operation (create, update, or delete).
- The response includes a list of successful keys and any errors encountered during the operation.
- This structure helps users understand which key actions succeeded and which failed.
- """
-
- success: list[str] = Field(default=[], description="A list of keys representing successful operations.")
- errors: list[dict[str, Any]] = Field(
- default=[],
- description="A list of errors encountered during the operation, each containing details about the issue.",
- )
-
-
-class VariableBulkResponse(BaseModel):
- """
- Serializer for responses to bulk variable operations.
-
- This represents the results of create, update, and delete actions performed on variables in bulk.
- Each action (if requested) is represented as a field containing details about successful keys and any encountered errors.
- Fields are populated in the response only if the respective action was part of the request, else are set None.
- """
-
- create: VariableBulkActionResponse | None = Field(
- default=None,
- description="Details of the bulk create operation, including successful keys and errors.",
- )
- update: VariableBulkActionResponse | None = Field(
- default=None,
- description="Details of the bulk update operation, including successful keys and errors.",
- )
- delete: VariableBulkActionResponse | None = Field(
- default=None,
- description="Details of the bulk delete operation, including successful keys and errors.",
- )
diff --git a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
index 6075d0c886e..b3a3fbc27f5 100644
--- a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
+++ b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
@@ -1808,14 +1808,14 @@ paths:
content:
application/json:
schema:
- $ref: '#/components/schemas/ConnectionBulkBody'
+ $ref: '#/components/schemas/BulkBody_ConnectionBody_'
responses:
'200':
description: Successful Response
content:
application/json:
schema:
- $ref: '#/components/schemas/ConnectionBulkResponse'
+ $ref: '#/components/schemas/BulkResponse'
'401':
content:
application/json:
@@ -4145,7 +4145,7 @@ paths:
content:
application/json:
schema:
- $ref: '#/components/schemas/PoolPostBody'
+ $ref: '#/components/schemas/PoolBody'
responses:
'201':
description: Successful Response
@@ -4188,14 +4188,14 @@ paths:
content:
application/json:
schema:
- $ref: '#/components/schemas/PoolBulkBody'
+ $ref: '#/components/schemas/BulkBody_PoolBody_'
responses:
'200':
description: Successful Response
content:
application/json:
schema:
- $ref: '#/components/schemas/PoolBulkResponse'
+ $ref: '#/components/schemas/BulkResponse'
'401':
content:
application/json:
@@ -6058,14 +6058,14 @@ paths:
content:
application/json:
schema:
- $ref: '#/components/schemas/VariableBulkBody'
+ $ref: '#/components/schemas/BulkBody_VariableBody_'
responses:
'200':
description: Successful Response
content:
application/json:
schema:
- $ref: '#/components/schemas/VariableBulkResponse'
+ $ref: '#/components/schemas/BulkResponse'
'401':
content:
application/json:
@@ -6691,7 +6691,7 @@ components:
- fail
- skip
title: BulkActionNotOnExistence
- description: Bulk Action to be taken if the entity does not exists.
+ description: Bulk Action to be taken if the entity does not exist.
BulkActionOnExistence:
type: string
enum:
@@ -6700,6 +6700,281 @@ components:
- overwrite
title: BulkActionOnExistence
description: Bulk Action to be taken if the entity already exists or not.
+ BulkActionResponse:
+ properties:
+ success:
+ items:
+ type: string
+ type: array
+ title: Success
+ description: A list of unique id/key representing successful operations.
+ default: []
+ errors:
+ items:
+ type: object
+ type: array
+ title: Errors
+ description: A list of errors encountered during the operation, each containing
+ details about the issue.
+ default: []
+ type: object
+ title: BulkActionResponse
+ description: 'Serializer for individual bulk action responses.
+
+
+ Represents the outcome of a single bulk operation (create, update, or delete).
+
+ The response includes a list of successful keys and any errors encountered
+ during the operation.
+
+ This structure helps users understand which key actions succeeded and which
+ failed.'
+ BulkBody_ConnectionBody_:
+ properties:
+ actions:
+ items:
+ oneOf:
+ - $ref: '#/components/schemas/BulkCreateAction_ConnectionBody_'
+ - $ref: '#/components/schemas/BulkUpdateAction_ConnectionBody_'
+ - $ref: '#/components/schemas/BulkDeleteAction_ConnectionBody_'
+ type: array
+ title: Actions
+ type: object
+ required:
+ - actions
+ title: BulkBody[ConnectionBody]
+ BulkBody_PoolBody_:
+ properties:
+ actions:
+ items:
+ oneOf:
+ - $ref: '#/components/schemas/BulkCreateAction_PoolBody_'
+ - $ref: '#/components/schemas/BulkUpdateAction_PoolBody_'
+ - $ref: '#/components/schemas/BulkDeleteAction_PoolBody_'
+ type: array
+ title: Actions
+ type: object
+ required:
+ - actions
+ title: BulkBody[PoolBody]
+ BulkBody_VariableBody_:
+ properties:
+ actions:
+ items:
+ oneOf:
+ - $ref: '#/components/schemas/BulkCreateAction_VariableBody_'
+ - $ref: '#/components/schemas/BulkUpdateAction_VariableBody_'
+ - $ref: '#/components/schemas/BulkDeleteAction_VariableBody_'
+ type: array
+ title: Actions
+ type: object
+ required:
+ - actions
+ title: BulkBody[VariableBody]
+ BulkCreateAction_ConnectionBody_:
+ properties:
+ action:
+ $ref: '#/components/schemas/BulkAction'
+ description: The action to be performed on the entities.
+ entities:
+ items:
+ $ref: '#/components/schemas/ConnectionBody'
+ type: array
+ title: Entities
+ description: A list of entities to be created.
+ action_on_existence:
+ $ref: '#/components/schemas/BulkActionOnExistence'
+ default: fail
+ type: object
+ required:
+ - action
+ - entities
+ title: BulkCreateAction[ConnectionBody]
+ BulkCreateAction_PoolBody_:
+ properties:
+ action:
+ $ref: '#/components/schemas/BulkAction'
+ description: The action to be performed on the entities.
+ entities:
+ items:
+ $ref: '#/components/schemas/PoolBody'
+ type: array
+ title: Entities
+ description: A list of entities to be created.
+ action_on_existence:
+ $ref: '#/components/schemas/BulkActionOnExistence'
+ default: fail
+ type: object
+ required:
+ - action
+ - entities
+ title: BulkCreateAction[PoolBody]
+ BulkCreateAction_VariableBody_:
+ properties:
+ action:
+ $ref: '#/components/schemas/BulkAction'
+ description: The action to be performed on the entities.
+ entities:
+ items:
+ $ref: '#/components/schemas/VariableBody'
+ type: array
+ title: Entities
+ description: A list of entities to be created.
+ action_on_existence:
+ $ref: '#/components/schemas/BulkActionOnExistence'
+ default: fail
+ type: object
+ required:
+ - action
+ - entities
+ title: BulkCreateAction[VariableBody]
+ BulkDeleteAction_ConnectionBody_:
+ properties:
+ action:
+ $ref: '#/components/schemas/BulkAction'
+ description: The action to be performed on the entities.
+ entities:
+ items:
+ type: string
+ type: array
+ title: Entities
+ description: A list of entity id/key to be deleted.
+ action_on_non_existence:
+ $ref: '#/components/schemas/BulkActionNotOnExistence'
+ default: fail
+ type: object
+ required:
+ - action
+ - entities
+ title: BulkDeleteAction[ConnectionBody]
+ BulkDeleteAction_PoolBody_:
+ properties:
+ action:
+ $ref: '#/components/schemas/BulkAction'
+ description: The action to be performed on the entities.
+ entities:
+ items:
+ type: string
+ type: array
+ title: Entities
+ description: A list of entity id/key to be deleted.
+ action_on_non_existence:
+ $ref: '#/components/schemas/BulkActionNotOnExistence'
+ default: fail
+ type: object
+ required:
+ - action
+ - entities
+ title: BulkDeleteAction[PoolBody]
+ BulkDeleteAction_VariableBody_:
+ properties:
+ action:
+ $ref: '#/components/schemas/BulkAction'
+ description: The action to be performed on the entities.
+ entities:
+ items:
+ type: string
+ type: array
+ title: Entities
+ description: A list of entity id/key to be deleted.
+ action_on_non_existence:
+ $ref: '#/components/schemas/BulkActionNotOnExistence'
+ default: fail
+ type: object
+ required:
+ - action
+ - entities
+ title: BulkDeleteAction[VariableBody]
+ BulkResponse:
+ properties:
+ create:
+ anyOf:
+ - $ref: '#/components/schemas/BulkActionResponse'
+ - type: 'null'
+ description: Details of the bulk create operation, including successful
+ keys and errors.
+ update:
+ anyOf:
+ - $ref: '#/components/schemas/BulkActionResponse'
+ - type: 'null'
+ description: Details of the bulk update operation, including successful
+ keys and errors.
+ delete:
+ anyOf:
+ - $ref: '#/components/schemas/BulkActionResponse'
+ - type: 'null'
+ description: Details of the bulk delete operation, including successful
+ keys and errors.
+ type: object
+ title: BulkResponse
+ description: 'Serializer for responses to bulk entity operations.
+
+
+ This represents the results of create, update, and delete actions performed
+ on entity in bulk.
+
+ Each action (if requested) is represented as a field containing details about
+ successful keys and any encountered errors.
+
+ Fields are populated in the response only if the respective action was part
+ of the request, else are set None.'
+ BulkUpdateAction_ConnectionBody_:
+ properties:
+ action:
+ $ref: '#/components/schemas/BulkAction'
+ description: The action to be performed on the entities.
+ entities:
+ items:
+ $ref: '#/components/schemas/ConnectionBody'
+ type: array
+ title: Entities
+ description: A list of entities to be updated.
+ action_on_non_existence:
+ $ref: '#/components/schemas/BulkActionNotOnExistence'
+ default: fail
+ type: object
+ required:
+ - action
+ - entities
+ title: BulkUpdateAction[ConnectionBody]
+ BulkUpdateAction_PoolBody_:
+ properties:
+ action:
+ $ref: '#/components/schemas/BulkAction'
+ description: The action to be performed on the entities.
+ entities:
+ items:
+ $ref: '#/components/schemas/PoolBody'
+ type: array
+ title: Entities
+ description: A list of entities to be updated.
+ action_on_non_existence:
+ $ref: '#/components/schemas/BulkActionNotOnExistence'
+ default: fail
+ type: object
+ required:
+ - action
+ - entities
+ title: BulkUpdateAction[PoolBody]
+ BulkUpdateAction_VariableBody_:
+ properties:
+ action:
+ $ref: '#/components/schemas/BulkAction'
+ description: The action to be performed on the entities.
+ entities:
+ items:
+ $ref: '#/components/schemas/VariableBody'
+ type: array
+ title: Entities
+ description: A list of entities to be updated.
+ action_on_non_existence:
+ $ref: '#/components/schemas/BulkActionNotOnExistence'
+ default: fail
+ type: object
+ required:
+ - action
+ - entities
+ title: BulkUpdateAction[VariableBody]
ClearTaskInstancesBody:
properties:
dry_run:
@@ -6947,139 +7222,6 @@ components:
- conn_type
title: ConnectionBody
description: Connection Serializer for requests body.
- ConnectionBulkActionResponse:
- properties:
- success:
- items:
- type: string
- type: array
- title: Success
- description: A list of connection_ids representing successful operations.
- errors:
- items:
- type: object
- type: array
- title: Errors
- description: A list of errors encountered during the operation, each containing
- details about the issue.
- type: object
- title: ConnectionBulkActionResponse
- description: 'Serializer for individual bulk action responses.
-
-
- Represents the outcome of a single bulk operation (create, update, or delete).
-
- The response includes a list of successful connection_ids and any errors encountered
- during the operation.
-
- This structure helps users understand which key actions succeeded and which
- failed.'
- ConnectionBulkBody:
- properties:
- actions:
- items:
- anyOf:
- - $ref: '#/components/schemas/ConnectionBulkCreateAction'
- - $ref: '#/components/schemas/ConnectionBulkUpdateAction'
- - $ref: '#/components/schemas/ConnectionBulkDeleteAction'
- type: array
- title: Actions
- description: A list of Connection actions to perform.
- type: object
- required:
- - actions
- title: ConnectionBulkBody
- description: Request body for bulk Connection operations (create, update, delete).
- ConnectionBulkCreateAction:
- properties:
- action:
- $ref: '#/components/schemas/BulkAction'
- default: create
- connections:
- items:
- $ref: '#/components/schemas/ConnectionBody'
- type: array
- title: Connections
- description: A list of connections to be created.
- action_on_existence:
- $ref: '#/components/schemas/BulkActionOnExistence'
- default: fail
- type: object
- required:
- - connections
- title: ConnectionBulkCreateAction
- description: Bulk Create Variable serializer for request bodies.
- ConnectionBulkDeleteAction:
- properties:
- action:
- $ref: '#/components/schemas/BulkAction'
- default: delete
- connection_ids:
- items:
- type: string
- type: array
- title: Connection Ids
- description: A list of connection IDs to be deleted.
- action_on_non_existence:
- $ref: '#/components/schemas/BulkActionNotOnExistence'
- default: fail
- type: object
- required:
- - connection_ids
- title: ConnectionBulkDeleteAction
- description: Bulk Delete Connection serializer for request bodies.
- ConnectionBulkResponse:
- properties:
- create:
- anyOf:
- - $ref: '#/components/schemas/ConnectionBulkActionResponse'
- - type: 'null'
- description: Details of the bulk create operation, including successful
- connection_ids and errors.
- update:
- anyOf:
- - $ref: '#/components/schemas/ConnectionBulkActionResponse'
- - type: 'null'
- description: Details of the bulk update operation, including successful
- connection_ids and errors.
- delete:
- anyOf:
- - $ref: '#/components/schemas/ConnectionBulkActionResponse'
- - type: 'null'
- description: Details of the bulk delete operation, including successful
- connection_ids and errors.
- type: object
- title: ConnectionBulkResponse
- description: 'Serializer for responses to bulk connection operations.
-
-
- This represents the results of create, update, and delete actions performed
- on connections in bulk.
-
- Each action (if requested) is represented as a field containing details about
- successful connection_ids and any encountered errors.
-
- Fields are populated in the response only if the respective action was part
- of the request, else are set None.'
- ConnectionBulkUpdateAction:
- properties:
- action:
- $ref: '#/components/schemas/BulkAction'
- default: update
- connections:
- items:
- $ref: '#/components/schemas/ConnectionBody'
- type: array
- title: Connections
- description: A list of connections to be updated.
- action_on_non_existence:
- $ref: '#/components/schemas/BulkActionNotOnExistence'
- default: fail
- type: object
- required:
- - connections
- title: ConnectionBulkUpdateAction
- description: Bulk Update Connection serializer for request bodies.
ConnectionCollectionResponse:
properties:
connections:
@@ -8900,139 +9042,30 @@ components:
- timetables
title: PluginResponse
description: Plugin serializer.
- PoolBulkActionResponse:
- properties:
- success:
- items:
- type: string
- type: array
- title: Success
- description: A list of pool names representing successful operations.
- errors:
- items:
- type: object
- type: array
- title: Errors
- description: A list of errors encountered during the operation, each containing
- details about the issue.
- type: object
- title: PoolBulkActionResponse
- description: 'Serializer for individual bulk action responses.
-
-
- Represents the outcome of a single bulk operation (create, update, or delete).
-
- The response includes a list of successful pool names and any errors encountered
- during the operation.
-
- This structure helps users understand which key actions succeeded and which
- failed.'
- PoolBulkBody:
+ PoolBody:
properties:
- actions:
- items:
- anyOf:
- - $ref: '#/components/schemas/PoolBulkCreateAction'
- - $ref: '#/components/schemas/PoolBulkUpdateAction'
- - $ref: '#/components/schemas/PoolBulkDeleteAction'
- type: array
- title: Actions
- description: A list of Pool actions to perform.
- type: object
- required:
- - actions
- title: PoolBulkBody
- description: Request body for bulk Pool operations (create, update, delete).
- PoolBulkCreateAction:
- properties:
- action:
- $ref: '#/components/schemas/BulkAction'
- default: create
- pools:
- items:
- $ref: '#/components/schemas/PoolPostBody'
- type: array
- title: Pools
- description: A list of pools to be created.
- action_on_existence:
- $ref: '#/components/schemas/BulkActionOnExistence'
- default: fail
- type: object
- required:
- - pools
- title: PoolBulkCreateAction
- description: Bulk Create Pool serializer for request bodies.
- PoolBulkDeleteAction:
- properties:
- action:
- $ref: '#/components/schemas/BulkAction'
- default: delete
- pool_names:
- items:
- type: string
- type: array
- title: Pool Names
- description: A list of pool names to be deleted.
- action_on_non_existence:
- $ref: '#/components/schemas/BulkActionNotOnExistence'
- default: fail
- type: object
- required:
- - pool_names
- title: PoolBulkDeleteAction
- description: Bulk Delete Pool serializer for request bodies.
- PoolBulkResponse:
- properties:
- create:
- anyOf:
- - $ref: '#/components/schemas/PoolBulkActionResponse'
- - type: 'null'
- description: Details of the bulk create operation, including successful
- pool names and errors.
- update:
- anyOf:
- - $ref: '#/components/schemas/PoolBulkActionResponse'
- - type: 'null'
- description: Details of the bulk update operation, including successful
- pool names and errors.
- delete:
+ name:
+ type: string
+ maxLength: 256
+ title: Name
+ slots:
+ type: integer
+ title: Slots
+ description:
anyOf:
- - $ref: '#/components/schemas/PoolBulkActionResponse'
+ - type: string
- type: 'null'
- description: Details of the bulk delete operation, including successful
- pool names and errors.
- type: object
- title: PoolBulkResponse
- description: 'Serializer for responses to bulk pool operations.
-
-
- This represents the results of create, update, and delete actions performed
- on pools in bulk.
-
- Each action (if requested) is represented as a field containing details about
- successful pool names and any encountered errors.
-
- Fields are populated in the response only if the respective action was part
- of the request, else are set None.'
- PoolBulkUpdateAction:
- properties:
- action:
- $ref: '#/components/schemas/BulkAction'
- default: update
- pools:
- items:
- $ref: '#/components/schemas/PoolPatchBody'
- type: array
- title: Pools
- description: A list of pools to be updated.
- action_on_non_existence:
- $ref: '#/components/schemas/BulkActionNotOnExistence'
- default: fail
+ title: Description
+ include_deferred:
+ type: boolean
+ title: Include Deferred
+ default: false
type: object
required:
- - pools
- title: PoolBulkUpdateAction
- description: Bulk Update Pool serializer for request bodies.
+ - name
+ - slots
+ title: PoolBody
+ description: Pool serializer for post bodies.
PoolCollectionResponse:
properties:
pools:
@@ -9074,30 +9107,6 @@ components:
type: object
title: PoolPatchBody
description: Pool serializer for patch bodies.
- PoolPostBody:
- properties:
- name:
- type: string
- maxLength: 256
- title: Name
- slots:
- type: integer
- title: Slots
- description:
- anyOf:
- - type: string
- - type: 'null'
- title: Description
- include_deferred:
- type: boolean
- title: Include Deferred
- default: false
- type: object
- required:
- - name
- - slots
- title: PoolPostBody
- description: Pool serializer for post bodies.
PoolResponse:
properties:
name:
@@ -10163,141 +10172,6 @@ components:
- value
title: VariableBody
description: Variable serializer for bodies.
- VariableBulkActionResponse:
- properties:
- success:
- items:
- type: string
- type: array
- title: Success
- description: A list of keys representing successful operations.
- default: []
- errors:
- items:
- type: object
- type: array
- title: Errors
- description: A list of errors encountered during the operation, each containing
- details about the issue.
- default: []
- type: object
- title: VariableBulkActionResponse
- description: 'Serializer for individual bulk action responses.
-
-
- Represents the outcome of a single bulk operation (create, update, or delete).
-
- The response includes a list of successful keys and any errors encountered
- during the operation.
-
- This structure helps users understand which key actions succeeded and which
- failed.'
- VariableBulkBody:
- properties:
- actions:
- items:
- anyOf:
- - $ref: '#/components/schemas/VariableBulkCreateAction'
- - $ref: '#/components/schemas/VariableBulkUpdateAction'
- - $ref: '#/components/schemas/VariableBulkDeleteAction'
- type: array
- title: Actions
- description: A list of variable actions to perform.
- type: object
- required:
- - actions
- title: VariableBulkBody
- description: Request body for bulk variable operations (create, update, delete).
- VariableBulkCreateAction:
- properties:
- action:
- $ref: '#/components/schemas/BulkAction'
- default: create
- variables:
- items:
- $ref: '#/components/schemas/VariableBody'
- type: array
- title: Variables
- description: A list of variables to be created.
- action_on_existence:
- $ref: '#/components/schemas/BulkActionOnExistence'
- default: fail
- type: object
- required:
- - variables
- title: VariableBulkCreateAction
- description: Bulk Create Variable serializer for request bodies.
- VariableBulkDeleteAction:
- properties:
- action:
- $ref: '#/components/schemas/BulkAction'
- default: delete
- keys:
- items:
- type: string
- type: array
- title: Keys
- description: A list of variable keys to be deleted.
- action_on_non_existence:
- $ref: '#/components/schemas/BulkActionNotOnExistence'
- default: fail
- type: object
- required:
- - keys
- title: VariableBulkDeleteAction
- description: Bulk Delete Variable serializer for request bodies.
- VariableBulkResponse:
- properties:
- create:
- anyOf:
- - $ref: '#/components/schemas/VariableBulkActionResponse'
- - type: 'null'
- description: Details of the bulk create operation, including successful
- keys and errors.
- update:
- anyOf:
- - $ref: '#/components/schemas/VariableBulkActionResponse'
- - type: 'null'
- description: Details of the bulk update operation, including successful
- keys and errors.
- delete:
- anyOf:
- - $ref: '#/components/schemas/VariableBulkActionResponse'
- - type: 'null'
- description: Details of the bulk delete operation, including successful
- keys and errors.
- type: object
- title: VariableBulkResponse
- description: 'Serializer for responses to bulk variable operations.
-
-
- This represents the results of create, update, and delete actions performed
- on variables in bulk.
-
- Each action (if requested) is represented as a field containing details about
- successful keys and any encountered errors.
-
- Fields are populated in the response only if the respective action was part
- of the request, else are set None.'
- VariableBulkUpdateAction:
- properties:
- action:
- $ref: '#/components/schemas/BulkAction'
- default: update
- variables:
- items:
- $ref: '#/components/schemas/VariableBody'
- type: array
- title: Variables
- description: A list of variables to be updated.
- action_on_non_existence:
- $ref: '#/components/schemas/BulkActionNotOnExistence'
- default: fail
- type: object
- required:
- - variables
- title: VariableBulkUpdateAction
- description: Bulk Update Variable serializer for request bodies.
VariableCollectionResponse:
properties:
variables:
diff --git a/airflow/api_fastapi/core_api/routes/public/connections.py b/airflow/api_fastapi/core_api/routes/public/connections.py
index 98a9bcf61dc..89a9b32d0a1 100644
--- a/airflow/api_fastapi/core_api/routes/public/connections.py
+++ b/airflow/api_fastapi/core_api/routes/public/connections.py
@@ -27,22 +27,18 @@ from sqlalchemy import select
from airflow.api_fastapi.common.db.common import SessionDep, 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.datamodels.common import BulkAction
+from airflow.api_fastapi.core_api.datamodels.common import (
+ BulkBody,
+ BulkResponse,
+)
from airflow.api_fastapi.core_api.datamodels.connections import (
ConnectionBody,
- ConnectionBulkActionResponse,
- ConnectionBulkBody,
- ConnectionBulkResponse,
ConnectionCollectionResponse,
ConnectionResponse,
ConnectionTestResponse,
)
from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
-from airflow.api_fastapi.core_api.services.public.connections import (
- handle_bulk_create,
- handle_bulk_delete,
- handle_bulk_update,
-)
+from airflow.api_fastapi.core_api.services.public.connections import BulkConnectionService
from airflow.configuration import conf
from airflow.models import Connection
from airflow.secrets.environment_variables import CONN_ENV_PREFIX
@@ -146,24 +142,11 @@ def post_connection(
@connections_router.patch("")
def bulk_connections(
- request: ConnectionBulkBody,
+ request: BulkBody[ConnectionBody],
session: SessionDep,
-) -> ConnectionBulkResponse:
+) -> BulkResponse:
"""Bulk create, update, and delete connections."""
- results: dict[str, ConnectionBulkActionResponse] = {}
-
- for action in request.actions:
- if action.action.value not in results:
- results[action.action.value] = ConnectionBulkActionResponse()
-
- if action.action == BulkAction.CREATE:
- handle_bulk_create(session, action, results[action.action.value]) # type: ignore
- elif action.action == BulkAction.UPDATE:
- handle_bulk_update(session, action, results[action.action.value]) # type: ignore
- elif action.action == BulkAction.DELETE:
- handle_bulk_delete(session, action, results[action.action.value]) # type: ignore
-
- return ConnectionBulkResponse(**results)
+ return BulkConnectionService(session=session, request=request).handle_request()
@connections_router.patch(
diff --git a/airflow/api_fastapi/core_api/routes/public/pools.py b/airflow/api_fastapi/core_api/routes/public/pools.py
index 7fc65f0f34c..f06deefc44f 100644
--- a/airflow/api_fastapi/core_api/routes/public/pools.py
+++ b/airflow/api_fastapi/core_api/routes/public/pools.py
@@ -26,23 +26,16 @@ from sqlalchemy import delete, select
from airflow.api_fastapi.common.db.common import SessionDep, 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.datamodels.common import BulkAction
+from airflow.api_fastapi.core_api.datamodels.common import BulkBody, BulkResponse
from airflow.api_fastapi.core_api.datamodels.pools import (
BasePool,
- PoolBulkActionResponse,
- PoolBulkBody,
- PoolBulkResponse,
+ PoolBody,
PoolCollectionResponse,
PoolPatchBody,
- PoolPostBody,
PoolResponse,
)
from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
-from airflow.api_fastapi.core_api.services.public.pools import (
- handle_bulk_create,
- handle_bulk_delete,
- handle_bulk_update,
-)
+from airflow.api_fastapi.core_api.services.public.pools import BulkPoolService
from airflow.models.pool import Pool
pools_router = AirflowRouter(tags=["Pool"], prefix="/pools")
@@ -175,7 +168,7 @@ def patch_pool(
), # handled by global exception handler
)
def post_pool(
- body: PoolPostBody,
+ body: PoolBody,
session: SessionDep,
) -> PoolResponse:
"""Create a Pool."""
@@ -186,21 +179,8 @@ def post_pool(
@pools_router.patch("")
def bulk_pools(
- request: PoolBulkBody,
+ request: BulkBody[PoolBody],
session: SessionDep,
-) -> PoolBulkResponse:
+) -> BulkResponse:
"""Bulk create, update, and delete pools."""
- results: dict[str, PoolBulkActionResponse] = {}
-
- for action in request.actions:
- if action.action.value not in results:
- results[action.action.value] = PoolBulkActionResponse()
-
- if action.action == BulkAction.CREATE:
- handle_bulk_create(session, action, results[action.action.value]) # type: ignore
- elif action.action == BulkAction.UPDATE:
- handle_bulk_update(session, action, results[action.action.value]) # type: ignore
- elif action.action == BulkAction.DELETE:
- handle_bulk_delete(session, action, results[action.action.value]) # type: ignore
-
- return PoolBulkResponse(**results)
+ return BulkPoolService(session=session, request=request).handle_request()
diff --git a/airflow/api_fastapi/core_api/routes/public/variables.py b/airflow/api_fastapi/core_api/routes/public/variables.py
index 48f34d98089..1dfd111c121 100644
--- a/airflow/api_fastapi/core_api/routes/public/variables.py
+++ b/airflow/api_fastapi/core_api/routes/public/variables.py
@@ -31,21 +31,14 @@ from airflow.api_fastapi.common.parameters import (
SortParam,
)
from airflow.api_fastapi.common.router import AirflowRouter
-from airflow.api_fastapi.core_api.datamodels.common import BulkAction
+from airflow.api_fastapi.core_api.datamodels.common import BulkBody, BulkResponse
from airflow.api_fastapi.core_api.datamodels.variables import (
VariableBody,
- VariableBulkActionResponse,
- VariableBulkBody,
- VariableBulkResponse,
VariableCollectionResponse,
VariableResponse,
)
from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
-from airflow.api_fastapi.core_api.services.public.variables import (
- handle_bulk_create,
- handle_bulk_delete,
- handle_bulk_update,
-)
+from airflow.api_fastapi.core_api.services.public.variables import BulkVariableService
from airflow.models.variable import Variable
variables_router = AirflowRouter(tags=["Variable"], prefix="/variables")
@@ -193,21 +186,8 @@ def post_variable(
@variables_router.patch("")
def bulk_variables(
- request: VariableBulkBody,
+ request: BulkBody[VariableBody],
session: SessionDep,
-) -> VariableBulkResponse:
+) -> BulkResponse:
"""Bulk create, update, and delete variables."""
- results: dict[str, VariableBulkActionResponse] = {}
-
- for action in request.actions:
- if action.action.value not in results:
- results[action.action.value] = VariableBulkActionResponse()
-
- if action.action == BulkAction.CREATE:
- handle_bulk_create(session, action, results[action.action.value]) # type: ignore
- elif action.action == BulkAction.UPDATE:
- handle_bulk_update(session, action, results[action.action.value]) # type: ignore
- elif action.action == BulkAction.DELETE:
- handle_bulk_delete(session, action, results[action.action.value]) # type: ignore
-
- return VariableBulkResponse(**results)
+ return BulkVariableService(session=session, request=request).handle_request()
diff --git a/airflow/api_fastapi/core_api/services/public/common.py b/airflow/api_fastapi/core_api/services/public/common.py
new file mode 100644
index 00000000000..dc1bd65e95f
--- /dev/null
+++ b/airflow/api_fastapi/core_api/services/public/common.py
@@ -0,0 +1,74 @@
+# 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 abc import ABC, abstractmethod
+from typing import Generic
+
+from sqlalchemy.orm import Session
+
+from airflow.api_fastapi.core_api.datamodels.common import (
+ BulkAction,
+ BulkActionResponse,
+ BulkBody,
+ BulkCreateAction,
+ BulkDeleteAction,
+ BulkResponse,
+ BulkUpdateAction,
+ T,
+)
+
+
+class BulkService(Generic[T], ABC):
+ """Base class for bulk service operations."""
+
+ def __init__(self, session: Session, request: BulkBody[T]):
+ self.session = session
+ self.request = request
+
+ def handle_request(self) -> BulkResponse:
+ """Handle request for bulk actions."""
+ results: dict[str, BulkActionResponse] = {}
+
+ for action in self.request.actions:
+ if action.action.value not in results:
+ results[action.action.value] = BulkActionResponse()
+
+ if action.action == BulkAction.CREATE:
+ self.handle_bulk_create(action, results[action.action.value]) # type: ignore
+ elif action.action == BulkAction.UPDATE:
+ self.handle_bulk_update(action, results[action.action.value]) # type: ignore
+ elif action.action == BulkAction.DELETE:
+ self.handle_bulk_delete(action, results[action.action.value]) # type: ignore
+
+ return BulkResponse(**results)
+
+ @abstractmethod
+ def handle_bulk_create(self, action: BulkCreateAction[T], results: BulkActionResponse) -> None:
+ """Bulk create entities."""
+ raise NotImplementedError
+
+ @abstractmethod
+ def handle_bulk_update(self, action: BulkUpdateAction[T], results: BulkActionResponse) -> None:
+ """Bulk update entities."""
+ raise NotImplementedError
+
+ @abstractmethod
+ def handle_bulk_delete(self, action: BulkDeleteAction[T], results: BulkActionResponse) -> None:
+ """Bulk delete entities."""
+ raise NotImplementedError
diff --git a/airflow/api_fastapi/core_api/services/public/connections.py b/airflow/api_fastapi/core_api/services/public/connections.py
index 16dfefc43a4..7aab305af5c 100644
--- a/airflow/api_fastapi/core_api/services/public/connections.py
+++ b/airflow/api_fastapi/core_api/services/public/connections.py
@@ -21,134 +21,134 @@ from fastapi import HTTPException, status
from pydantic import ValidationError
from sqlalchemy import select
-from airflow.api_fastapi.common.db.common import SessionDep
-from airflow.api_fastapi.core_api.datamodels.common import BulkActionNotOnExistence, BulkActionOnExistence
-from airflow.api_fastapi.core_api.datamodels.connections import (
- ConnectionBody,
- ConnectionBulkActionResponse,
- ConnectionBulkCreateAction,
- ConnectionBulkDeleteAction,
- ConnectionBulkUpdateAction,
+from airflow.api_fastapi.core_api.datamodels.common import (
+ BulkActionNotOnExistence,
+ BulkActionOnExistence,
+ BulkActionResponse,
+ BulkCreateAction,
+ BulkDeleteAction,
+ BulkUpdateAction,
)
+from airflow.api_fastapi.core_api.datamodels.connections import ConnectionBody
+from airflow.api_fastapi.core_api.services.public.common import BulkService
from airflow.models.connection import Connection
-def categorize_connections(session: SessionDep, connection_ids: set) -> tuple[dict, set, set]:
- """
- Categorize the given connection_ids into matched_connection_ids and not_found_connection_ids based on existing connection_ids.
-
- Existed connections are returned as a dict of {connection_id : Connection}.
-
- :param session: SQLAlchemy session
- :param connection_ids: set of connection_ids
- :return: tuple of dict of existed connections, set of matched connection_ids, set of not found connection_ids
- """
- existed_connections = session.execute(
- select(Connection).filter(Connection.conn_id.in_(connection_ids))
- ).scalars()
- existed_connections_dict = {conn.conn_id: conn for conn in existed_connections}
- matched_connection_ids = set(existed_connections_dict.keys())
- not_found_connection_ids = connection_ids - matched_connection_ids
- return existed_connections_dict, matched_connection_ids, not_found_connection_ids
-
-
-def handle_bulk_create(
- session: SessionDep, action: ConnectionBulkCreateAction, results: ConnectionBulkActionResponse
-) -> None:
- """Bulk create connections."""
- to_create_connection_ids = {connection.connection_id for connection in action.connections}
- existed_connections_dict, matched_connection_ids, not_found_connection_ids = categorize_connections(
- session, to_create_connection_ids
- )
- try:
- if action.action_on_existence == BulkActionOnExistence.FAIL and matched_connection_ids:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"The connections with these connection_ids: {matched_connection_ids} already exist.",
- )
- elif action.action_on_existence == BulkActionOnExistence.SKIP:
- create_connection_ids = not_found_connection_ids
- else:
- create_connection_ids = to_create_connection_ids
-
- for connection in action.connections:
- if connection.connection_id in create_connection_ids:
- if connection.connection_id in matched_connection_ids:
- existed_connection = existed_connections_dict[connection.connection_id]
+class BulkConnectionService(BulkService[ConnectionBody]):
+ """Service for handling bulk operations on connections."""
+
+ def categorize_connections(self, connection_ids: set) -> tuple[dict, set, set]:
+ """
+ Categorize the given connection_ids into matched_connection_ids and not_found_connection_ids based on existing connection_ids.
+
+ Existed connections are returned as a dict of {connection_id : Connection}.
+
+ :param connection_ids: set of connection_ids
+ :return: tuple of dict of existed connections, set of matched connection_ids, set of not found connection_ids
+ """
+ existed_connections = self.session.execute(
+ select(Connection).filter(Connection.conn_id.in_(connection_ids))
+ ).scalars()
+ existed_connections_dict = {conn.conn_id: conn for conn in existed_connections}
+ matched_connection_ids = set(existed_connections_dict.keys())
+ not_found_connection_ids = connection_ids - matched_connection_ids
+ return existed_connections_dict, matched_connection_ids, not_found_connection_ids
+
+ def handle_bulk_create(
+ self, action: BulkCreateAction[ConnectionBody], results: BulkActionResponse
+ ) -> None:
+ """Bulk create connections."""
+ to_create_connection_ids = {connection.connection_id for connection in action.entities}
+ existed_connections_dict, matched_connection_ids, not_found_connection_ids = (
+ self.categorize_connections(to_create_connection_ids)
+ )
+ try:
+ if action.action_on_existence == BulkActionOnExistence.FAIL and matched_connection_ids:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=f"The connections with these connection_ids: {matched_connection_ids} already exist.",
+ )
+ elif action.action_on_existence == BulkActionOnExistence.SKIP:
+ create_connection_ids = not_found_connection_ids
+ else:
+ create_connection_ids = to_create_connection_ids
+
+ for connection in action.entities:
+ if connection.connection_id in create_connection_ids:
+ if connection.connection_id in matched_connection_ids:
+ existed_connection = existed_connections_dict[connection.connection_id]
+ for key, val in connection.model_dump(by_alias=True).items():
+ setattr(existed_connection, key, val)
+ else:
+ self.session.add(Connection(**connection.model_dump(by_alias=True)))
+ results.success.append(connection.connection_id)
+
+ except HTTPException as e:
+ results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
+
+ def handle_bulk_update(
+ self, action: BulkUpdateAction[ConnectionBody], results: BulkActionResponse
+ ) -> None:
+ """Bulk Update connections."""
+ to_update_connection_ids = {connection.connection_id for connection in action.entities}
+ _, matched_connection_ids, not_found_connection_ids = self.categorize_connections(
+ to_update_connection_ids
+ )
+
+ try:
+ if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_connection_ids:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=f"The connections with these connection_ids: {not_found_connection_ids} were not found.",
+ )
+ elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+ update_connection_ids = matched_connection_ids
+ else:
+ update_connection_ids = to_update_connection_ids
+
+ for connection in action.entities:
+ if connection.connection_id in update_connection_ids:
+ old_connection = self.session.scalar(
+ select(Connection).filter(Connection.conn_id == connection.connection_id).limit(1)
+ )
+ ConnectionBody(**connection.model_dump())
for key, val in connection.model_dump(by_alias=True).items():
- setattr(existed_connection, key, val)
- else:
- session.add(Connection(**connection.model_dump(by_alias=True)))
- results.success.append(connection.connection_id)
-
- except HTTPException as e:
- results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
-
-
-def handle_bulk_update(
- session: SessionDep, action: ConnectionBulkUpdateAction, results: ConnectionBulkActionResponse
-) -> None:
- """Bulk Update connections."""
- to_update_connection_ids = {connection.connection_id for connection in action.connections}
- _, matched_connection_ids, not_found_connection_ids = categorize_connections(
- session, to_update_connection_ids
- )
-
- try:
- if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_connection_ids:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail=f"The connections with these connection_ids: {not_found_connection_ids} were not found.",
- )
- elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
- update_connection_ids = matched_connection_ids
- else:
- update_connection_ids = to_update_connection_ids
-
- for connection in action.connections:
- if connection.connection_id in update_connection_ids:
- old_connection = session.scalar(
- select(Connection).filter(Connection.conn_id == connection.connection_id).limit(1)
+ setattr(old_connection, key, val)
+ results.success.append(connection.connection_id)
+
+ except HTTPException as e:
+ results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
+
+ except ValidationError as e:
+ results.errors.append({"error": f"{e.errors()}"})
+
+ def handle_bulk_delete(
+ self, action: BulkDeleteAction[ConnectionBody], results: BulkActionResponse
+ ) -> None:
+ """Bulk delete connections."""
+ to_delete_connection_ids = set(action.entities)
+ _, matched_connection_ids, not_found_connection_ids = self.categorize_connections(
+ to_delete_connection_ids
+ )
+
+ try:
+ if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_connection_ids:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=f"The connections with these connection_ids: {not_found_connection_ids} were not found.",
)
- ConnectionBody(**connection.model_dump())
- for key, val in connection.model_dump(by_alias=True).items():
- setattr(old_connection, key, val)
- results.success.append(connection.connection_id)
-
- except HTTPException as e:
- results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
-
- except ValidationError as e:
- results.errors.append({"error": f"{e.errors()}"})
-
-
-def handle_bulk_delete(
- session: SessionDep, action: ConnectionBulkDeleteAction, results: ConnectionBulkActionResponse
-) -> None:
- """Bulk delete connections."""
- to_delete_connection_ids = set(action.connection_ids)
- _, matched_connection_ids, not_found_connection_ids = categorize_connections(
- session, to_delete_connection_ids
- )
-
- try:
- if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_connection_ids:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail=f"The connections with these connection_ids: {not_found_connection_ids} were not found.",
- )
- elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
- delete_connection_ids = matched_connection_ids
- else:
- delete_connection_ids = to_delete_connection_ids
-
- for connection_id in delete_connection_ids:
- existing_connection = session.scalar(
- select(Connection).where(Connection.conn_id == connection_id).limit(1)
- )
- if existing_connection:
- session.delete(existing_connection)
- results.success.append(connection_id)
-
- except HTTPException as e:
- results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
+ elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+ delete_connection_ids = matched_connection_ids
+ else:
+ delete_connection_ids = to_delete_connection_ids
+
+ for connection_id in delete_connection_ids:
+ existing_connection = self.session.scalar(
+ select(Connection).where(Connection.conn_id == connection_id).limit(1)
+ )
+ if existing_connection:
+ self.session.delete(existing_connection)
+ results.success.append(connection_id)
+
+ except HTTPException as e:
+ results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
diff --git a/airflow/api_fastapi/core_api/services/public/pools.py b/airflow/api_fastapi/core_api/services/public/pools.py
index 19f365ef264..4a78e12e4f4 100644
--- a/airflow/api_fastapi/core_api/services/public/pools.py
+++ b/airflow/api_fastapi/core_api/services/public/pools.py
@@ -21,132 +21,129 @@ from fastapi import HTTPException, status
from pydantic import ValidationError
from sqlalchemy import select
-from airflow.api_fastapi.common.db.common import SessionDep
-from airflow.api_fastapi.core_api.datamodels.common import BulkActionNotOnExistence, BulkActionOnExistence
+from airflow.api_fastapi.core_api.datamodels.common import (
+ BulkActionNotOnExistence,
+ BulkActionOnExistence,
+ BulkActionResponse,
+ BulkCreateAction,
+ BulkDeleteAction,
+ BulkUpdateAction,
+)
from airflow.api_fastapi.core_api.datamodels.pools import (
- PoolBulkActionResponse,
- PoolBulkCreateAction,
- PoolBulkDeleteAction,
- PoolBulkUpdateAction,
- PoolPatchBody,
+ PoolBody,
)
+from airflow.api_fastapi.core_api.services.public.common import BulkService
from airflow.models.pool import Pool
-def categorize_pools(session: SessionDep, pool_names: set) -> tuple[dict, set, set]:
- """
- Categorize the given pool_names into matched_pool_names and not_found_pool_names based on existing pool_names.
-
- Existing pools are returned as a dict of {pool_name : Pool}.
-
- :param session: SQLAlchemy session
- :param pool_names: set of pool_names
- :return: tuple of dict of existed pools, set of matched pool_names, set of not found pool_names
- """
- existed_pools = session.execute(select(Pool).filter(Pool.pool.in_(pool_names))).scalars()
- existing_pools_dict = {pool.pool: pool for pool in existed_pools}
- matched_pool_names = set(existing_pools_dict.keys())
- not_found_pool_names = pool_names - matched_pool_names
- return existing_pools_dict, matched_pool_names, not_found_pool_names
-
-
-def handle_bulk_create(
- session: SessionDep, action: PoolBulkCreateAction, results: PoolBulkActionResponse
-) -> None:
- """Bulk create pools."""
- to_create_pool_names = {pool.pool for pool in action.pools}
- existing_pools_dict, matched_pool_names, not_found_pool_names = categorize_pools(
- session, to_create_pool_names
- )
- try:
- if action.action_on_existence == BulkActionOnExistence.FAIL and matched_pool_names:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"The pools with these pool names: {matched_pool_names} already exist.",
- )
- elif action.action_on_existence == BulkActionOnExistence.SKIP:
- create_pool_names = not_found_pool_names
- else:
- create_pool_names = to_create_pool_names
-
- for pool in action.pools:
- if pool.pool in create_pool_names:
- if pool.pool in matched_pool_names:
- existed_pool = existing_pools_dict[pool.pool]
- for key, val in pool.model_dump().items():
- setattr(existed_pool, key, val)
- else:
- session.add(Pool(**pool.model_dump()))
- results.success.append(pool.pool)
-
- session.flush()
-
- except HTTPException as e:
- results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
-
-
-def handle_bulk_update(
- session: SessionDep, action: PoolBulkUpdateAction, results: PoolBulkActionResponse
-) -> None:
- """Bulk Update pools."""
- to_update_pool_names = {pool.name for pool in action.pools}
- _, matched_pool_names, not_found_pool_names = categorize_pools(session, to_update_pool_names)
-
- try:
- if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_pool_names:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail=f"The pools with these pool names: {not_found_pool_names} were not found.",
- )
- elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
- update_pool_names = matched_pool_names
- else:
- update_pool_names = to_update_pool_names
-
- for pool in action.pools:
- if pool.name in update_pool_names:
- old_pool = session.scalar(select(Pool).filter(Pool.pool == pool.name).limit(1))
-
- data = {key: val for key, val in pool.model_dump(by_alias=True).items() if val is not None}
-
- try:
- PoolPatchBody(**data)
-
- for key, val in data.items():
- setattr(old_pool, key, val)
-
- results.success.append(str(pool.name))
-
- except ValidationError as e:
- results.errors.append({"error": f"{e.errors()}"})
-
- except HTTPException as e:
- results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
-
-
-def handle_bulk_delete(
- session: SessionDep, action: PoolBulkDeleteAction, results: PoolBulkActionResponse
-) -> None:
- """Bulk delete pools."""
- to_delete_pool_names = set(action.pool_names)
- _, matched_pool_names, not_found_pool_names = categorize_pools(session, to_delete_pool_names)
-
- try:
- if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_pool_names:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail=f"The pools with these pool names: {not_found_pool_names} were not found.",
- )
- elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
- delete_pool_names = matched_pool_names
- else:
- delete_pool_names = to_delete_pool_names
-
- for pool_name in delete_pool_names:
- existing_pool = session.scalar(select(Pool).where(Pool.pool == pool_name).limit(1))
- if existing_pool:
- session.delete(existing_pool)
- results.success.append(pool_name)
-
- except HTTPException as e:
- results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
+class BulkPoolService(BulkService[PoolBody]):
+ """Service for handling bulk operations on pools."""
+
+ def categorize_pools(self, pool_names: set) -> tuple[dict, set, set]:
+ """
+ Categorize the given pool_names into matched_pool_names and not_found_pool_names based on existing pool_names.
+
+ Existing pools are returned as a dict of {pool_name : Pool}.
+
+ :param pool_names: set of pool_names
+ :return: tuple of dict of existed pools, set of matched pool_names, set of not found pool_names
+ """
+ existed_pools = self.session.execute(select(Pool).filter(Pool.pool.in_(pool_names))).scalars()
+ existing_pools_dict = {pool.pool: pool for pool in existed_pools}
+ matched_pool_names = set(existing_pools_dict.keys())
+ not_found_pool_names = pool_names - matched_pool_names
+ return existing_pools_dict, matched_pool_names, not_found_pool_names
+
+ def handle_bulk_create(self, action: BulkCreateAction[PoolBody], results: BulkActionResponse) -> None:
+ """Bulk create pools."""
+ to_create_pool_names = {pool.pool for pool in action.entities}
+ existing_pools_dict, matched_pool_names, not_found_pool_names = self.categorize_pools(
+ to_create_pool_names
+ )
+ try:
+ if action.action_on_existence == BulkActionOnExistence.FAIL and matched_pool_names:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=f"The pools with these pool names: {matched_pool_names} already exist.",
+ )
+ elif action.action_on_existence == BulkActionOnExistence.SKIP:
+ create_pool_names = not_found_pool_names
+ else:
+ create_pool_names = to_create_pool_names
+
+ for pool in action.entities:
+ if pool.pool in create_pool_names:
+ if pool.pool in matched_pool_names:
+ existed_pool = existing_pools_dict[pool.pool]
+ for key, val in pool.model_dump().items():
+ setattr(existed_pool, key, val)
+ else:
+ self.session.add(Pool(**pool.model_dump()))
+ results.success.append(pool.pool)
+
+ self.session.flush()
+
+ except HTTPException as e:
+ results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
+
+ def handle_bulk_update(self, action: BulkUpdateAction[PoolBody], results: BulkActionResponse) -> None:
+ """Bulk Update pools."""
+ to_update_pool_names = {pool.pool for pool in action.entities}
+ _, matched_pool_names, not_found_pool_names = self.categorize_pools(to_update_pool_names)
+
+ try:
+ if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_pool_names:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=f"The pools with these pool names: {not_found_pool_names} were not found.",
+ )
+ elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+ update_pool_names = matched_pool_names
+ else:
+ update_pool_names = to_update_pool_names
+
+ for pool in action.entities:
+ if pool.pool in update_pool_names:
+ old_pool = self.session.scalar(select(Pool).filter(Pool.pool == pool.pool).limit(1))
+
+ data = {
+ key: val for key, val in pool.model_dump(by_alias=True).items() if val is not None
+ }
+ try:
+ PoolBody(**data)
+
+ for key, val in data.items():
+ setattr(old_pool, key, val)
+
+ results.success.append(str(pool.pool))
+
+ except ValidationError as e:
+ results.errors.append({"error": f"{e.errors()}"})
+
+ except HTTPException as e:
+ results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
+
+ def handle_bulk_delete(self, action: BulkDeleteAction[PoolBody], results: BulkActionResponse) -> None:
+ """Bulk delete pools."""
+ to_delete_pool_names = set(action.entities)
+ _, matched_pool_names, not_found_pool_names = self.categorize_pools(to_delete_pool_names)
+
+ try:
+ if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_pool_names:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=f"The pools with these pool names: {not_found_pool_names} were not found.",
+ )
+ elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+ delete_pool_names = matched_pool_names
+ else:
+ delete_pool_names = to_delete_pool_names
+
+ for pool_name in delete_pool_names:
+ existing_pool = self.session.scalar(select(Pool).where(Pool.pool == pool_name).limit(1))
+ if existing_pool:
+ self.session.delete(existing_pool)
+ results.success.append(pool_name)
+
+ except HTTPException as e:
+ results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
diff --git a/airflow/api_fastapi/core_api/services/public/variables.py b/airflow/api_fastapi/core_api/services/public/variables.py
index 1ab501dd0b9..e602faa4e7d 100644
--- a/airflow/api_fastapi/core_api/services/public/variables.py
+++ b/airflow/api_fastapi/core_api/services/public/variables.py
@@ -21,113 +21,113 @@ from fastapi import HTTPException, status
from pydantic import ValidationError
from sqlalchemy import select
-from airflow.api_fastapi.common.db.common import SessionDep
-from airflow.api_fastapi.core_api.datamodels.common import BulkActionNotOnExistence, BulkActionOnExistence
+from airflow.api_fastapi.core_api.datamodels.common import (
+ BulkActionNotOnExistence,
+ BulkActionOnExistence,
+ BulkActionResponse,
+ BulkCreateAction,
+ BulkDeleteAction,
+ BulkUpdateAction,
+)
from airflow.api_fastapi.core_api.datamodels.variables import (
VariableBody,
- VariableBulkActionResponse,
- VariableBulkCreateAction,
- VariableBulkDeleteAction,
- VariableBulkUpdateAction,
)
+from airflow.api_fastapi.core_api.services.public.common import BulkService
from airflow.models.variable import Variable
-def categorize_keys(session: SessionDep, keys: set) -> tuple[set, set]:
- """Categorize the given keys into matched_keys and not_found_keys based on existing keys."""
- existing_keys = {variable for variable in session.execute(select(Variable.key)).scalars()}
- matched_keys = existing_keys & keys
- not_found_keys = keys - existing_keys
- return matched_keys, not_found_keys
-
-
-def handle_bulk_create(
- session: SessionDep, action: VariableBulkCreateAction, results: VariableBulkActionResponse
-) -> None:
- """Bulk create variables."""
- to_create_keys = {variable.key for variable in action.variables}
- matched_keys, not_found_keys = categorize_keys(session, to_create_keys)
-
- try:
- if action.action_on_existence == BulkActionOnExistence.FAIL and matched_keys:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"The variables with these keys: {matched_keys} already exist.",
- )
- elif action.action_on_existence == BulkActionOnExistence.SKIP:
- create_keys = not_found_keys
- else:
- create_keys = to_create_keys
-
- for variable in action.variables:
- if variable.key in create_keys:
- Variable.set(
- key=variable.key, value=variable.value, description=variable.description, session=session
+class BulkVariableService(BulkService[VariableBody]):
+ """Service for handling bulk operations on variables."""
+
+ def categorize_keys(self, keys: set) -> tuple[set, set]:
+ """Categorize the given keys into matched_keys and not_found_keys based on existing keys."""
+ existing_keys = {variable for variable in self.session.execute(select(Variable.key)).scalars()}
+ matched_keys = existing_keys & keys
+ not_found_keys = keys - existing_keys
+ return matched_keys, not_found_keys
+
+ def handle_bulk_create(self, action: BulkCreateAction, results: BulkActionResponse) -> None:
+ """Bulk create variables."""
+ to_create_keys = {variable.key for variable in action.entities}
+ matched_keys, not_found_keys = self.categorize_keys(to_create_keys)
+
+ try:
+ if action.action_on_existence == BulkActionOnExistence.FAIL and matched_keys:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=f"The variables with these keys: {matched_keys} already exist.",
+ )
+ elif action.action_on_existence == BulkActionOnExistence.SKIP:
+ create_keys = not_found_keys
+ else:
+ create_keys = to_create_keys
+
+ for variable in action.entities:
+ if variable.key in create_keys:
+ Variable.set(
+ key=variable.key,
+ value=variable.value,
+ description=variable.description,
+ session=self.session,
+ )
+ results.success.append(variable.key)
+
+ except HTTPException as e:
+ results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
+
+ def handle_bulk_update(self, action: BulkUpdateAction, results: BulkActionResponse) -> None:
+ """Bulk Update variables."""
+ to_update_keys = {variable.key for variable in action.entities}
+ matched_keys, not_found_keys = self.categorize_keys(to_update_keys)
+
+ try:
+ if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_keys:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=f"The variables with these keys: {not_found_keys} were not found.",
+ )
+ elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+ update_keys = matched_keys
+ else:
+ update_keys = to_update_keys
+
+ for variable in action.entities:
+ if variable.key in update_keys:
+ old_variable = self.session.scalar(select(Variable).filter_by(key=variable.key).limit(1))
+ VariableBody(**variable.model_dump())
+ data = variable.model_dump(exclude={"key"}, by_alias=True)
+
+ for key, val in data.items():
+ setattr(old_variable, key, val)
+ results.success.append(variable.key)
+
+ except HTTPException as e:
+ results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
+
+ except ValidationError as e:
+ results.errors.append({"error": f"{e.errors()}"})
+
+ def handle_bulk_delete(self, action: BulkDeleteAction, results: BulkActionResponse) -> None:
+ """Bulk delete variables."""
+ to_delete_keys = set(action.entities)
+ matched_keys, not_found_keys = self.categorize_keys(to_delete_keys)
+
+ try:
+ if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_keys:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail=f"The variables with these keys: {not_found_keys} were not found.",
)
- results.success.append(variable.key)
-
- except HTTPException as e:
- results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
-
-
-def handle_bulk_update(
- session: SessionDep, action: VariableBulkUpdateAction, results: VariableBulkActionResponse
-) -> None:
- """Bulk Update variables."""
- to_update_keys = {variable.key for variable in action.variables}
- matched_keys, not_found_keys = categorize_keys(session, to_update_keys)
-
- try:
- if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_keys:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail=f"The variables with these keys: {not_found_keys} were not found.",
- )
- elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
- update_keys = matched_keys
- else:
- update_keys = to_update_keys
-
- for variable in action.variables:
- if variable.key in update_keys:
- old_variable = session.scalar(select(Variable).filter_by(key=variable.key).limit(1))
- VariableBody(**variable.model_dump())
- data = variable.model_dump(exclude={"key"}, by_alias=True)
-
- for key, val in data.items():
- setattr(old_variable, key, val)
- results.success.append(variable.key)
-
- except HTTPException as e:
- results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
-
- except ValidationError as e:
- results.errors.append({"error": f"{e.errors()}"})
-
-
-def handle_bulk_delete(
- session: SessionDep, action: VariableBulkDeleteAction, results: VariableBulkActionResponse
-) -> None:
- """Bulk delete variables."""
- to_delete_keys = set(action.keys)
- matched_keys, not_found_keys = categorize_keys(session, to_delete_keys)
-
- try:
- if action.action_on_non_existence == BulkActionNotOnExistence.FAIL and not_found_keys:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail=f"The variables with these keys: {not_found_keys} were not found.",
- )
- elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
- delete_keys = matched_keys
- else:
- delete_keys = to_delete_keys
-
- for key in delete_keys:
- existing_variable = session.scalar(select(Variable).where(Variable.key == key).limit(1))
- if existing_variable:
- session.delete(existing_variable)
- results.success.append(key)
-
- except HTTPException as e:
- results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
+ elif action.action_on_non_existence == BulkActionNotOnExistence.SKIP:
+ delete_keys = matched_keys
+ else:
+ delete_keys = to_delete_keys
+
+ for key in delete_keys:
+ existing_variable = self.session.scalar(select(Variable).where(Variable.key == key).limit(1))
+ if existing_variable:
+ self.session.delete(existing_variable)
+ results.success.append(key)
+
+ except HTTPException as e:
+ results.errors.append({"error": f"{e.detail}", "status_code": e.status_code})
diff --git a/airflow/ui/openapi-gen/queries/queries.ts b/airflow/ui/openapi-gen/queries/queries.ts
index 5245d8a112c..af43979e0ea 100644
--- a/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow/ui/openapi-gen/queries/queries.ts
@@ -33,9 +33,11 @@ import {
} from "../requests/services.gen";
import {
BackfillPostBody,
+ BulkBody_ConnectionBody_,
+ BulkBody_PoolBody_,
+ BulkBody_VariableBody_,
ClearTaskInstancesBody,
ConnectionBody,
- ConnectionBulkBody,
CreateAssetEventsBody,
DAGPatchBody,
DAGRunClearBody,
@@ -44,13 +46,11 @@ import {
DagRunState,
DagWarningType,
PatchTaskInstanceBody,
- PoolBulkBody,
+ PoolBody,
PoolPatchBody,
- PoolPostBody,
TaskInstancesBatchBody,
TriggerDAGRunPostBody,
VariableBody,
- VariableBulkBody,
} from "../requests/types.gen";
import * as Common from "./common";
@@ -3130,7 +3130,7 @@ export const usePoolServicePostPool = <
TData,
TError,
{
- requestBody: PoolPostBody;
+ requestBody: PoolBody;
},
TContext
>,
@@ -3141,7 +3141,7 @@ export const usePoolServicePostPool = <
TData,
TError,
{
- requestBody: PoolPostBody;
+ requestBody: PoolBody;
},
TContext
>({
@@ -3382,7 +3382,7 @@ export const useConnectionServicePatchConnection = <
* Bulk create, update, and delete connections.
* @param data The data for the request.
* @param data.requestBody
- * @returns ConnectionBulkResponse Successful Response
+ * @returns BulkResponse Successful Response
* @throws ApiError
*/
export const useConnectionServiceBulkConnections = <
@@ -3395,7 +3395,7 @@ export const useConnectionServiceBulkConnections = <
TData,
TError,
{
- requestBody: ConnectionBulkBody;
+ requestBody: BulkBody_ConnectionBody_;
},
TContext
>,
@@ -3406,7 +3406,7 @@ export const useConnectionServiceBulkConnections = <
TData,
TError,
{
- requestBody: ConnectionBulkBody;
+ requestBody: BulkBody_ConnectionBody_;
},
TContext
>({
@@ -3760,7 +3760,7 @@ export const usePoolServicePatchPool = <
* Bulk create, update, and delete pools.
* @param data The data for the request.
* @param data.requestBody
- * @returns PoolBulkResponse Successful Response
+ * @returns BulkResponse Successful Response
* @throws ApiError
*/
export const usePoolServiceBulkPools = <
@@ -3773,7 +3773,7 @@ export const usePoolServiceBulkPools = <
TData,
TError,
{
- requestBody: PoolBulkBody;
+ requestBody: BulkBody_PoolBody_;
},
TContext
>,
@@ -3784,7 +3784,7 @@ export const usePoolServiceBulkPools = <
TData,
TError,
{
- requestBody: PoolBulkBody;
+ requestBody: BulkBody_PoolBody_;
},
TContext
>({
@@ -3839,7 +3839,7 @@ export const useVariableServicePatchVariable = <
* Bulk create, update, and delete variables.
* @param data The data for the request.
* @param data.requestBody
- * @returns VariableBulkResponse Successful Response
+ * @returns BulkResponse Successful Response
* @throws ApiError
*/
export const useVariableServiceBulkVariables = <
@@ -3852,7 +3852,7 @@ export const useVariableServiceBulkVariables = <
TData,
TError,
{
- requestBody: VariableBulkBody;
+ requestBody: BulkBody_VariableBody_;
},
TContext
>,
@@ -3863,7 +3863,7 @@ export const useVariableServiceBulkVariables = <
TData,
TError,
{
- requestBody: VariableBulkBody;
+ requestBody: BulkBody_VariableBody_;
},
TContext
>({
diff --git a/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 1873dbbd9e2..4ff859b5c59 100644
--- a/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -526,7 +526,7 @@ export const $BulkActionNotOnExistence = {
type: "string",
enum: ["fail", "skip"],
title: "BulkActionNotOnExistence",
- description: "Bulk Action to be taken if the entity does not exists.",
+ description: "Bulk Action to be taken if the entity does not exist.",
} as const;
export const $BulkActionOnExistence = {
@@ -536,6 +536,373 @@ export const $BulkActionOnExistence = {
description: "Bulk Action to be taken if the entity already exists or not.",
} as const;
+export const $BulkActionResponse = {
+ properties: {
+ success: {
+ items: {
+ type: "string",
+ },
+ type: "array",
+ title: "Success",
+ description: "A list of unique id/key representing successful operations.",
+ default: [],
+ },
+ errors: {
+ items: {
+ type: "object",
+ },
+ type: "array",
+ title: "Errors",
+ description:
+ "A list of errors encountered during the operation, each containing details about the issue.",
+ default: [],
+ },
+ },
+ type: "object",
+ title: "BulkActionResponse",
+ description: `Serializer for individual bulk action responses.
+
+Represents the outcome of a single bulk operation (create, update, or delete).
+The response includes a list of successful keys and any errors encountered during the operation.
+This structure helps users understand which key actions succeeded and which failed.`,
+} as const;
+
+export const $BulkBody_ConnectionBody_ = {
+ properties: {
+ actions: {
+ items: {
+ oneOf: [
+ {
+ $ref: "#/components/schemas/BulkCreateAction_ConnectionBody_",
+ },
+ {
+ $ref: "#/components/schemas/BulkUpdateAction_ConnectionBody_",
+ },
+ {
+ $ref: "#/components/schemas/BulkDeleteAction_ConnectionBody_",
+ },
+ ],
+ },
+ type: "array",
+ title: "Actions",
+ },
+ },
+ type: "object",
+ required: ["actions"],
+ title: "BulkBody[ConnectionBody]",
+} as const;
+
+export const $BulkBody_PoolBody_ = {
+ properties: {
+ actions: {
+ items: {
+ oneOf: [
+ {
+ $ref: "#/components/schemas/BulkCreateAction_PoolBody_",
+ },
+ {
+ $ref: "#/components/schemas/BulkUpdateAction_PoolBody_",
+ },
+ {
+ $ref: "#/components/schemas/BulkDeleteAction_PoolBody_",
+ },
+ ],
+ },
+ type: "array",
+ title: "Actions",
+ },
+ },
+ type: "object",
+ required: ["actions"],
+ title: "BulkBody[PoolBody]",
+} as const;
+
+export const $BulkBody_VariableBody_ = {
+ properties: {
+ actions: {
+ items: {
+ oneOf: [
+ {
+ $ref: "#/components/schemas/BulkCreateAction_VariableBody_",
+ },
+ {
+ $ref: "#/components/schemas/BulkUpdateAction_VariableBody_",
+ },
+ {
+ $ref: "#/components/schemas/BulkDeleteAction_VariableBody_",
+ },
+ ],
+ },
+ type: "array",
+ title: "Actions",
+ },
+ },
+ type: "object",
+ required: ["actions"],
+ title: "BulkBody[VariableBody]",
+} as const;
+
+export const $BulkCreateAction_ConnectionBody_ = {
+ properties: {
+ action: {
+ $ref: "#/components/schemas/BulkAction",
+ description: "The action to be performed on the entities.",
+ },
+ entities: {
+ items: {
+ $ref: "#/components/schemas/ConnectionBody",
+ },
+ type: "array",
+ title: "Entities",
+ description: "A list of entities to be created.",
+ },
+ action_on_existence: {
+ $ref: "#/components/schemas/BulkActionOnExistence",
+ default: "fail",
+ },
+ },
+ type: "object",
+ required: ["action", "entities"],
+ title: "BulkCreateAction[ConnectionBody]",
+} as const;
+
+export const $BulkCreateAction_PoolBody_ = {
+ properties: {
+ action: {
+ $ref: "#/components/schemas/BulkAction",
+ description: "The action to be performed on the entities.",
+ },
+ entities: {
+ items: {
+ $ref: "#/components/schemas/PoolBody",
+ },
+ type: "array",
+ title: "Entities",
+ description: "A list of entities to be created.",
+ },
+ action_on_existence: {
+ $ref: "#/components/schemas/BulkActionOnExistence",
+ default: "fail",
+ },
+ },
+ type: "object",
+ required: ["action", "entities"],
+ title: "BulkCreateAction[PoolBody]",
+} as const;
+
+export const $BulkCreateAction_VariableBody_ = {
+ properties: {
+ action: {
+ $ref: "#/components/schemas/BulkAction",
+ description: "The action to be performed on the entities.",
+ },
+ entities: {
+ items: {
+ $ref: "#/components/schemas/VariableBody",
+ },
+ type: "array",
+ title: "Entities",
+ description: "A list of entities to be created.",
+ },
+ action_on_existence: {
+ $ref: "#/components/schemas/BulkActionOnExistence",
+ default: "fail",
+ },
+ },
+ type: "object",
+ required: ["action", "entities"],
+ title: "BulkCreateAction[VariableBody]",
+} as const;
+
+export const $BulkDeleteAction_ConnectionBody_ = {
+ properties: {
+ action: {
+ $ref: "#/components/schemas/BulkAction",
+ description: "The action to be performed on the entities.",
+ },
+ entities: {
+ items: {
+ type: "string",
+ },
+ type: "array",
+ title: "Entities",
+ description: "A list of entity id/key to be deleted.",
+ },
+ action_on_non_existence: {
+ $ref: "#/components/schemas/BulkActionNotOnExistence",
+ default: "fail",
+ },
+ },
+ type: "object",
+ required: ["action", "entities"],
+ title: "BulkDeleteAction[ConnectionBody]",
+} as const;
+
+export const $BulkDeleteAction_PoolBody_ = {
+ properties: {
+ action: {
+ $ref: "#/components/schemas/BulkAction",
+ description: "The action to be performed on the entities.",
+ },
+ entities: {
+ items: {
+ type: "string",
+ },
+ type: "array",
+ title: "Entities",
+ description: "A list of entity id/key to be deleted.",
+ },
+ action_on_non_existence: {
+ $ref: "#/components/schemas/BulkActionNotOnExistence",
+ default: "fail",
+ },
+ },
+ type: "object",
+ required: ["action", "entities"],
+ title: "BulkDeleteAction[PoolBody]",
+} as const;
+
+export const $BulkDeleteAction_VariableBody_ = {
+ properties: {
+ action: {
+ $ref: "#/components/schemas/BulkAction",
+ description: "The action to be performed on the entities.",
+ },
+ entities: {
+ items: {
+ type: "string",
+ },
+ type: "array",
+ title: "Entities",
+ description: "A list of entity id/key to be deleted.",
+ },
+ action_on_non_existence: {
+ $ref: "#/components/schemas/BulkActionNotOnExistence",
+ default: "fail",
+ },
+ },
+ type: "object",
+ required: ["action", "entities"],
+ title: "BulkDeleteAction[VariableBody]",
+} as const;
+
+export const $BulkResponse = {
+ properties: {
+ create: {
+ anyOf: [
+ {
+ $ref: "#/components/schemas/BulkActionResponse",
+ },
+ {
+ type: "null",
+ },
+ ],
+ description: "Details of the bulk create operation, including successful keys and errors.",
+ },
+ update: {
+ anyOf: [
+ {
+ $ref: "#/components/schemas/BulkActionResponse",
+ },
+ {
+ type: "null",
+ },
+ ],
+ description: "Details of the bulk update operation, including successful keys and errors.",
+ },
+ delete: {
+ anyOf: [
+ {
+ $ref: "#/components/schemas/BulkActionResponse",
+ },
+ {
+ type: "null",
+ },
+ ],
+ description: "Details of the bulk delete operation, including successful keys and errors.",
+ },
+ },
+ type: "object",
+ title: "BulkResponse",
+ description: `Serializer for responses to bulk entity operations.
+
+This represents the results of create, update, and delete actions performed on entity in bulk.
+Each action (if requested) is represented as a field containing details about successful keys and any encountered errors.
+Fields are populated in the response only if the respective action was part of the request, else are set None.`,
+} as const;
+
+export const $BulkUpdateAction_ConnectionBody_ = {
+ properties: {
+ action: {
+ $ref: "#/components/schemas/BulkAction",
+ description: "The action to be performed on the entities.",
+ },
+ entities: {
+ items: {
+ $ref: "#/components/schemas/ConnectionBody",
+ },
+ type: "array",
+ title: "Entities",
+ description: "A list of entities to be updated.",
+ },
+ action_on_non_existence: {
+ $ref: "#/components/schemas/BulkActionNotOnExistence",
+ default: "fail",
+ },
+ },
+ type: "object",
+ required: ["action", "entities"],
+ title: "BulkUpdateAction[ConnectionBody]",
+} as const;
+
+export const $BulkUpdateAction_PoolBody_ = {
+ properties: {
+ action: {
+ $ref: "#/components/schemas/BulkAction",
+ description: "The action to be performed on the entities.",
+ },
+ entities: {
+ items: {
+ $ref: "#/components/schemas/PoolBody",
+ },
+ type: "array",
+ title: "Entities",
+ description: "A list of entities to be updated.",
+ },
+ action_on_non_existence: {
+ $ref: "#/components/schemas/BulkActionNotOnExistence",
+ default: "fail",
+ },
+ },
+ type: "object",
+ required: ["action", "entities"],
+ title: "BulkUpdateAction[PoolBody]",
+} as const;
+
+export const $BulkUpdateAction_VariableBody_ = {
+ properties: {
+ action: {
+ $ref: "#/components/schemas/BulkAction",
+ description: "The action to be performed on the entities.",
+ },
+ entities: {
+ items: {
+ $ref: "#/components/schemas/VariableBody",
+ },
+ type: "array",
+ title: "Entities",
+ description: "A list of entities to be updated.",
+ },
+ action_on_non_existence: {
+ $ref: "#/components/schemas/BulkActionNotOnExistence",
+ default: "fail",
+ },
+ },
+ type: "object",
+ required: ["action", "entities"],
+ title: "BulkUpdateAction[VariableBody]",
+} as const;
+
export const $ClearTaskInstancesBody = {
properties: {
dry_run: {
@@ -909,182 +1276,6 @@ export const $ConnectionBody = {
description: "Connection Serializer for requests body.",
} as const;
-export const $ConnectionBulkActionResponse = {
- properties: {
- success: {
- items: {
- type: "string",
- },
- type: "array",
- title: "Success",
- description: "A list of connection_ids representing successful operations.",
- },
- errors: {
- items: {
- type: "object",
- },
- type: "array",
- title: "Errors",
- description:
- "A list of errors encountered during the operation, each containing details about the issue.",
- },
- },
- type: "object",
- title: "ConnectionBulkActionResponse",
- description: `Serializer for individual bulk action responses.
-
-Represents the outcome of a single bulk operation (create, update, or delete).
-The response includes a list of successful connection_ids and any errors encountered during the operation.
-This structure helps users understand which key actions succeeded and which failed.`,
-} as const;
-
-export const $ConnectionBulkBody = {
- properties: {
- actions: {
- items: {
- anyOf: [
- {
- $ref: "#/components/schemas/ConnectionBulkCreateAction",
- },
- {
- $ref: "#/components/schemas/ConnectionBulkUpdateAction",
- },
- {
- $ref: "#/components/schemas/ConnectionBulkDeleteAction",
- },
- ],
- },
- type: "array",
- title: "Actions",
- description: "A list of Connection actions to perform.",
- },
- },
- type: "object",
- required: ["actions"],
- title: "ConnectionBulkBody",
- description: "Request body for bulk Connection operations (create, update, delete).",
-} as const;
-
-export const $ConnectionBulkCreateAction = {
- properties: {
- action: {
- $ref: "#/components/schemas/BulkAction",
- default: "create",
- },
- connections: {
- items: {
- $ref: "#/components/schemas/ConnectionBody",
- },
- type: "array",
- title: "Connections",
- description: "A list of connections to be created.",
- },
- action_on_existence: {
- $ref: "#/components/schemas/BulkActionOnExistence",
- default: "fail",
- },
- },
- type: "object",
- required: ["connections"],
- title: "ConnectionBulkCreateAction",
- description: "Bulk Create Variable serializer for request bodies.",
-} as const;
-
-export const $ConnectionBulkDeleteAction = {
- properties: {
- action: {
- $ref: "#/components/schemas/BulkAction",
- default: "delete",
- },
- connection_ids: {
- items: {
- type: "string",
- },
- type: "array",
- title: "Connection Ids",
- description: "A list of connection IDs to be deleted.",
- },
- action_on_non_existence: {
- $ref: "#/components/schemas/BulkActionNotOnExistence",
- default: "fail",
- },
- },
- type: "object",
- required: ["connection_ids"],
- title: "ConnectionBulkDeleteAction",
- description: "Bulk Delete Connection serializer for request bodies.",
-} as const;
-
-export const $ConnectionBulkResponse = {
- properties: {
- create: {
- anyOf: [
- {
- $ref: "#/components/schemas/ConnectionBulkActionResponse",
- },
- {
- type: "null",
- },
- ],
- description: "Details of the bulk create operation, including successful connection_ids and errors.",
- },
- update: {
- anyOf: [
- {
- $ref: "#/components/schemas/ConnectionBulkActionResponse",
- },
- {
- type: "null",
- },
- ],
- description: "Details of the bulk update operation, including successful connection_ids and errors.",
- },
- delete: {
- anyOf: [
- {
- $ref: "#/components/schemas/ConnectionBulkActionResponse",
- },
- {
- type: "null",
- },
- ],
- description: "Details of the bulk delete operation, including successful connection_ids and errors.",
- },
- },
- type: "object",
- title: "ConnectionBulkResponse",
- description: `Serializer for responses to bulk connection operations.
-
-This represents the results of create, update, and delete actions performed on connections in bulk.
-Each action (if requested) is represented as a field containing details about successful connection_ids and any encountered errors.
-Fields are populated in the response only if the respective action was part of the request, else are set None.`,
-} as const;
-
-export const $ConnectionBulkUpdateAction = {
- properties: {
- action: {
- $ref: "#/components/schemas/BulkAction",
- default: "update",
- },
- connections: {
- items: {
- $ref: "#/components/schemas/ConnectionBody",
- },
- type: "array",
- title: "Connections",
- description: "A list of connections to be updated.",
- },
- action_on_non_existence: {
- $ref: "#/components/schemas/BulkActionNotOnExistence",
- default: "fail",
- },
- },
- type: "object",
- required: ["connections"],
- title: "ConnectionBulkUpdateAction",
- description: "Bulk Update Connection serializer for request bodies.",
-} as const;
-
export const $ConnectionCollectionResponse = {
properties: {
connections: {
@@ -3832,219 +4023,77 @@ export const $PluginResponse = {
type: "string",
},
type: "array",
- title: "Operator Extra Links",
- },
- source: {
- type: "string",
- title: "Source",
- },
- listeners: {
- items: {
- type: "string",
- },
- type: "array",
- title: "Listeners",
- },
- timetables: {
- items: {
- type: "string",
- },
- type: "array",
- title: "Timetables",
- },
- },
- type: "object",
- required: [
- "name",
- "macros",
- "flask_blueprints",
- "fastapi_apps",
- "appbuilder_views",
- "appbuilder_menu_items",
- "global_operator_extra_links",
- "operator_extra_links",
- "source",
- "listeners",
- "timetables",
- ],
- title: "PluginResponse",
- description: "Plugin serializer.",
-} as const;
-
-export const $PoolBulkActionResponse = {
- properties: {
- success: {
- items: {
- type: "string",
- },
- type: "array",
- title: "Success",
- description: "A list of pool names representing successful operations.",
- },
- errors: {
- items: {
- type: "object",
- },
- type: "array",
- title: "Errors",
- description:
- "A list of errors encountered during the operation, each containing details about the issue.",
- },
- },
- type: "object",
- title: "PoolBulkActionResponse",
- description: `Serializer for individual bulk action responses.
-
-Represents the outcome of a single bulk operation (create, update, or delete).
-The response includes a list of successful pool names and any errors encountered during the operation.
-This structure helps users understand which key actions succeeded and which failed.`,
-} as const;
-
-export const $PoolBulkBody = {
- properties: {
- actions: {
- items: {
- anyOf: [
- {
- $ref: "#/components/schemas/PoolBulkCreateAction",
- },
- {
- $ref: "#/components/schemas/PoolBulkUpdateAction",
- },
- {
- $ref: "#/components/schemas/PoolBulkDeleteAction",
- },
- ],
- },
- type: "array",
- title: "Actions",
- description: "A list of Pool actions to perform.",
- },
- },
- type: "object",
- required: ["actions"],
- title: "PoolBulkBody",
- description: "Request body for bulk Pool operations (create, update, delete).",
-} as const;
-
-export const $PoolBulkCreateAction = {
- properties: {
- action: {
- $ref: "#/components/schemas/BulkAction",
- default: "create",
- },
- pools: {
- items: {
- $ref: "#/components/schemas/PoolPostBody",
- },
- type: "array",
- title: "Pools",
- description: "A list of pools to be created.",
- },
- action_on_existence: {
- $ref: "#/components/schemas/BulkActionOnExistence",
- default: "fail",
- },
- },
- type: "object",
- required: ["pools"],
- title: "PoolBulkCreateAction",
- description: "Bulk Create Pool serializer for request bodies.",
-} as const;
-
-export const $PoolBulkDeleteAction = {
- properties: {
- action: {
- $ref: "#/components/schemas/BulkAction",
- default: "delete",
+ title: "Operator Extra Links",
+ },
+ source: {
+ type: "string",
+ title: "Source",
},
- pool_names: {
+ listeners: {
items: {
type: "string",
},
type: "array",
- title: "Pool Names",
- description: "A list of pool names to be deleted.",
+ title: "Listeners",
},
- action_on_non_existence: {
- $ref: "#/components/schemas/BulkActionNotOnExistence",
- default: "fail",
+ timetables: {
+ items: {
+ type: "string",
+ },
+ type: "array",
+ title: "Timetables",
},
},
type: "object",
- required: ["pool_names"],
- title: "PoolBulkDeleteAction",
- description: "Bulk Delete Pool serializer for request bodies.",
+ required: [
+ "name",
+ "macros",
+ "flask_blueprints",
+ "fastapi_apps",
+ "appbuilder_views",
+ "appbuilder_menu_items",
+ "global_operator_extra_links",
+ "operator_extra_links",
+ "source",
+ "listeners",
+ "timetables",
+ ],
+ title: "PluginResponse",
+ description: "Plugin serializer.",
} as const;
-export const $PoolBulkResponse = {
+export const $PoolBody = {
properties: {
- create: {
- anyOf: [
- {
- $ref: "#/components/schemas/PoolBulkActionResponse",
- },
- {
- type: "null",
- },
- ],
- description: "Details of the bulk create operation, including successful pool names and errors.",
+ name: {
+ type: "string",
+ maxLength: 256,
+ title: "Name",
},
- update: {
- anyOf: [
- {
- $ref: "#/components/schemas/PoolBulkActionResponse",
- },
- {
- type: "null",
- },
- ],
- description: "Details of the bulk update operation, including successful pool names and errors.",
+ slots: {
+ type: "integer",
+ title: "Slots",
},
- delete: {
+ description: {
anyOf: [
{
- $ref: "#/components/schemas/PoolBulkActionResponse",
+ type: "string",
},
{
type: "null",
},
],
- description: "Details of the bulk delete operation, including successful pool names and errors.",
- },
- },
- type: "object",
- title: "PoolBulkResponse",
- description: `Serializer for responses to bulk pool operations.
-
-This represents the results of create, update, and delete actions performed on pools in bulk.
-Each action (if requested) is represented as a field containing details about successful pool names and any encountered errors.
-Fields are populated in the response only if the respective action was part of the request, else are set None.`,
-} as const;
-
-export const $PoolBulkUpdateAction = {
- properties: {
- action: {
- $ref: "#/components/schemas/BulkAction",
- default: "update",
- },
- pools: {
- items: {
- $ref: "#/components/schemas/PoolPatchBody",
- },
- type: "array",
- title: "Pools",
- description: "A list of pools to be updated.",
+ title: "Description",
},
- action_on_non_existence: {
- $ref: "#/components/schemas/BulkActionNotOnExistence",
- default: "fail",
+ include_deferred: {
+ type: "boolean",
+ title: "Include Deferred",
+ default: false,
},
},
type: "object",
- required: ["pools"],
- title: "PoolBulkUpdateAction",
- description: "Bulk Update Pool serializer for request bodies.",
+ required: ["name", "slots"],
+ title: "PoolBody",
+ description: "Pool serializer for post bodies.",
} as const;
export const $PoolCollectionResponse = {
@@ -4119,40 +4168,6 @@ export const $PoolPatchBody = {
description: "Pool serializer for patch bodies.",
} as const;
-export const $PoolPostBody = {
- properties: {
- name: {
- type: "string",
- maxLength: 256,
- title: "Name",
- },
- slots: {
- type: "integer",
- title: "Slots",
- },
- description: {
- anyOf: [
- {
- type: "string",
- },
- {
- type: "null",
- },
- ],
- title: "Description",
- },
- include_deferred: {
- type: "boolean",
- title: "Include Deferred",
- default: false,
- },
- },
- type: "object",
- required: ["name", "slots"],
- title: "PoolPostBody",
- description: "Pool serializer for post bodies.",
-} as const;
-
export const $PoolResponse = {
properties: {
name: {
@@ -5844,184 +5859,6 @@ export const $VariableBody = {
description: "Variable serializer for bodies.",
} as const;
-export const $VariableBulkActionResponse = {
- properties: {
- success: {
- items: {
- type: "string",
- },
- type: "array",
- title: "Success",
- description: "A list of keys representing successful operations.",
- default: [],
- },
- errors: {
- items: {
- type: "object",
- },
- type: "array",
- title: "Errors",
- description:
- "A list of errors encountered during the operation, each containing details about the issue.",
- default: [],
- },
- },
- type: "object",
- title: "VariableBulkActionResponse",
- description: `Serializer for individual bulk action responses.
-
-Represents the outcome of a single bulk operation (create, update, or delete).
-The response includes a list of successful keys and any errors encountered during the operation.
-This structure helps users understand which key actions succeeded and which failed.`,
-} as const;
-
-export const $VariableBulkBody = {
- properties: {
- actions: {
- items: {
- anyOf: [
- {
- $ref: "#/components/schemas/VariableBulkCreateAction",
- },
- {
- $ref: "#/components/schemas/VariableBulkUpdateAction",
- },
- {
- $ref: "#/components/schemas/VariableBulkDeleteAction",
- },
- ],
- },
- type: "array",
- title: "Actions",
- description: "A list of variable actions to perform.",
- },
- },
- type: "object",
- required: ["actions"],
- title: "VariableBulkBody",
- description: "Request body for bulk variable operations (create, update, delete).",
-} as const;
-
-export const $VariableBulkCreateAction = {
- properties: {
- action: {
- $ref: "#/components/schemas/BulkAction",
- default: "create",
- },
- variables: {
- items: {
- $ref: "#/components/schemas/VariableBody",
- },
- type: "array",
- title: "Variables",
- description: "A list of variables to be created.",
- },
- action_on_existence: {
- $ref: "#/components/schemas/BulkActionOnExistence",
- default: "fail",
- },
- },
- type: "object",
- required: ["variables"],
- title: "VariableBulkCreateAction",
- description: "Bulk Create Variable serializer for request bodies.",
-} as const;
-
-export const $VariableBulkDeleteAction = {
- properties: {
- action: {
- $ref: "#/components/schemas/BulkAction",
- default: "delete",
- },
- keys: {
- items: {
- type: "string",
- },
- type: "array",
- title: "Keys",
- description: "A list of variable keys to be deleted.",
- },
- action_on_non_existence: {
- $ref: "#/components/schemas/BulkActionNotOnExistence",
- default: "fail",
- },
- },
- type: "object",
- required: ["keys"],
- title: "VariableBulkDeleteAction",
- description: "Bulk Delete Variable serializer for request bodies.",
-} as const;
-
-export const $VariableBulkResponse = {
- properties: {
- create: {
- anyOf: [
- {
- $ref: "#/components/schemas/VariableBulkActionResponse",
- },
- {
- type: "null",
- },
- ],
- description: "Details of the bulk create operation, including successful keys and errors.",
- },
- update: {
- anyOf: [
- {
- $ref: "#/components/schemas/VariableBulkActionResponse",
- },
- {
- type: "null",
- },
- ],
- description: "Details of the bulk update operation, including successful keys and errors.",
- },
- delete: {
- anyOf: [
- {
- $ref: "#/components/schemas/VariableBulkActionResponse",
- },
- {
- type: "null",
- },
- ],
- description: "Details of the bulk delete operation, including successful keys and errors.",
- },
- },
- type: "object",
- title: "VariableBulkResponse",
- description: `Serializer for responses to bulk variable operations.
-
-This represents the results of create, update, and delete actions performed on variables in bulk.
-Each action (if requested) is represented as a field containing details about successful keys and any encountered errors.
-Fields are populated in the response only if the respective action was part of the request, else are set None.`,
-} as const;
-
-export const $VariableBulkUpdateAction = {
- properties: {
- action: {
- $ref: "#/components/schemas/BulkAction",
- default: "update",
- },
- variables: {
- items: {
- $ref: "#/components/schemas/VariableBody",
- },
- type: "array",
- title: "Variables",
- description: "A list of variables to be updated.",
- },
- action_on_non_existence: {
- $ref: "#/components/schemas/BulkActionNotOnExistence",
- default: "fail",
- },
- },
- type: "object",
- required: ["variables"],
- title: "VariableBulkUpdateAction",
- description: "Bulk Update Variable serializer for request bodies.",
-} as const;
-
export const $VariableCollectionResponse = {
properties: {
variables: {
diff --git a/airflow/ui/openapi-gen/requests/services.gen.ts b/airflow/ui/openapi-gen/requests/services.gen.ts
index d013b352a82..9a22483f682 100644
--- a/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -1144,7 +1144,7 @@ export class ConnectionService {
* Bulk create, update, and delete connections.
* @param data The data for the request.
* @param data.requestBody
- * @returns ConnectionBulkResponse Successful Response
+ * @returns BulkResponse Successful Response
* @throws ApiError
*/
public static bulkConnections(data: BulkConnectionsData): CancelablePromise<BulkConnectionsResponse> {
@@ -2794,7 +2794,7 @@ export class PoolService {
* Bulk create, update, and delete pools.
* @param data The data for the request.
* @param data.requestBody
- * @returns PoolBulkResponse Successful Response
+ * @returns BulkResponse Successful Response
* @throws ApiError
*/
public static bulkPools(data: BulkPoolsData): CancelablePromise<BulkPoolsResponse> {
@@ -3117,7 +3117,7 @@ export class VariableService {
* Bulk create, update, and delete variables.
* @param data The data for the request.
* @param data.requestBody
- * @returns VariableBulkResponse Successful Response
+ * @returns BulkResponse Successful Response
* @throws ApiError
*/
public static bulkVariables(data: BulkVariablesData): CancelablePromise<BulkVariablesResponse> {
diff --git a/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow/ui/openapi-gen/requests/types.gen.ts
index 1b597efdb07..60c53b753ca 100644
--- a/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -147,7 +147,7 @@ export type BaseInfoResponse = {
export type BulkAction = "create" | "delete" | "update";
/**
- * Bulk Action to be taken if the entity does not exists.
+ * Bulk Action to be taken if the entity does not exist.
*/
export type BulkActionNotOnExistence = "fail" | "skip";
@@ -156,6 +156,172 @@ export type BulkActionNotOnExistence = "fail" | "skip";
*/
export type BulkActionOnExistence = "fail" | "skip" | "overwrite";
+/**
+ * Serializer for individual bulk action responses.
+ *
+ * Represents the outcome of a single bulk operation (create, update, or delete).
+ * The response includes a list of successful keys and any errors encountered during the operation.
+ * This structure helps users understand which key actions succeeded and which failed.
+ */
+export type BulkActionResponse = {
+ /**
+ * A list of unique id/key representing successful operations.
+ */
+ success?: Array<string>;
+ /**
+ * A list of errors encountered during the operation, each containing details about the issue.
+ */
+ errors?: Array<{
+ [key: string]: unknown;
+ }>;
+};
+
+export type BulkBody_ConnectionBody_ = {
+ actions: Array<
+ BulkCreateAction_ConnectionBody_ | BulkUpdateAction_ConnectionBody_ | BulkDeleteAction_ConnectionBody_
+ >;
+};
+
+export type BulkBody_PoolBody_ = {
+ actions: Array<BulkCreateAction_PoolBody_ | BulkUpdateAction_PoolBody_ | BulkDeleteAction_PoolBody_>;
+};
+
+export type BulkBody_VariableBody_ = {
+ actions: Array<
+ BulkCreateAction_VariableBody_ | BulkUpdateAction_VariableBody_ | BulkDeleteAction_VariableBody_
+ >;
+};
+
+export type BulkCreateAction_ConnectionBody_ = {
+ /**
+ * The action to be performed on the entities.
+ */
+ action: BulkAction;
+ /**
+ * A list of entities to be created.
+ */
+ entities: Array<ConnectionBody>;
+ action_on_existence?: BulkActionOnExistence;
+};
+
+export type BulkCreateAction_PoolBody_ = {
+ /**
+ * The action to be performed on the entities.
+ */
+ action: BulkAction;
+ /**
+ * A list of entities to be created.
+ */
+ entities: Array<PoolBody>;
+ action_on_existence?: BulkActionOnExistence;
+};
+
+export type BulkCreateAction_VariableBody_ = {
+ /**
+ * The action to be performed on the entities.
+ */
+ action: BulkAction;
+ /**
+ * A list of entities to be created.
+ */
+ entities: Array<VariableBody>;
+ action_on_existence?: BulkActionOnExistence;
+};
+
+export type BulkDeleteAction_ConnectionBody_ = {
+ /**
+ * The action to be performed on the entities.
+ */
+ action: BulkAction;
+ /**
+ * A list of entity id/key to be deleted.
+ */
+ entities: Array<string>;
+ action_on_non_existence?: BulkActionNotOnExistence;
+};
+
+export type BulkDeleteAction_PoolBody_ = {
+ /**
+ * The action to be performed on the entities.
+ */
+ action: BulkAction;
+ /**
+ * A list of entity id/key to be deleted.
+ */
+ entities: Array<string>;
+ action_on_non_existence?: BulkActionNotOnExistence;
+};
+
+export type BulkDeleteAction_VariableBody_ = {
+ /**
+ * The action to be performed on the entities.
+ */
+ action: BulkAction;
+ /**
+ * A list of entity id/key to be deleted.
+ */
+ entities: Array<string>;
+ action_on_non_existence?: BulkActionNotOnExistence;
+};
+
+/**
+ * Serializer for responses to bulk entity operations.
+ *
+ * This represents the results of create, update, and delete actions performed on entity in bulk.
+ * Each action (if requested) is represented as a field containing details about successful keys and any encountered errors.
+ * Fields are populated in the response only if the respective action was part of the request, else are set None.
+ */
+export type BulkResponse = {
+ /**
+ * Details of the bulk create operation, including successful keys and errors.
+ */
+ create?: BulkActionResponse | null;
+ /**
+ * Details of the bulk update operation, including successful keys and errors.
+ */
+ update?: BulkActionResponse | null;
+ /**
+ * Details of the bulk delete operation, including successful keys and errors.
+ */
+ delete?: BulkActionResponse | null;
+};
+
+export type BulkUpdateAction_ConnectionBody_ = {
+ /**
+ * The action to be performed on the entities.
+ */
+ action: BulkAction;
+ /**
+ * A list of entities to be updated.
+ */
+ entities: Array<ConnectionBody>;
+ action_on_non_existence?: BulkActionNotOnExistence;
+};
+
+export type BulkUpdateAction_PoolBody_ = {
+ /**
+ * The action to be performed on the entities.
+ */
+ action: BulkAction;
+ /**
+ * A list of entities to be updated.
+ */
+ entities: Array<PoolBody>;
+ action_on_non_existence?: BulkActionNotOnExistence;
+};
+
+export type BulkUpdateAction_VariableBody_ = {
+ /**
+ * The action to be performed on the entities.
+ */
+ action: BulkAction;
+ /**
+ * A list of entities to be updated.
+ */
+ entities: Array<VariableBody>;
+ action_on_non_existence?: BulkActionNotOnExistence;
+};
+
/**
* Request body for Clear Task Instances endpoint.
*/
@@ -240,94 +406,6 @@ export type ConnectionBody = {
extra?: string | null;
};
-/**
- * Serializer for individual bulk action responses.
- *
- * Represents the outcome of a single bulk operation (create, update, or delete).
- * The response includes a list of successful connection_ids and any errors encountered during the operation.
- * This structure helps users understand which key actions succeeded and which failed.
- */
-export type ConnectionBulkActionResponse = {
- /**
- * A list of connection_ids representing successful operations.
- */
- success?: Array<string>;
- /**
- * A list of errors encountered during the operation, each containing details about the issue.
- */
- errors?: Array<{
- [key: string]: unknown;
- }>;
-};
-
-/**
- * Request body for bulk Connection operations (create, update, delete).
- */
-export type ConnectionBulkBody = {
- /**
- * A list of Connection actions to perform.
- */
- actions: Array<ConnectionBulkCreateAction | ConnectionBulkUpdateAction | ConnectionBulkDeleteAction>;
-};
-
-/**
- * Bulk Create Variable serializer for request bodies.
- */
-export type ConnectionBulkCreateAction = {
- action?: BulkAction;
- /**
- * A list of connections to be created.
- */
- connections: Array<ConnectionBody>;
- action_on_existence?: BulkActionOnExistence;
-};
-
-/**
- * Bulk Delete Connection serializer for request bodies.
- */
-export type ConnectionBulkDeleteAction = {
- action?: BulkAction;
- /**
- * A list of connection IDs to be deleted.
- */
- connection_ids: Array<string>;
- action_on_non_existence?: BulkActionNotOnExistence;
-};
-
-/**
- * Serializer for responses to bulk connection operations.
- *
- * This represents the results of create, update, and delete actions performed on connections in bulk.
- * Each action (if requested) is represented as a field containing details about successful connection_ids and any encountered errors.
- * Fields are populated in the response only if the respective action was part of the request, else are set None.
- */
-export type ConnectionBulkResponse = {
- /**
- * Details of the bulk create operation, including successful connection_ids and errors.
- */
- create?: ConnectionBulkActionResponse | null;
- /**
- * Details of the bulk update operation, including successful connection_ids and errors.
- */
- update?: ConnectionBulkActionResponse | null;
- /**
- * Details of the bulk delete operation, including successful connection_ids and errors.
- */
- delete?: ConnectionBulkActionResponse | null;
-};
-
-/**
- * Bulk Update Connection serializer for request bodies.
- */
-export type ConnectionBulkUpdateAction = {
- action?: BulkAction;
- /**
- * A list of connections to be updated.
- */
- connections: Array<ConnectionBody>;
- action_on_non_existence?: BulkActionNotOnExistence;
-};
-
/**
* Connection Collection serializer for responses.
*/
@@ -981,91 +1059,13 @@ export type PluginResponse = {
};
/**
- * Serializer for individual bulk action responses.
- *
- * Represents the outcome of a single bulk operation (create, update, or delete).
- * The response includes a list of successful pool names and any errors encountered during the operation.
- * This structure helps users understand which key actions succeeded and which failed.
- */
-export type PoolBulkActionResponse = {
- /**
- * A list of pool names representing successful operations.
- */
- success?: Array<string>;
- /**
- * A list of errors encountered during the operation, each containing details about the issue.
- */
- errors?: Array<{
- [key: string]: unknown;
- }>;
-};
-
-/**
- * Request body for bulk Pool operations (create, update, delete).
- */
-export type PoolBulkBody = {
- /**
- * A list of Pool actions to perform.
- */
- actions: Array<PoolBulkCreateAction | PoolBulkUpdateAction | PoolBulkDeleteAction>;
-};
-
-/**
- * Bulk Create Pool serializer for request bodies.
- */
-export type PoolBulkCreateAction = {
- action?: BulkAction;
- /**
- * A list of pools to be created.
- */
- pools: Array<PoolPostBody>;
- action_on_existence?: BulkActionOnExistence;
-};
-
-/**
- * Bulk Delete Pool serializer for request bodies.
- */
-export type PoolBulkDeleteAction = {
- action?: BulkAction;
- /**
- * A list of pool names to be deleted.
- */
- pool_names: Array<string>;
- action_on_non_existence?: BulkActionNotOnExistence;
-};
-
-/**
- * Serializer for responses to bulk pool operations.
- *
- * This represents the results of create, update, and delete actions performed on pools in bulk.
- * Each action (if requested) is represented as a field containing details about successful pool names and any encountered errors.
- * Fields are populated in the response only if the respective action was part of the request, else are set None.
- */
-export type PoolBulkResponse = {
- /**
- * Details of the bulk create operation, including successful pool names and errors.
- */
- create?: PoolBulkActionResponse | null;
- /**
- * Details of the bulk update operation, including successful pool names and errors.
- */
- update?: PoolBulkActionResponse | null;
- /**
- * Details of the bulk delete operation, including successful pool names and errors.
- */
- delete?: PoolBulkActionResponse | null;
-};
-
-/**
- * Bulk Update Pool serializer for request bodies.
+ * Pool serializer for post bodies.
*/
-export type PoolBulkUpdateAction = {
- action?: BulkAction;
- /**
- * A list of pools to be updated.
- */
- pools: Array<PoolPatchBody>;
- action_on_non_existence?: BulkActionNotOnExistence;
+export type PoolBody = {
+ name: string;
+ slots: number;
+ description?: string | null;
+ include_deferred?: boolean;
};
/**
@@ -1086,16 +1086,6 @@ export type PoolPatchBody = {
include_deferred?: boolean | null;
};
-/**
- * Pool serializer for post bodies.
- */
-export type PoolPostBody = {
- name: string;
- slots: number;
- description?: string | null;
- include_deferred?: boolean;
-};
-
/**
* Pool serializer for responses.
*/
@@ -1453,94 +1443,6 @@ export type VariableBody = {
description?: string | null;
};
-/**
- * Serializer for individual bulk action responses.
- *
- * Represents the outcome of a single bulk operation (create, update, or delete).
- * The response includes a list of successful keys and any errors encountered during the operation.
- * This structure helps users understand which key actions succeeded and which failed.
- */
-export type VariableBulkActionResponse = {
- /**
- * A list of keys representing successful operations.
- */
- success?: Array<string>;
- /**
- * A list of errors encountered during the operation, each containing details about the issue.
- */
- errors?: Array<{
- [key: string]: unknown;
- }>;
-};
-
-/**
- * Request body for bulk variable operations (create, update, delete).
- */
-export type VariableBulkBody = {
- /**
- * A list of variable actions to perform.
- */
- actions: Array<VariableBulkCreateAction | VariableBulkUpdateAction | VariableBulkDeleteAction>;
-};
-
-/**
- * Bulk Create Variable serializer for request bodies.
- */
-export type VariableBulkCreateAction = {
- action?: BulkAction;
- /**
- * A list of variables to be created.
- */
- variables: Array<VariableBody>;
- action_on_existence?: BulkActionOnExistence;
-};
-
-/**
- * Bulk Delete Variable serializer for request bodies.
- */
-export type VariableBulkDeleteAction = {
- action?: BulkAction;
- /**
- * A list of variable keys to be deleted.
- */
- keys: Array<string>;
- action_on_non_existence?: BulkActionNotOnExistence;
-};
-
-/**
- * Serializer for responses to bulk variable operations.
- *
- * This represents the results of create, update, and delete actions performed on variables in bulk.
- * Each action (if requested) is represented as a field containing details about successful keys and any encountered errors.
- * Fields are populated in the response only if the respective action was part of the request, else are set None.
- */
-export type VariableBulkResponse = {
- /**
- * Details of the bulk create operation, including successful keys and errors.
- */
- create?: VariableBulkActionResponse | null;
- /**
- * Details of the bulk update operation, including successful keys and errors.
- */
- update?: VariableBulkActionResponse | null;
- /**
- * Details of the bulk delete operation, including successful keys and errors.
- */
- delete?: VariableBulkActionResponse | null;
-};
-
-/**
- * Bulk Update Variable serializer for request bodies.
- */
-export type VariableBulkUpdateAction = {
- action?: BulkAction;
- /**
- * A list of variables to be updated.
- */
- variables: Array<VariableBody>;
- action_on_non_existence?: BulkActionNotOnExistence;
-};
-
/**
* Variable Collection serializer for responses.
*/
@@ -1878,10 +1780,10 @@ export type PostConnectionData = {
export type PostConnectionResponse = ConnectionResponse;
export type BulkConnectionsData = {
- requestBody: ConnectionBulkBody;
+ requestBody: BulkBody_ConnectionBody_;
};
-export type BulkConnectionsResponse = ConnectionBulkResponse;
+export type BulkConnectionsResponse = BulkResponse;
export type TestConnectionData = {
requestBody: ConnectionBody;
@@ -2337,16 +2239,16 @@ export type GetPoolsData = {
export type GetPoolsResponse = PoolCollectionResponse;
export type PostPoolData = {
- requestBody: PoolPostBody;
+ requestBody: PoolBody;
};
export type PostPoolResponse = PoolResponse;
export type BulkPoolsData = {
- requestBody: PoolBulkBody;
+ requestBody: BulkBody_PoolBody_;
};
-export type BulkPoolsResponse = PoolBulkResponse;
+export type BulkPoolsResponse = BulkResponse;
export type GetProvidersData = {
limit?: number;
@@ -2429,10 +2331,10 @@ export type PostVariableData = {
export type PostVariableResponse = VariableResponse;
export type BulkVariablesData = {
- requestBody: VariableBulkBody;
+ requestBody: BulkBody_VariableBody_;
};
-export type BulkVariablesResponse = VariableBulkResponse;
+export type BulkVariablesResponse = BulkResponse;
export type ReparseDagFileData = {
fileToken: string;
@@ -3298,7 +3200,7 @@ export type $OpenApiTs = {
/**
* Successful Response
*/
- 200: ConnectionBulkResponse;
+ 200: BulkResponse;
/**
* Unauthorized
*/
@@ -4628,7 +4530,7 @@ export type $OpenApiTs = {
/**
* Successful Response
*/
- 200: PoolBulkResponse;
+ 200: BulkResponse;
/**
* Unauthorized
*/
@@ -4925,7 +4827,7 @@ export type $OpenApiTs = {
/**
* Successful Response
*/
- 200: VariableBulkResponse;
+ 200: BulkResponse;
/**
* Unauthorized
*/
diff --git a/airflow/ui/src/pages/Variables/DeleteVariablesButton.tsx b/airflow/ui/src/pages/Variables/DeleteVariablesButton.tsx
index 368936243b2..dfb1f7d20cb 100644
--- a/airflow/ui/src/pages/Variables/DeleteVariablesButton.tsx
+++ b/airflow/ui/src/pages/Variables/DeleteVariablesButton.tsx
@@ -81,7 +81,7 @@ const DeleteVariablesButton = ({ clearSelections, deleteKeys: variableKeys }: Pr
{
action: "delete" as const,
action_on_non_existence: "fail",
- keys: variableKeys,
+ entities: variableKeys,
},
],
},
diff --git a/airflow/ui/src/pages/Variables/ImportVariablesForm.tsx b/airflow/ui/src/pages/Variables/ImportVariablesForm.tsx
index 5645c91d4ba..7eede0f205c 100644
--- a/airflow/ui/src/pages/Variables/ImportVariablesForm.tsx
+++ b/airflow/ui/src/pages/Variables/ImportVariablesForm.tsx
@@ -105,7 +105,7 @@ const ImportVariablesForm = ({ onClose }: ImportVariablesFormProps) => {
{
action: "create" as const,
action_if_exists: actionIfExists,
- variables: Object.entries(fileContent).map(([key, value]) => ({
+ entities: Object.entries(fileContent).map(([key, value]) => ({
key,
value,
})),
diff --git a/pyproject.toml b/pyproject.toml
index c6448b61e84..4e246020256 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -342,6 +342,10 @@ testing = ["dev", "providers.tests", "task_sdk.tests", "tests_common", "tests"]
# Pydantic also require models to be imported during execution
"airflow/serialization/pydantic/*.py" = ["I002", "UP007", "TC001"]
+# The FastAPI models are not parsed well with Pydantic when __future__.annotations is used
+# It is forcing for Union to replace with `|` pipe instead
+"airflow/api_fastapi/core_api/datamodels/common.py" = ["UP007"]
+
# Provider's get_provider_info.py files are generated and do not need "from __future__ import annotations"
"providers/**/get_provider_info.py" = ["I002"]
diff --git a/tests/api_fastapi/core_api/routes/public/test_connections.py b/tests/api_fastapi/core_api/routes/public/test_connections.py
index 622365d9662..8a0d7ef6346 100644
--- a/tests/api_fastapi/core_api/routes/public/test_connections.py
+++ b/tests/api_fastapi/core_api/routes/public/test_connections.py
@@ -647,7 +647,7 @@ class TestBulkConnections(TestConnectionEndpoint):
"actions": [
{
"action": "create",
- "connections": [
+ "entities": [
{
"connection_id": "NOT_EXISTING_CONN_ID",
"conn_type": "NOT_EXISTING_CONN_TYPE",
@@ -670,7 +670,7 @@ class TestBulkConnections(TestConnectionEndpoint):
"actions": [
{
"action": "create",
- "connections": [
+ "entities": [
{
"connection_id": TEST_CONN_ID,
"conn_type": TEST_CONN_TYPE,
@@ -697,7 +697,7 @@ class TestBulkConnections(TestConnectionEndpoint):
"actions": [
{
"action": "create",
- "connections": [
+ "entities": [
{
"connection_id": TEST_CONN_ID,
"conn_type": TEST_CONN_TYPE,
@@ -721,7 +721,7 @@ class TestBulkConnections(TestConnectionEndpoint):
"actions": [
{
"action": "create",
- "connections": [
+ "entities": [
{
"connection_id": TEST_CONN_ID,
"conn_type": TEST_CONN_TYPE,
@@ -753,7 +753,7 @@ class TestBulkConnections(TestConnectionEndpoint):
"actions": [
{
"action": "update",
- "connections": [
+ "entities": [
{
"connection_id": TEST_CONN_ID,
"conn_type": TEST_CONN_TYPE,
@@ -777,7 +777,7 @@ class TestBulkConnections(TestConnectionEndpoint):
"actions": [
{
"action": "update",
- "connections": [
+ "entities": [
{
"connection_id": "NOT_EXISTING_CONN_ID",
"conn_type": "NOT_EXISTING_CONN_TYPE",
@@ -800,7 +800,7 @@ class TestBulkConnections(TestConnectionEndpoint):
"actions": [
{
"action": "update",
- "connections": [
+ "entities": [
{
"connection_id": "NOT_EXISTING_CONN_ID",
"conn_type": "NOT_EXISTING_CONN_TYPE",
@@ -828,7 +828,7 @@ class TestBulkConnections(TestConnectionEndpoint):
"actions": [
{
"action": "delete",
- "connection_ids": [TEST_CONN_ID],
+ "entities": [TEST_CONN_ID],
}
]
},
@@ -845,7 +845,7 @@ class TestBulkConnections(TestConnectionEndpoint):
"actions": [
{
"action": "delete",
- "connection_ids": ["NOT_EXISTING_CONN_ID"],
+ "entities": ["NOT_EXISTING_CONN_ID"],
"action_on_non_existence": "skip",
}
]
@@ -863,7 +863,7 @@ class TestBulkConnections(TestConnectionEndpoint):
"actions": [
{
"action": "delete",
- "connection_ids": ["NOT_EXISTING_CONN_ID"],
+ "entities": ["NOT_EXISTING_CONN_ID"],
"action_on_non_existence": "fail",
}
]
@@ -886,7 +886,7 @@ class TestBulkConnections(TestConnectionEndpoint):
"actions": [
{
"action": "create",
- "connections": [
+ "entities": [
{
"connection_id": "NOT_EXISTING_CONN_ID",
"conn_type": "NOT_EXISTING_CONN_TYPE",
@@ -896,7 +896,7 @@ class TestBulkConnections(TestConnectionEndpoint):
},
{
"action": "update",
- "connections": [
+ "entities": [
{
"connection_id": TEST_CONN_ID,
"conn_type": TEST_CONN_TYPE,
@@ -907,7 +907,7 @@ class TestBulkConnections(TestConnectionEndpoint):
},
{
"action": "delete",
- "connection_ids": [TEST_CONN_ID],
+ "entities": [TEST_CONN_ID],
"action_on_non_existence": "skip",
},
]
diff --git a/tests/api_fastapi/core_api/routes/public/test_pools.py b/tests/api_fastapi/core_api/routes/public/test_pools.py
index 5a3d6531088..fbf1ef1a246 100644
--- a/tests/api_fastapi/core_api/routes/public/test_pools.py
+++ b/tests/api_fastapi/core_api/routes/public/test_pools.py
@@ -392,7 +392,7 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "create",
- "pools": [
+ "entities": [
{"name": "pool3", "slots": 10, "description": "New Description"},
{"name": "pool4", "slots": 20, "description": "New Description"},
],
@@ -408,7 +408,7 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "create",
- "pools": [
+ "entities": [
{"name": "pool3", "slots": 10, "description": "New Description"},
{"name": "pool1", "slots": 20, "description": "New Description"},
],
@@ -424,7 +424,7 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "create",
- "pools": [
+ "entities": [
{"name": "pool3", "slots": 10, "description": "New Description"},
{"name": "pool2", "slots": 20, "description": "New Description"},
],
@@ -440,7 +440,7 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "create",
- "pools": [{"name": "pool2", "slots": 20, "description": "New Description"}],
+ "entities": [{"name": "pool2", "slots": 20, "description": "New Description"}],
"action_on_existence": "fail",
}
]
@@ -463,7 +463,7 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "update",
- "pools": [{"name": "pool2", "slots": 10, "description": "New Description"}],
+ "entities": [{"name": "pool2", "slots": 10, "description": "New Description"}],
"action_on_non_existence": "fail",
}
]
@@ -476,7 +476,7 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "update",
- "pools": [{"name": "pool4", "slots": 20, "description": "New Description"}],
+ "entities": [{"name": "pool4", "slots": 20, "description": "New Description"}],
"action_on_non_existence": "skip",
}
]
@@ -489,7 +489,7 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "update",
- "pools": [{"name": "pool4", "slots": 10, "description": "New Description"}],
+ "entities": [{"name": "pool4", "slots": 10, "description": "New Description"}],
"action_on_non_existence": "fail",
}
]
@@ -508,29 +508,17 @@ class TestBulkPools(TestPoolsEndpoint):
),
# Test successful delete
(
- {
- "actions": [
- {"action": "delete", "pool_names": ["pool1"], "action_on_non_existence": "skip"}
- ]
- },
+ {"actions": [{"action": "delete", "entities": ["pool1"], "action_on_non_existence": "skip"}]},
{"delete": {"success": ["pool1"], "errors": []}},
),
# Test delete with skip
(
- {
- "actions": [
- {"action": "delete", "pool_names": ["pool3"], "action_on_non_existence": "skip"}
- ]
- },
+ {"actions": [{"action": "delete", "entities": ["pool3"], "action_on_non_existence": "skip"}]},
{"delete": {"success": [], "errors": []}},
),
# Test delete not found
(
- {
- "actions": [
- {"action": "delete", "pool_names": ["pool4"], "action_on_non_existence": "fail"}
- ]
- },
+ {"actions": [{"action": "delete", "entities": ["pool4"], "action_on_non_existence": "fail"}]},
{
"delete": {
"success": [],
@@ -549,15 +537,15 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "create",
- "pools": [{"name": "pool6", "slots": 10, "description": "New Description"}],
+ "entities": [{"name": "pool6", "slots": 10, "description": "New Description"}],
"action_on_existence": "skip",
},
{
"action": "update",
- "pools": [{"name": "pool1", "slots": 10, "description": "New Description"}],
+ "entities": [{"name": "pool1", "slots": 10, "description": "New Description"}],
"action_on_non_existence": "fail",
},
- {"action": "delete", "pool_names": ["pool2"], "action_on_non_existence": "skip"},
+ {"action": "delete", "entities": ["pool2"], "action_on_non_existence": "skip"},
]
},
{
@@ -572,15 +560,15 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "create",
- "pools": [{"name": "pool1", "slots": 10, "description": "New Description"}],
+ "entities": [{"name": "pool1", "slots": 10, "description": "New Description"}],
"action_on_existence": "fail",
},
{
"action": "update",
- "pools": [{"name": "pool1", "slots": 100, "description": "New Description"}],
+ "entities": [{"name": "pool1", "slots": 100, "description": "New Description"}],
"action_on_non_existence": "fail",
},
- {"action": "delete", "pool_names": ["pool4"], "action_on_non_existence": "skip"},
+ {"action": "delete", "entities": ["pool4"], "action_on_non_existence": "skip"},
]
},
{
@@ -603,15 +591,15 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "create",
- "pools": [{"name": "pool1", "slots": 10, "description": "New Description"}],
+ "entities": [{"name": "pool1", "slots": 10, "description": "New Description"}],
"action_on_existence": "skip",
},
{
"action": "update",
- "pools": [{"name": "pool5", "slots": 10, "description": "New Description"}],
+ "entities": [{"name": "pool5", "slots": 10, "description": "New Description"}],
"action_on_non_existence": "skip",
},
- {"action": "delete", "pool_names": ["pool5"], "action_on_non_existence": "skip"},
+ {"action": "delete", "entities": ["pool5"], "action_on_non_existence": "skip"},
]
},
{
@@ -626,15 +614,17 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "create",
- "pools": [{"name": "pool5", "slots": 10, "description": "New Description"}],
+ "entities": [{"name": "pool5", "slots": 10, "description": "New Description"}],
"action_on_existence": "fail",
},
{
"action": "update",
- "pools": [{"name": "pool5", "slots": 100, "description": "New test Description"}],
+ "entities": [
+ {"name": "pool5", "slots": 100, "description": "New test Description"}
+ ],
"action_on_non_existence": "fail",
},
- {"action": "delete", "pool_names": ["pool5"], "action_on_non_existence": "fail"},
+ {"action": "delete", "entities": ["pool5"], "action_on_non_existence": "fail"},
]
},
{
@@ -649,33 +639,45 @@ class TestBulkPools(TestPoolsEndpoint):
"actions": [
{
"action": "create",
- "pools": [{"name": "pool1", "slots": 100, "description": "New test Description"}],
+ "entities": [
+ {"name": "pool1", "slots": 100, "description": "New test Description"}
+ ],
"action_on_existence": "fail",
},
{
"action": "update",
- "pools": [{"name": "pool1", "slots": 100, "description": "New test Description"}],
+ "entities": [
+ {"name": "pool1", "slots": 100, "description": "New test Description"}
+ ],
"action_on_non_existence": "fail",
},
{
"action": "create",
- "pools": [{"name": "pool5", "slots": 100, "description": "New test Description"}],
+ "entities": [
+ {"name": "pool5", "slots": 100, "description": "New test Description"}
+ ],
"action_on_existence": "fail",
},
{
"action": "update",
- "pools": [{"name": "pool8", "slots": 100, "description": "New test Description"}],
+ "entities": [
+ {"name": "pool8", "slots": 100, "description": "New test Description"}
+ ],
"action_on_non_existence": "fail",
},
- {"action": "delete", "pool_names": ["pool2"], "action_on_non_existence": "fail"},
+ {"action": "delete", "entities": ["pool2"], "action_on_non_existence": "fail"},
{
"action": "create",
- "pools": [{"name": "pool6", "slots": 100, "description": "New test Description"}],
+ "entities": [
+ {"name": "pool6", "slots": 100, "description": "New test Description"}
+ ],
"action_on_existence": "fail",
},
{
"action": "update",
- "pools": [{"name": "pool9", "slots": 100, "description": "New test Description"}],
+ "entities": [
+ {"name": "pool9", "slots": 100, "description": "New test Description"}
+ ],
"action_on_non_existence": "fail",
},
]
diff --git a/tests/api_fastapi/core_api/routes/public/test_variables.py b/tests/api_fastapi/core_api/routes/public/test_variables.py
index ebbb17350a8..5de6ff1f699 100644
--- a/tests/api_fastapi/core_api/routes/public/test_variables.py
+++ b/tests/api_fastapi/core_api/routes/public/test_variables.py
@@ -444,7 +444,7 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "create",
- "variables": [
+ "entities": [
{"key": "new_var1", "value": "new_value1", "description": "New variable 1"},
{"key": "new_var2", "value": "new_value2", "description": "New variable 2"},
],
@@ -460,7 +460,7 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "create",
- "variables": [
+ "entities": [
{
"key": "test_variable_key",
"value": "new_value1",
@@ -480,7 +480,7 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "create",
- "variables": [
+ "entities": [
{
"key": "test_variable_key",
"value": "new_value1",
@@ -500,7 +500,7 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "create",
- "variables": [
+ "entities": [
{
"key": "test_variable_key",
"value": "new_value",
@@ -529,7 +529,7 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "update",
- "variables": [
+ "entities": [
{
"key": "test_variable_key",
"value": "updated_value",
@@ -548,7 +548,7 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "update",
- "variables": [
+ "entities": [
{
"key": "key_not_present",
"value": "updated_value",
@@ -567,7 +567,7 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "update",
- "variables": [
+ "entities": [
{
"key": "nonexistent_var",
"value": "updated_value",
@@ -594,7 +594,11 @@ class TestBulkVariables(TestVariableEndpoint):
(
{
"actions": [
- {"action": "delete", "keys": ["test_variable_key"], "action_on_non_existence": "skip"}
+ {
+ "action": "delete",
+ "entities": ["test_variable_key"],
+ "action_on_non_existence": "skip",
+ }
]
},
{"delete": {"success": ["test_variable_key"], "errors": []}},
@@ -603,7 +607,11 @@ class TestBulkVariables(TestVariableEndpoint):
(
{
"actions": [
- {"action": "delete", "keys": ["key_not_present"], "action_on_non_existence": "skip"}
+ {
+ "action": "delete",
+ "entities": ["key_not_present"],
+ "action_on_non_existence": "skip",
+ }
]
},
{"delete": {"success": [], "errors": []}},
@@ -612,7 +620,11 @@ class TestBulkVariables(TestVariableEndpoint):
(
{
"actions": [
- {"action": "delete", "keys": ["nonexistent_var"], "action_on_non_existence": "fail"}
+ {
+ "action": "delete",
+ "entities": ["nonexistent_var"],
+ "action_on_non_existence": "fail",
+ }
]
},
{
@@ -633,12 +645,12 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "create",
- "variables": [{"key": "new_var1", "value": "new_value1"}],
+ "entities": [{"key": "new_var1", "value": "new_value1"}],
"action_on_existence": "skip",
},
{
"action": "update",
- "variables": [
+ "entities": [
{
"key": "test_variable_key",
"value": "updated_value",
@@ -649,7 +661,7 @@ class TestBulkVariables(TestVariableEndpoint):
},
{
"action": "delete",
- "keys": ["dictionary_password"],
+ "entities": ["dictionary_password"],
"action_on_non_existence": "skip",
},
]
@@ -666,7 +678,7 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "create",
- "variables": [
+ "entities": [
{
"key": "test_variable_key",
"value": "new_value",
@@ -677,7 +689,7 @@ class TestBulkVariables(TestVariableEndpoint):
},
{
"action": "update",
- "variables": [
+ "entities": [
{
"key": "dictionary_password",
"value": "updated_value",
@@ -686,7 +698,11 @@ class TestBulkVariables(TestVariableEndpoint):
],
"action_on_non_existence": "fail",
},
- {"action": "delete", "keys": ["nonexistent_var"], "action_on_non_existence": "skip"},
+ {
+ "action": "delete",
+ "entities": ["nonexistent_var"],
+ "action_on_non_existence": "skip",
+ },
]
},
{
@@ -709,7 +725,7 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "create",
- "variables": [
+ "entities": [
{
"key": "test_variable_key",
"value": "new_value1",
@@ -720,7 +736,7 @@ class TestBulkVariables(TestVariableEndpoint):
},
{
"action": "update",
- "variables": [
+ "entities": [
{
"key": "nonexistent_var",
"value": "updated_value",
@@ -729,7 +745,11 @@ class TestBulkVariables(TestVariableEndpoint):
],
"action_on_non_existence": "skip",
},
- {"action": "delete", "keys": ["nonexistent_var"], "action_on_non_existence": "skip"},
+ {
+ "action": "delete",
+ "entities": ["nonexistent_var"],
+ "action_on_non_existence": "skip",
+ },
]
},
{
@@ -744,7 +764,7 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "create",
- "variables": [
+ "entities": [
{
"key": "new_variable_key",
"value": "new_value1",
@@ -755,7 +775,7 @@ class TestBulkVariables(TestVariableEndpoint):
},
{
"action": "update",
- "variables": [
+ "entities": [
{
"key": "new_variable_key",
"value": "updated_value",
@@ -764,7 +784,11 @@ class TestBulkVariables(TestVariableEndpoint):
],
"action_on_non_existence": "fail",
},
- {"action": "delete", "keys": ["new_variable_key"], "action_on_non_existence": "fail"},
+ {
+ "action": "delete",
+ "entities": ["new_variable_key"],
+ "action_on_non_existence": "fail",
+ },
]
},
{
@@ -779,7 +803,7 @@ class TestBulkVariables(TestVariableEndpoint):
"actions": [
{
"action": "create",
- "variables": [
+ "entities": [
{
"key": "test_variable_key",
"value": "new_value1",
@@ -790,7 +814,7 @@ class TestBulkVariables(TestVariableEndpoint):
},
{
"action": "update",
- "variables": [
+ "entities": [
{
"key": "test_variable_key",
"value": "updated_value",
@@ -801,7 +825,7 @@ class TestBulkVariables(TestVariableEndpoint):
},
{
"action": "create",
- "variables": [
+ "entities": [
{
"key": "new_key",
"value": "new_value1",
@@ -812,7 +836,7 @@ class TestBulkVariables(TestVariableEndpoint):
},
{
"action": "update",
- "variables": [
+ "entities": [
{
"key": "nonexistent_var",
"value": "updated_value",
@@ -823,12 +847,12 @@ class TestBulkVariables(TestVariableEndpoint):
},
{
"action": "delete",
- "keys": ["dictionary_password"],
+ "entities": ["dictionary_password"],
"action_on_non_existence": "fail",
},
{
"action": "create",
- "variables": [
+ "entities": [
{
"key": "new_variable_key_1",
"value": "new_value1",
@@ -839,7 +863,7 @@ class TestBulkVariables(TestVariableEndpoint):
},
{
"action": "update",
- "variables": [
+ "entities": [
{
"key": "new_variable_key",
"value": "updated_value",