This is an automated email from the ASF dual-hosted git repository.
dabla 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 bbf2c5b1c47 Fix Lang-SDK decoder typing for pydantic 2.14 TypeForm
signature (#74628)
bbf2c5b1c47 is described below
commit bbf2c5b1c476ffb66267b0c1b9b8aa0f22d02b1f
Author: Dev-iL <[email protected]>
AuthorDate: Thu Oct 15 17:24:49 2026 +0300
Fix Lang-SDK decoder typing for pydantic 2.14 TypeForm signature (#74628)
pydantic 2.14 types TypeAdapter's argument as TypeForm[T], so mypy
rejects the Lang-SDK decoder built from get_args(ToManager)[0], which
is not a static type expression.
Name the ToManager message union ToManagerMessage and build the
Lang-SDK message type LangSDKToManager from it; the runtime union is
the same object as before. Declare each concrete processor's decoder
with its own precise TypeAdapter type instead of on
BaseDagFileProcessorProcess: TypeAdapter is invariant, so the Lang-SDK
decoder, which accepts a superset of ToManager, cannot override a base
TypeAdapter[ToManager].
---
airflow-core/src/airflow/dag_processing/lang_sdk_processor.py | 11 ++++++-----
airflow-core/src/airflow/dag_processing/processor.py | 11 ++++++-----
2 files changed, 12 insertions(+), 10 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py
b/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py
index 9a0998dd051..9e753e61059 100644
--- a/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py
+++ b/airflow-core/src/airflow/dag_processing/lang_sdk_processor.py
@@ -26,7 +26,7 @@ import time
from contextlib import suppress
from pathlib import Path
from socket import socket
-from typing import TYPE_CHECKING, Annotated, ClassVar, Literal, cast, get_args
+from typing import TYPE_CHECKING, Annotated, ClassVar, Literal, cast
import attrs
import msgspec
@@ -42,7 +42,7 @@ from airflow.dag_processing.processor import (
BaseDagFileProcessorProcess,
DagFileParseRequest,
DagFileParsingResult,
- ToManager,
+ ToManagerMessage,
)
from airflow.exceptions import DeserializationError
from airflow.models.pool import Pool
@@ -101,6 +101,9 @@ class LangSDKRuntimeSchemaVersion(BaseModel):
type: Literal["LangSDKRuntimeSchemaVersion"] =
"LangSDKRuntimeSchemaVersion"
+LangSDKToManager = Annotated[LangSDKRuntimeSchemaVersion | ToManagerMessage,
Field(discriminator="type")]
+
+
def _get_import_timeout(path: str) -> float | None:
"""Return the ``get_dagbag_import_timeout`` policy's timeout for *path*;
``None`` means none."""
timeout = settings.get_dagbag_import_timeout(path)
@@ -165,9 +168,7 @@ class
LangSDKDagFileProcessorProcess(BaseDagFileProcessorProcess):
``DagFileParseRequest`` itself, so the request is sent once it has
connected.
"""
- decoder = TypeAdapter(
- Annotated[LangSDKRuntimeSchemaVersion | get_args(ToManager)[0],
Field(discriminator="type")]
- )
+ decoder: ClassVar[TypeAdapter[LangSDKToManager]] =
TypeAdapter[LangSDKToManager](LangSDKToManager)
_listeners: dict[_Channel, socket]
_parse_request: DagFileParseRequest
diff --git a/airflow-core/src/airflow/dag_processing/processor.py
b/airflow-core/src/airflow/dag_processing/processor.py
index e6f3f422a0a..9530d7be8c9 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -135,7 +135,7 @@ class DagFileParsingResult(BaseModel):
type: Literal["DagFileParsingResult"] = "DagFileParsingResult"
-ToManager = Annotated[
+ToManagerMessage = (
DagFileParsingResult
| GetConnection
| GetVariable
@@ -151,9 +151,10 @@ ToManager = Annotated[
| GetXComCount
| GetXComSequenceItem
| GetXComSequenceSlice
- | MaskSecret,
- Field(discriminator="type"),
-]
+ | MaskSecret
+)
+
+ToManager = Annotated[ToManagerMessage, Field(discriminator="type")]
ToDagProcessor = Annotated[
DagFileParseRequest
@@ -586,7 +587,6 @@ class BaseDagFileProcessorProcess(WatchedSubprocess,
LoggingMixin):
)
"""The file's parse log, which the process closes. Without one, the output
is discarded."""
parsing_result: DagFileParsingResult | None = None
- decoder: ClassVar[TypeAdapter[ToManager]] =
TypeAdapter[ToManager](ToManager)
had_callbacks: bool = False # Track if this process was started with
callbacks to prevent stale DAG detection false positives
client: Client | None = None
@@ -711,6 +711,7 @@ class DagFileProcessorProcess(BaseDagFileProcessorProcess):
"""
client: Client
+ decoder: ClassVar[TypeAdapter[ToManager]] =
TypeAdapter[ToManager](ToManager)
_request_handlers: ClassVar[dict[type[BaseModel],
RequestHandler[DagFileProcessorProcess]]] = {
**BaseDagFileProcessorProcess._common_request_handlers,
