Seraph-x commented on code in PR #75503:
URL: https://github.com/apache/airflow/pull/75503#discussion_r4422457640
##########
task-sdk/src/airflow/sdk/execution_time/schema/__init__.py:
##########
@@ -91,12 +91,15 @@ def registered_models_by_name() -> dict[str,
type[BaseModel]]:
single head class, so a name clash is a programmer error that must
surface immediately rather than silently picking a winner.
"""
- from airflow.dag_processing.processor import ToDagProcessor, ToManager
+ from airflow.dag_processing.lang_sdk_processor import DagFileParseRequest
as LangSDKDagFileParseRequest
+ from airflow.dag_processing.processor import DagFileParseRequest,
ToDagProcessor, ToManager
from airflow.sdk.execution_time.comms import ToSupervisor, ToTask
by_name: dict[str, type[BaseModel]] = {}
for source in (ToTask, ToSupervisor, ToManager, ToDagProcessor):
- for model in _iter_model_types(source):
+ for member in _iter_model_types(source):
+ # The wire registry describes language runtimes, not the Python
parser's request.
+ model = LangSDKDagFileParseRequest if member is
DagFileParseRequest else member
Review Comment:
Reworked this so the registry walks a language-SDK-specific
`LangSDKToDagProcessor` union. The shared response members live in
`ToDagProcessorResponse`, which both the Python and language-SDK unions reuse.
The registry loop is now generic and contains no request-specific substitution.
Validation:
- 60 Task SDK schema and migration tests passed
- Dag processor message-union test passed
- Supervisor schema snapshot and version checks passed
---
Drafted-by: OpenAI Codex; reviewed by @Seraph-x before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
