jason810496 commented on code in PR #73441: URL: https://github.com/apache/airflow/pull/73441#discussion_r4146315226
########## scripts/ci/lang_sdk_serialization/compare.py: ########## @@ -0,0 +1,233 @@ +# 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. +r""" +Check that a language SDK serializes Dags the way Airflow does. + +Runs the SDK's serializer and serialize_python.py over test_dags.yaml, each writing its serialization +to a JSON file in a temporary directory. The Python side also takes the SDK's output as Airflow +receives it, filling in the fields the SDK leaves to Airflow's config, and loads it through Airflow's +deserializer. The two are then compared field by field, so this fails both when the serializers drift +apart and when Airflow cannot read what the SDK writes. + +The SDK's serializer is the command after ``--``. It runs from the repository root, with the paths of +test_dags.yaml and of the JSON file to write appended, and writes each Dag as +``DagSerialization.to_dict`` would, keyed by Dag id. The Python side needs ``uv``:: + + python3 scripts/ci/lang_sdk_serialization/compare.py --sdk typescript -- \ + pnpm --dir ts-sdk exec tsx tests/conformance/serialize_typescript.ts + +On a failure the files are kept, and their directory is printed. +""" + +from __future__ import annotations + +import argparse +import json +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path +from typing import Any + +HERE = Path(__file__).resolve().parent +REPO_ROOT = HERE.parents[2] +TEST_DAGS = HERE / "test_dags.yaml" +SCHEMA = REPO_ROOT / "airflow-core" / "src" / "airflow" / "serialization" / "schema.json" + +# Name the file a Dag was declared in: Python's own Dag file, or the SDK's bundle. +DAG_KEYS_NOT_COMPARED = frozenset({"fileloc", "relative_fileloc", "_processor_dags_folder"}) + +TASK_KEYS_NOT_COMPARED = frozenset( + { + # Operator identity: Python names the Python class that ran, a language SDK a fixed pair. + # Neither is ever imported on the Airflow side. + "task_type", + "_task_module", + # The SDK's marker for a task in its language, which Python has no equivalent for. + "language", + # Python bookkeeping for mapped tasks and retry policies, neither of which a language SDK's + # Dag can declare. + "_needs_expansion", + "has_retry_policy", + # The stub contract every language SDK task carries, which the Python operator built here + # has no counterpart for. Loading the SDK's output through Airflow checks them. + "is_stub", + "_arg_bindings", Review Comment: We're kind of "overriding" the purpose of `_arg_bindings` here. On Python side, `_arg_bindings` will store the arguments if the `@task.stub`. However, on the SDK side, for both `TaskHandler` and the ordinary task, we will store the arguments of the actual function. So they really represent different entity. Btw, the Dag processing side will support the validation of `@task.stub` and the `TaskHandler` based on https://github.com/apache/airflow/pull/71929 design. -- 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]
