This is an automated email from the ASF dual-hosted git repository.
shahar1 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 740d8bda336 Fix airflowctl commands failing against older Airflow
servers (#70954)
740d8bda336 is described below
commit 740d8bda336e226d2f12257f2a4e84742453fd07
Author: Wei Lee <[email protected]>
AuthorDate: Sat Oct 31 06:15:49 2026 +0900
Fix airflowctl commands failing against older Airflow servers (#70954)
The airflowctl datamodels are generated from the newest Airflow API spec,
so a server on an older Airflow line legitimately omits fields those models
declare as required, and every command touching such a response aborts on a
validation error.
A field the server did not send is left unset rather than filled in with a
type default: a synthesized False, 0 or "" is indistinguishable from a value
the server really sent and would flow into --output json unnoticed. Fields
under a union of several models are the exception and stay strict, because
pydantic picks a union member by shape and an all-optional member would
absorb a payload meant for a different one. Responses that disagree with the
models for any other reason still fail, so a genuine defect on either side
keeps surfacing.
---
airflow-ctl/src/airflowctl/api/operations.py | 254 +++++++++------
.../tests/airflow_ctl/api/test_operations.py | 353 ++++++++++++++++++++-
2 files changed, 510 insertions(+), 97 deletions(-)
diff --git a/airflow-ctl/src/airflowctl/api/operations.py
b/airflow-ctl/src/airflowctl/api/operations.py
index 4e0c9f962ce..8c62ed3d98e 100644
--- a/airflow-ctl/src/airflowctl/api/operations.py
+++ b/airflow-ctl/src/airflowctl/api/operations.py
@@ -18,12 +18,17 @@
from __future__ import annotations
import datetime
+import functools
import json
-from typing import TYPE_CHECKING, Any, TypeVar, get_args
+import operator
+import sys
+import types
+import typing
+from typing import TYPE_CHECKING, Any, TypeVar, cast, get_args, get_origin
import httpx
-import structlog
-from pydantic import BaseModel, ValidationError
+from pydantic import BaseModel, RootModel, ValidationError, create_model
+from pydantic.fields import FieldInfo
from airflowctl.api.datamodels.auth_generated import LoginBody, LoginResponse
from airflowctl.api.datamodels.generated import (
@@ -95,7 +100,6 @@ from airflowctl.exceptions import
AirflowCtlConnectionException
if TYPE_CHECKING:
from airflowctl.api.client import Client
-log = structlog.get_logger(logger_name=__name__)
T = TypeVar("T", bound=BaseModel)
@@ -177,41 +181,106 @@ def _check_flag_and_exit_if_server_response_error(func):
return wrapped
-TYPE_DEFAULTS = {
- bool: False,
- int: 0,
- float: 0.0,
- str: "",
- list: [],
- dict: {},
-}
+# Cache of tolerant twins, keyed by the strict model they were built from.
+_TOLERANT_MODELS: dict[type[BaseModel], type[BaseModel]] = {}
-def get_field_default(annotation) -> Any:
+def _is_polymorphic_union(annotation: Any) -> bool:
+ """
+ Return whether ``annotation`` is a union with two or more non-``None``
members.
+
+ E.g. ``AssetExpressionAsset | AssetExpressionAlias``. ``X | None`` (one
real member) is a
+ plain optional and must still be rewritten. A union of two or more models
is different:
+ pydantic's smart-union picks whichever member the payload merely *fits*,
and every field
+ of a tolerant twin fits trivially (they are all optional). Put a twin in
such a union and
+ a payload shaped for member B can silently validate as an empty member A
instead of
+ raising -- the wrong-shape response is swallowed rather than surfaced. The
cost of
+ leaving such a union strict is that a field missing anywhere beneath it is
not
+ tolerated: it surfaces as a validation error rather than degrading to
``None``.
+ """
+ if get_origin(annotation) not in (typing.Union, types.UnionType):
+ return False
+ return len([arg for arg in get_args(annotation) if arg is not type(None)])
>= 2
+
+
+def _tolerate_missing(annotation: Any) -> Any:
+ """Rewrite ``annotation``, swapping any nested response model for its
tolerant twin."""
+ if _is_polymorphic_union(annotation):
+ return annotation
+ if isinstance(annotation, type) and issubclass(annotation, BaseModel):
+ return _tolerant_model(annotation)
+ origin = get_origin(annotation)
+ if origin is None:
+ return annotation
args = get_args(annotation)
- if args:
- non_none = [a for a in args if a is not type(None)]
- if non_none:
- return get_field_default(non_none[0])
- return TYPE_DEFAULTS.get(annotation, None)
-
-
-def fill_missing_fields(data: dict, model: type[BaseModel]) -> dict:
- for field_name, field_info in model.model_fields.items():
- annotation = field_info.annotation
- args = get_args(annotation)
- if field_name not in data and field_info.is_required():
- data[field_name] = get_field_default(annotation)
- elif field_name in data and isinstance(data[field_name], dict):
- if isinstance(annotation, type) and issubclass(annotation,
BaseModel):
- data[field_name] = fill_missing_fields(data[field_name],
annotation)
- elif field_name in data and isinstance(data[field_name], list) and
args:
- if isinstance(args[0], type) and issubclass(args[0], BaseModel):
- data[field_name] = [
- fill_missing_fields(item, args[0]) if isinstance(item,
dict) else item
- for item in data[field_name]
- ]
- return data
+ rewritten = [_tolerate_missing(arg) for arg in args]
+ if all(new_arg is old_arg for new_arg, old_arg in zip(rewritten, args)):
+ return annotation
+ if origin in (typing.Union, types.UnionType):
+ return functools.reduce(operator.or_, rewritten)
+ return origin[tuple(rewritten)]
+
+
+def _tolerant_model(model: type[BaseModel]) -> type[BaseModel]:
+ """
+ Build (or fetch the cached) tolerant twin of ``model``.
+
+ The twin is a ``pydantic.create_model`` subclass of ``model`` with every
required field
+ -- including ones nested inside it, recursively -- widened to ``X | None``
defaulting to
+ ``None``. A field an older server omits is then left honestly unset
instead of being
+ synthesized to a type default that would be indistinguishable from a real
value. Because
+ the twin subclasses the original model, ``isinstance(twin_instance,
model)`` still holds.
+
+ A ``RootModel`` wraps a single scalar rather than named fields, so "a
field is missing"
+ has no meaning for it; it is returned unchanged. Polymorphic unions (see
+ ``_is_polymorphic_union``) are never rewritten. The generated datamodels
contain mutually
+ self-referential schemas (``AssetExpressionAll`` <->
``AssetExpressionAny``), but every
+ path back to a model already being built currently runs through such a
union, so the
+ recursion terminates; the smoke test over all generated models guards this.
+ """
+ if model in _TOLERANT_MODELS:
+ return _TOLERANT_MODELS[model]
+ if issubclass(model, RootModel):
+ return model
+ overrides: dict[str, Any] = {}
+ for field_name, field in model.model_fields.items():
+ annotation = _tolerate_missing(field.annotation)
+ if field.is_required():
+ overrides[field_name] = (annotation | None,
FieldInfo.merge_field_infos(field, default=None))
+ elif annotation is not field.annotation:
+ overrides[field_name] = (annotation, field)
+ twin = create_model(f"Partial{model.__name__}", __base__=model,
**overrides) if overrides else model
+ _TOLERANT_MODELS[model] = twin
+ return twin
+
+
+def validate_response(content: bytes, data_model: type[T]) -> T:
+ """
+ Validate a server response, tolerating fields an older server does not
send.
+
+ The datamodels are generated from the newest Airflow API spec, so a server
on an
+ older Airflow line can legitimately omit fields the models declare as
required. Any
+ such field is left ``None`` -- never synthesized to a type default
(``False``, ``0``,
+ ``""``, ...), which would be indistinguishable from a real value the
server actually
+ sent. Any other validation error means the response genuinely disagrees
with the
+ model, which is a defect on one side or the other and must surface instead
of being
+ papered over.
+ """
+ try:
+ return data_model.model_validate_json(content)
+ except ValidationError as validation_error:
+ errors = validation_error.errors()
+ if any(error["type"] != "missing" for error in errors):
+ raise
+ tolerant = cast("T",
_tolerant_model(data_model).model_validate_json(content))
+ # stderr, not the logger: structlog is unconfigured here and would
write to stdout,
+ # corrupting ``--output json``.
+ fields = [".".join(str(part) for part in error["loc"]) for error in
errors]
+ sys.stderr.write(
+ f"Warning: {data_model.__name__} response omitted fields {fields};
leaving them unset (None). "
+ "The server is likely on an older Airflow line than these
datamodels.\n"
+ )
+ return tolerant
class BaseOperations:
@@ -242,17 +311,12 @@ class BaseOperations:
shared_params = {"limit": limit, **(params or {})}
- def safe_validate(content: bytes) -> BaseModel:
- try:
- return data_model.model_validate_json(content)
- except ValidationError:
- raw = fill_missing_fields(json.loads(content), data_model)
- return data_model.model_validate(raw)
-
self.response = self.client.get(path, params={**shared_params,
"offset": offset})
- first_pass = safe_validate(self.response.content)
+ first_pass = validate_response(self.response.content, data_model)
total_entries = first_pass.total_entries
- if total_entries < limit:
+ # An older server that omits total_entries leaves it None; treat that
as "no more
+ # pages" rather than raising on the comparison below.
+ if total_entries is None or total_entries < limit:
return first_pass
found_key = None
for key, value in first_pass.model_dump().items():
@@ -263,7 +327,7 @@ class BaseOperations:
offset = offset + limit
while offset < total_entries:
self.response = self.client.get(path, params={**shared_params,
"offset": offset})
- entry = safe_validate(self.response.content)
+ entry = validate_response(self.response.content, data_model)
offset = offset + limit
entry_list.extend(getattr(entry, found_key))
return data_model(**{found_key: entry_list, "total_entries":
total_entries})
@@ -290,12 +354,12 @@ class AssetsOperations(BaseOperations):
def get(self, asset_id: str) -> AssetResponse | ServerResponseError:
"""Get an asset from the API server."""
self.response = self.client.get(f"assets/{asset_id}")
- return AssetResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, AssetResponse)
def get_alias(self, asset_alias_id: str) -> AssetAliasResponse |
ServerResponseError:
"""Get an asset alias by its ID from the API server."""
self.response = self.client.get(f"assets/aliases/{asset_alias_id}")
- return AssetAliasResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, AssetAliasResponse)
def list(self) -> AssetCollectionResponse | ServerResponseError:
"""List all assets from the API server."""
@@ -315,29 +379,29 @@ class AssetsOperations(BaseOperations):
self.response = self.client.post(
"assets/events", json=asset_event_body.model_dump(mode="json",
exclude_none=True)
)
- return AssetEventResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, AssetEventResponse)
def materialize(self, asset_id: str) -> DAGRunResponse |
ServerResponseError:
"""Materialize an asset."""
self.response = self.client.post(f"assets/{asset_id}/materialize",
json={})
- return DAGRunResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, DAGRunResponse)
def get_queued_events(self, asset_id: str) ->
QueuedEventCollectionResponse | ServerResponseError:
"""Get queued events for an asset."""
self.response = self.client.get(f"assets/{asset_id}/queuedEvents")
- return
QueuedEventCollectionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content,
QueuedEventCollectionResponse)
def get_dag_queued_events(
self, dag_id: str, before: str
) -> QueuedEventCollectionResponse | ServerResponseError:
"""Get queued events for a dag."""
self.response = self.client.get(f"dags/{dag_id}/assets/queuedEvents",
params={"before": before})
- return
QueuedEventCollectionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content,
QueuedEventCollectionResponse)
def get_dag_queued_event(self, dag_id: str, asset_id: str) ->
QueuedEventResponse | ServerResponseError:
"""Get a queued event for a dag."""
self.response =
self.client.get(f"dags/{dag_id}/assets/{asset_id}/queuedEvents")
- return QueuedEventResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, QueuedEventResponse)
def delete_queued_events(self, asset_id: str) -> str | ServerResponseError:
"""Delete a queued event for an asset."""
@@ -363,7 +427,7 @@ class AssetsOperations(BaseOperations):
def get_state_store(self, asset_id: str, key: str) ->
AssetStateStoreResponse | ServerResponseError:
"""Get a single asset state store entry."""
self.response = self.client.get(f"assets/{asset_id}/state-store/{key}")
- return
AssetStateStoreResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content,
AssetStateStoreResponse)
def set_state_store(self, asset_id: str, key: str, value: str) -> str |
ServerResponseError:
"""Set an asset state store value. Creates or overwrites the key."""
@@ -393,7 +457,7 @@ class BackfillOperations(BaseOperations):
self.response = self.client.post(
"backfills", json=backfill.model_dump(mode="json",
exclude_none=True, exclude_defaults=True)
)
- return BackfillResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, BackfillResponse)
def create_dry_run(self, backfill: BackfillPostBody) -> BackfillResponse |
ServerResponseError:
"""Create a dry run backfill."""
@@ -401,12 +465,12 @@ class BackfillOperations(BaseOperations):
"backfills/dry_run",
json=backfill.model_dump(mode="json", exclude_none=True,
exclude_defaults=True),
)
- return BackfillResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, BackfillResponse)
def get(self, backfill_id: str) -> BackfillResponse | ServerResponseError:
"""Get a backfill."""
self.response = self.client.get(f"backfills/{backfill_id}")
- return BackfillResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, BackfillResponse)
def list(self, dag_id: str) -> BackfillCollectionResponse |
ServerResponseError:
"""List all backfills."""
@@ -416,17 +480,17 @@ class BackfillOperations(BaseOperations):
def pause(self, backfill_id: str) -> BackfillResponse |
ServerResponseError:
"""Pause a backfill."""
self.response = self.client.put(f"backfills/{backfill_id}/pause")
- return BackfillResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, BackfillResponse)
def unpause(self, backfill_id: str) -> BackfillResponse |
ServerResponseError:
"""Unpause a backfill."""
self.response = self.client.put(f"backfills/{backfill_id}/unpause")
- return BackfillResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, BackfillResponse)
def cancel(self, backfill_id: str) -> BackfillResponse |
ServerResponseError:
"""Cancel a backfill."""
self.response = self.client.put(f"backfills/{backfill_id}/cancel")
- return BackfillResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, BackfillResponse)
class ConfigOperations(BaseOperations):
@@ -435,12 +499,12 @@ class ConfigOperations(BaseOperations):
def get(self, section: str, option: str) -> Config | ServerResponseError:
"""Get a config from the API server."""
self.response =
self.client.get(f"/config/section/{section}/option/{option}")
- return Config.model_validate_json(self.response.content)
+ return validate_response(self.response.content, Config)
def list(self) -> Config | ServerResponseError:
"""List all configs from the API server."""
self.response = self.client.get("/config")
- return Config.model_validate_json(self.response.content)
+ return validate_response(self.response.content, Config)
class ConnectionsOperations(BaseOperations):
@@ -449,7 +513,7 @@ class ConnectionsOperations(BaseOperations):
def get(self, conn_id: str) -> ConnectionResponse | ServerResponseError:
"""Get a connection from the API server."""
self.response = self.client.get(f"connections/{conn_id}")
- return ConnectionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, ConnectionResponse)
def list(self) -> ConnectionCollectionResponse | ServerResponseError:
"""List all connections from the API server."""
@@ -463,14 +527,14 @@ class ConnectionsOperations(BaseOperations):
self.response = self.client.post(
"connections", json=connection.model_dump(mode="json",
by_alias=True, exclude_none=True)
)
- return ConnectionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, ConnectionResponse)
def bulk(self, connections: BulkBodyConnectionBody) -> BulkResponse |
ServerResponseError:
"""CRUD multiple connections."""
self.response = self.client.patch(
"connections", json=connections.model_dump(mode="json",
by_alias=True)
)
- return BulkResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, BulkResponse)
def create_defaults(self) -> None | ServerResponseError:
"""Create default connections."""
@@ -491,7 +555,7 @@ class ConnectionsOperations(BaseOperations):
f"connections/{connection.connection_id}",
json=connection.model_dump(mode="json", by_alias=True,
exclude_none=True),
)
- return ConnectionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, ConnectionResponse)
def test(
self,
@@ -502,7 +566,7 @@ class ConnectionsOperations(BaseOperations):
"connections/test",
json=connection.model_dump(mode="json", by_alias=True,
exclude_none=True),
)
- return
ConnectionTestResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, ConnectionTestResponse)
class DagsOperations(BaseOperations):
@@ -511,12 +575,12 @@ class DagsOperations(BaseOperations):
def get(self, dag_id: str) -> DAGResponse | ServerResponseError:
"""Get a Dag."""
self.response = self.client.get(f"dags/{dag_id}")
- return DAGResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, DAGResponse)
def get_details(self, dag_id: str) -> DAGDetailsResponse |
ServerResponseError:
"""Get a Dag details."""
self.response = self.client.get(f"dags/{dag_id}/details")
- return DAGDetailsResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, DAGDetailsResponse)
def get_tags(self) -> DAGTagCollectionResponse | ServerResponseError:
"""Get all Dag tags."""
@@ -528,7 +592,7 @@ class DagsOperations(BaseOperations):
def update(self, dag_id: str, dag_body: DAGPatchBody) -> DAGResponse |
ServerResponseError:
self.response = self.client.patch(f"dags/{dag_id}",
json=dag_body.model_dump(mode="json"))
- return DAGResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, DAGResponse)
def delete(self, dag_id: str) -> str | ServerResponseError:
self.client.delete(f"dags/{dag_id}")
@@ -536,18 +600,18 @@ class DagsOperations(BaseOperations):
def get_import_error(self, import_error_id: str) -> ImportErrorResponse |
ServerResponseError:
self.response = self.client.get(f"importErrors/{import_error_id}")
- return ImportErrorResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, ImportErrorResponse)
def list_import_errors(self) -> ImportErrorCollectionResponse |
ServerResponseError:
return super().execute_list(path="importErrors",
data_model=ImportErrorCollectionResponse)
def get_stats(self, dag_ids: list) -> DagStatsCollectionResponse |
ServerResponseError: # type: ignore
self.response = self.client.get("dagStats", params={"dag_ids":
dag_ids})
- return
DagStatsCollectionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content,
DagStatsCollectionResponse)
def get_version(self, dag_id: str, version_number: int) ->
DagVersionResponse | ServerResponseError:
self.response =
self.client.get(f"dags/{dag_id}/dagVersions/{version_number}")
- return DagVersionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, DagVersionResponse)
def list_version(self, dag_id: str) -> DAGVersionCollectionResponse |
ServerResponseError:
return super().execute_list(
@@ -566,7 +630,7 @@ class DagsOperations(BaseOperations):
self.response = self.client.post(
f"dags/{dag_id}/dagRuns",
json=trigger_dag_run.model_dump(mode="json", exclude_defaults=True)
)
- return DAGRunResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, DAGRunResponse)
class DagRunOperations(BaseOperations):
@@ -580,7 +644,7 @@ class DagRunOperations(BaseOperations):
f"/dags/{dag_id}/dagRuns/{dag_run_id}",
extensions={"airflowctl_suppress_error_log": suppress_error_log},
)
- return DAGRunResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, DAGRunResponse)
def list(
self,
@@ -642,7 +706,7 @@ class DagRunOperations(BaseOperations):
params=params,
extensions={"airflowctl_suppress_error_log": suppress_error_log},
)
- return
DAGRunCollectionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content,
DAGRunCollectionResponse)
def delete(self, dag_id: str, dag_run_id: str) -> str |
ServerResponseError:
"""Delete a Dag run."""
@@ -678,7 +742,7 @@ class JobsOperations(BaseOperations):
if limit is not None or offset is not None:
self.response = self.client.get("jobs", params=params)
- return
JobCollectionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content,
JobCollectionResponse)
return super().execute_list(path="jobs",
data_model=JobCollectionResponse, params=params)
@@ -689,7 +753,7 @@ class PoolsOperations(BaseOperations):
def get(self, pool_name: str) -> PoolResponse | ServerResponseError:
"""Get a pool."""
self.response = self.client.get(f"pools/{pool_name}")
- return PoolResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, PoolResponse)
def list(self) -> PoolCollectionResponse | ServerResponseError:
"""List all pools."""
@@ -698,12 +762,12 @@ class PoolsOperations(BaseOperations):
def create(self, pool: PoolBody) -> PoolResponse | ServerResponseError:
"""Create a pool."""
self.response = self.client.post("pools",
json=pool.model_dump(mode="json", exclude_none=True))
- return PoolResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, PoolResponse)
def bulk(self, pools: BulkBodyPoolBody) -> BulkResponse |
ServerResponseError:
"""CRUD multiple pools."""
self.response = self.client.patch("pools",
json=pools.model_dump(mode="json"))
- return BulkResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, BulkResponse)
def delete(self, pool: str) -> str | ServerResponseError:
"""Delete a pool."""
@@ -715,7 +779,7 @@ class PoolsOperations(BaseOperations):
self.response = self.client.patch(
f"pools/{pool_body.pool}", json=pool_body.model_dump(mode="json",
exclude_none=True)
)
- return PoolResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, PoolResponse)
class ProvidersOperations(BaseOperations):
@@ -724,7 +788,7 @@ class ProvidersOperations(BaseOperations):
def get(self, provider_name: str) -> ProviderDetailsResponse |
ServerResponseError:
"""Get detailed information for a provider."""
self.response = self.client.get(f"providers/{provider_name}")
- return
ProviderDetailsResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content,
ProviderDetailsResponse)
def list(self) -> ProviderCollectionResponse | ServerResponseError:
"""List all providers."""
@@ -759,7 +823,7 @@ class TaskInstancesOperations(BaseOperations):
path,
extensions={"airflowctl_suppress_error_log": suppress_error_log},
)
- return TaskInstanceResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, TaskInstanceResponse)
def get_dependencies(
self,
@@ -778,7 +842,7 @@ class TaskInstancesOperations(BaseOperations):
f"{path}/dependencies",
extensions={"airflowctl_suppress_error_log": suppress_error_log},
)
- return
TaskDependencyCollectionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content,
TaskDependencyCollectionResponse)
def list(self, dag_id: str, dag_run_id: str) ->
TaskInstanceCollectionResponse | ServerResponseError:
"""List task instances for a Dag run."""
@@ -805,7 +869,7 @@ class TaskInstancesOperations(BaseOperations):
f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/state-store/{key}",
params=_build_query_params(map_index=map_index),
)
- return
TaskStateStoreResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, TaskStateStoreResponse)
def set_state_store(
self,
@@ -880,7 +944,7 @@ class TasksOperations(BaseOperations):
f"dags/{dag_id}/clearTaskInstances",
json=clear_task_instances.model_dump(mode="json",
exclude_defaults=True),
)
- return
TaskInstanceCollectionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content,
TaskInstanceCollectionResponse)
class VariablesOperations(BaseOperations):
@@ -889,7 +953,7 @@ class VariablesOperations(BaseOperations):
def get(self, variable_key: str) -> VariableResponse | ServerResponseError:
"""Get a variable."""
self.response = self.client.get(f"variables/{variable_key}")
- return VariableResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, VariableResponse)
def list(self) -> VariableCollectionResponse | ServerResponseError:
"""List all variables."""
@@ -900,12 +964,12 @@ class VariablesOperations(BaseOperations):
self.response = self.client.post(
"variables", json=variable.model_dump(mode="json",
exclude_none=True)
)
- return VariableResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, VariableResponse)
def bulk(self, variables: BulkBodyVariableBody) -> BulkResponse |
ServerResponseError:
"""CRUD multiple variables."""
self.response = self.client.patch("variables",
json=variables.model_dump(mode="json"))
- return BulkResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, BulkResponse)
def delete(self, variable_key: str) -> str | ServerResponseError:
"""Delete a variable."""
@@ -917,7 +981,7 @@ class VariablesOperations(BaseOperations):
self.response = self.client.patch(
f"variables/{variable.key}", json=variable.model_dump(mode="json",
exclude_none=True)
)
- return VariableResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content, VariableResponse)
class VersionOperations(BaseOperations):
@@ -926,7 +990,7 @@ class VersionOperations(BaseOperations):
def get(self) -> VersionInfo | ServerResponseError:
"""Get the version."""
self.response = self.client.get("version")
- return VersionInfo.model_validate_json(self.response.content)
+ return validate_response(self.response.content, VersionInfo)
class XComOperations(BaseOperations):
@@ -948,7 +1012,7 @@ class XComOperations(BaseOperations):
f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries/{key}",
params=params,
)
- return XComResponseNative.model_validate_json(self.response.content)
+ return validate_response(self.response.content, XComResponseNative)
def list(
self,
@@ -990,7 +1054,7 @@ class XComOperations(BaseOperations):
f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries",
json=body.model_dump(mode="json", exclude_unset=True,
exclude_none=True),
)
- return XComResponseNative.model_validate_json(self.response.content)
+ return validate_response(self.response.content, XComResponseNative)
def edit(
self,
@@ -1012,7 +1076,7 @@ class XComOperations(BaseOperations):
f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries/{key}",
json=body.model_dump(mode="json", exclude_unset=True,
exclude_none=True),
)
- return XComResponseNative.model_validate_json(self.response.content)
+ return validate_response(self.response.content, XComResponseNative)
def delete(
self,
@@ -1043,4 +1107,4 @@ class PluginsOperations(BaseOperations):
def list_import_errors(self) -> PluginImportErrorCollectionResponse |
ServerResponseError:
"""List plugin import errors from the API server."""
self.response = self.client.get("plugins/importErrors")
- return
PluginImportErrorCollectionResponse.model_validate_json(self.response.content)
+ return validate_response(self.response.content,
PluginImportErrorCollectionResponse)
diff --git a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
index 26a02351abc..692403dfc8c 100644
--- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
+++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
@@ -26,8 +26,9 @@ from unittest.mock import Mock
import httpx
import pytest
-from pydantic import BaseModel
+from pydantic import BaseModel, RootModel, ValidationError
+import airflowctl.api.datamodels.generated as generated_datamodels
from airflowctl.api.client import Client, ClientKind
from airflowctl.api.datamodels.auth_generated import LoginBody, LoginResponse
from airflowctl.api.datamodels.generated import (
@@ -116,7 +117,7 @@ from airflowctl.api.datamodels.generated import (
XComResponse,
XComResponseNative,
)
-from airflowctl.api.operations import BaseOperations, _build_query_params
+from airflowctl.api.operations import BaseOperations, _build_query_params,
_tolerant_model
from airflowctl.exceptions import AirflowCtlConnectionException
if TYPE_CHECKING:
@@ -1345,6 +1346,93 @@ class TestDagOperations:
response = client.dags.get_details("dag_id")
assert response == self.dag_details_response
+ @pytest.mark.parametrize(
+ ("operation", "path", "payload_attr"),
+ [
+ (lambda dags, patch_body: dags.get("dag_id"),
"/api/v2/dags/dag_id", "dag_response"),
+ (
+ lambda dags, patch_body: dags.get_details("dag_id"),
+ "/api/v2/dags/dag_id/details",
+ "dag_details_response",
+ ),
+ (
+ lambda dags, patch_body: dags.update(dag_id="dag_id",
dag_body=patch_body),
+ "/api/v2/dags/dag_id",
+ "dag_response",
+ ),
+ ],
+ ids=["get", "get_details", "update"],
+ )
+ @pytest.mark.parametrize(
+ "omitted_field",
+ ["is_backfillable", "timetable_periodic", "owners", "tags"],
+ )
+ def test_single_object_response_tolerates_field_an_older_server_omits(
+ self, operation, path, payload_attr, omitted_field
+ ):
+ """A server on an older Airflow line omits fields the generated models
declare as required.
+
+ The omitted field must be left honestly ``None`` rather than filled
with a
+ synthesized type default that would be indistinguishable from a real
value.
+ """
+ payload = json.loads(getattr(self, payload_attr).model_dump_json())
+ del payload[omitted_field]
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == path
+ return httpx.Response(200, json=payload)
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = operation(client.dags, self.dag_patch_body)
+
+ assert response.dag_id == self.dag_id
+ assert getattr(response, omitted_field) is None
+
+ def
test_collection_response_tolerates_field_an_older_server_omits_from_an_item(self):
+ payload = json.loads(self.dag_collection_response.model_dump_json())
+ del payload["dags"][0]["owners"]
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == "/api/v2/dags"
+ return httpx.Response(200, json=payload)
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.dags.list()
+
+ assert response.dags[0].owners is None
+
+ def
test_single_object_response_warns_on_stderr_about_the_fields_it_leaves_unset(self,
capsys):
+ payload = json.loads(self.dag_response.model_dump_json())
+ del payload["is_backfillable"]
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(200, json=payload)
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ client.dags.get("dag_id")
+
+ captured = capsys.readouterr()
+ assert captured.out == ""
+ assert "DAGResponse" in captured.err
+ assert "is_backfillable" in captured.err
+
+ def
test_single_object_response_raises_when_a_present_field_mismatches(self):
+ payload = json.loads(self.dag_response.model_dump_json())
+ del payload["is_backfillable"]
+ payload["max_active_tasks"] = "not-an-int"
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(200, json=payload)
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ with pytest.raises(ValidationError) as exc_info:
+ client.dags.get("dag_id")
+
+ assert {error["loc"] for error in exc_info.value.errors()} == {
+ ("is_backfillable",),
+ ("max_active_tasks",),
+ }
+
def test_get_tags(self):
def handle_request(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/api/v2/dagTags"
@@ -1699,6 +1787,267 @@ class TestDagRunOperations:
assert response == self.dag_run_id
+class TestValidateResponseTolerance:
+ """
+ Cross-cutting coverage for ``validate_response``'s tolerant-twin fallback.
+
+ These target failure modes the old type-defaults implementation could not
handle at
+ all: it had no default for a required enum or datetime field, so it
produced ``None``
+ for them anyway, and the second (real) ``model_validate`` call then raised
because
+ ``None`` is not a valid enum member or datetime -- the very omission the
fallback was
+ supposed to tolerate blew up one call later instead.
+ """
+
+ dag_id = "dag_id"
+ dag_run_id = "dag_run_id"
+
+ def _dag_run_payload(self, dag_run_id: str | None = None) -> dict:
+ return json.loads(
+ DAGRunResponse(
+ dag_display_name=dag_run_id or self.dag_run_id,
+ dag_run_id=dag_run_id or self.dag_run_id,
+ dag_id=self.dag_id,
+ logical_date=datetime.datetime(2025, 1, 1),
+ queued_at=datetime.datetime(2025, 1, 1),
+ start_date=datetime.datetime(2025, 1, 1),
+ end_date=datetime.datetime(2025, 1, 1),
+ data_interval_start=datetime.datetime(2025, 1, 1),
+ data_interval_end=datetime.datetime(2025, 1, 1),
+ last_scheduling_decision=datetime.datetime(2025, 1, 1),
+ run_after=datetime.datetime(2025, 1, 1),
+ run_type=DagRunType.MANUAL,
+ state=DagRunState.RUNNING,
+ triggered_by=DagRunTriggeredByType.UI,
+ conf={},
+ note=None,
+ dag_versions=[],
+ duration=None,
+ triggering_user_name=None,
+ bundle_version=None,
+ partition_key=None,
+ partition_date=None,
+ ).model_dump_json()
+ )
+
+ def test_missing_required_enum_field_is_left_none(self):
+ """``DAGRunResponse.state`` is a required enum with no default -- the
old
+ type-defaults fallback had no entry for it at all and produced
``None``, which the
+ second ``model_validate`` call then rejected as not being a valid enum
member."""
+ payload = self._dag_run_payload()
+ del payload["state"]
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(200, json=payload)
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.dag_runs.get(dag_id=self.dag_id,
dag_run_id=self.dag_run_id)
+
+ assert response.state is None
+
+ def test_missing_required_datetime_field_is_left_none(self):
+ """``DAGRunResponse.run_after`` is a required datetime with no
default."""
+ payload = self._dag_run_payload()
+ del payload["run_after"]
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(200, json=payload)
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.dag_runs.get(dag_id=self.dag_id,
dag_run_id=self.dag_run_id)
+
+ assert response.run_after is None
+
+ def
test_tolerant_twin_instance_is_still_isinstance_of_the_original_model(self):
+ payload = self._dag_run_payload()
+ del payload["state"]
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(200, json=payload)
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.dag_runs.get(dag_id=self.dag_id,
dag_run_id=self.dag_run_id)
+
+ assert isinstance(response, DAGRunResponse)
+
+ def _dag_details_payload(self) -> dict:
+ return json.loads(
+ DAGDetailsResponse(
+ dag_id=self.dag_id,
+ dag_display_name=self.dag_id,
+ is_paused=False,
+ is_stale=False,
+ last_parsed_time=None,
+ last_parse_duration=None,
+ last_expired=None,
+ bundle_name=None,
+ bundle_version=None,
+ relative_fileloc=None,
+ fileloc=None,
+ description=None,
+ timetable_summary=None,
+ timetable_description=None,
+ timetable_partitioned=False,
+ timetable_periodic=True,
+ tags=[],
+ max_active_tasks=1,
+ max_active_runs=None,
+ max_consecutive_failed_dag_runs=1,
+ has_task_concurrency_limits=True,
+ has_import_errors=False,
+ next_dagrun_logical_date=None,
+ next_dagrun_data_interval_start=None,
+ next_dagrun_data_interval_end=None,
+ next_dagrun_run_after=None,
+ allowed_run_types=None,
+ owners=[],
+ catchup=False,
+ dag_run_timeout=None,
+ asset_expression=None,
+ doc_md=None,
+ start_date=None,
+ end_date=None,
+ is_paused_upon_creation=False,
+ params={},
+ render_template_as_native_obj=True,
+ template_search_path=None,
+ timezone=None,
+ last_parsed=None,
+ default_args=None,
+ is_backfillable=True,
+ file_token="file_token",
+ concurrency=1,
+ latest_dag_version=None,
+ ).model_dump_json()
+ )
+
+ def
test_polymorphic_union_member_missing_a_required_field_still_raises(self,
capsys):
+ """
+ Regression test for a real defect a prior version of the tolerant-twin
builder had:
+ rewriting every member of a model-valued union (here
+ ``DAGDetailsResponse.asset_expression``) to its all-optional twin let
pydantic's
+ smart-union match a payload shaped for one member -- but missing that
member's
+ required subfields -- anyway, silently validating to an empty node
instead of
+ raising. A polymorphic union (two or more non-``None`` model members)
must never be
+ rewritten, so no member can swallow a malformed payload this way; only
the field
+ being missing entirely may still fall back to ``None``.
+ """
+ payload = self._dag_details_payload()
+ del payload["timetable_periodic"] # an unrelated field an older
server omits
+ # Shaped like an AssetExpressionAsset leaf but missing its required
"uri" and
+ # "group" -- no union member actually matches this payload.
+ payload["asset_expression"] = {"asset": {"name": "n"}}
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ return httpx.Response(200, json=payload)
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ with pytest.raises(ValidationError):
+ client.dags.get_details(dag_id=self.dag_id)
+
+ assert capsys.readouterr().err == ""
+
+ def
test_execute_list_reassembly_tolerates_a_later_page_missing_a_field(self):
+ """
+ ``BaseOperations.execute_list`` reassembles pages by constructing a
fresh
+ ``data_model`` and re-validating it. If a later page's item needed the
tolerant
+ fallback for a required, non-Optional field (e.g. a missing ``bool``),
that
+ reassembly must not re-raise on the very field it just tolerated.
+ """
+
+ class _StrictItem(BaseModel):
+ name: str
+ is_special: bool
+
+ class _StrictItemCollection(BaseModel):
+ items: list[_StrictItem]
+ total_entries: int
+
+ first_page = {"items": [{"name": "a", "is_special": True}],
"total_entries": 2}
+ second_page_item = {"name": "b"} # is_special omitted, as an older
server would.
+
+ pages = [
+ Mock(spec=httpx.Response, content=json.dumps(first_page)),
+ Mock(spec=httpx.Response, content=json.dumps({"items":
[second_page_item], "total_entries": 2})),
+ ]
+ mock_client = Mock(spec=Client)
+ mock_client.get.side_effect = pages
+ base_operation = BaseOperations(client=mock_client)
+
+ response = base_operation.execute_list(path="items",
data_model=_StrictItemCollection, limit=1)
+
+ assert [item.name for item in response.items] == ["a", "b"]
+ assert response.items[0].is_special is True
+ assert response.items[1].is_special is None
+
+ def
test_execute_list_returns_the_first_page_when_total_entries_is_omitted(self):
+ class _Item(BaseModel):
+ name: str
+
+ class _ItemCollection(BaseModel):
+ items: list[_Item]
+ total_entries: int
+
+ mock_client = Mock(spec=Client)
+ mock_client.get.return_value = Mock(
+ spec=httpx.Response, content=json.dumps({"items": [{"name": "a"}]})
+ )
+ base_operation = BaseOperations(client=mock_client)
+
+ response = base_operation.execute_list(path="items",
data_model=_ItemCollection, limit=1)
+
+ assert [item.name for item in response.items] == ["a"]
+ assert response.total_entries is None
+ mock_client.get.assert_called_once()
+
+
+class TestTolerantModelBuilderCoversAllGeneratedModels:
+ """
+ Structural smoke test for the tolerant-twin builder itself.
+
+ Every non-``RootModel`` model in the generated datamodels must produce a
tolerant twin
+ that validates an empty payload, proving the builder neither explodes nor
infinitely
+ recurses on any of them -- including the two that are mutually
self-referential
+ (``AssetExpressionAll`` <-> ``AssetExpressionAny``), which no longer needs
a cycle guard
+ now that a polymorphic union (the only path back into either model) is
never rewritten.
+ """
+
+ def
test_every_generated_model_builds_a_tolerant_twin_that_validates_an_empty_payload(self):
+ models = [
+ obj
+ for obj in vars(generated_datamodels).values()
+ if isinstance(obj, type)
+ and issubclass(obj, BaseModel)
+ and not issubclass(obj, RootModel)
+ and obj.__module__ == generated_datamodels.__name__
+ ]
+ # A generous floor, not the exact count: guards against a filter bug
(e.g. the
+ # RootModel exclusion swallowing everything) making this loop -- and
the assertion
+ # below -- vacuously pass over zero models.
+ assert len(models) > 150
+
+ failures: dict[str, str] = {}
+ for model in models:
+ try:
+ twin = _tolerant_model(model)
+ twin.model_validate({})
+ except Exception as exc:
+ failures[model.__name__] = repr(exc)
+
+ assert failures == {}
+
+ def test_tolerant_twin_preserves_the_original_field_alias(self):
+ """
+ ``FieldInfo.merge_field_infos`` is a semi-public pydantic API; if a
future pydantic
+ release changes how it copies metadata, this must fail loudly here
rather than
+ silently drop the alias and only surface when a real server payload
keyed by the
+ alias (``ConnectionsOperations`` reads/writes ``schema`` on the wire,
see
+ ``ConnectionResponse.schema_``) fails to bind to the tolerant twin.
+ """
+ twin = _tolerant_model(ConnectionResponse)
+
+ assert twin.model_fields["schema_"].alias == "schema"
+
+
class TestJobsOperations:
job_response = JobResponse(
id=1,
