Skip to content

Commit 8e60132

Browse files
committed
TS SDK: serialize a native Dag to Dag JSON
1 parent 385d276 commit 8e60132

16 files changed

Lines changed: 3133 additions & 6 deletions

File tree

‎.gitignore‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -340,3 +340,8 @@ dev/registry/providers.json
340340
# absolute paths written by sandbox-add-project-root.sh; per
341341
# https://github.com/apache/magpie/issues/197).
342342
/.claude/settings.local.json
343+
344+
# Generated by ts-sdk/tests/conformance/serialize_typescript.ts, to hand to
345+
# serialize_python.py --verify. The recorded Python side is committed; this one
346+
# is rebuilt on demand.
347+
ts-sdk/tests/conformance/serialized_typescript.json

‎.pre-commit-config.yaml‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -318,6 +318,18 @@ repos:
318318
(?x)
319319
^airflow-core/src/airflow/serialization/schema\.json$|
320320
^ts-sdk/schema/dag-schema\.json$
321+
- id: check-ts-sdk-conformance-recording
322+
name: Check the recorded Python Dag serialization the TS SDK is compared against
323+
description: "Re-run Airflow's serializer over the conformance fixtures and fail if the recording drifted"
324+
entry: ./ts-sdk/scripts/ci/prek/check_conformance_recording.py
325+
language: python
326+
pass_filenames: false
327+
require_serial: true
328+
files: >
329+
(?x)
330+
^airflow-core/src/airflow/serialization/serialized_objects\.py$|
331+
^airflow-core/src/airflow/serialization/schema\.json$|
332+
^ts-sdk/tests/conformance/.*$
321333
- id: check-go-version-in-sync
322334
name: Check Go toolchain version is consistent across build files
323335
entry: ./scripts/ci/prek/check_go_version_in_sync.py

‎airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -346,6 +346,29 @@ an id of either.
346346

347347
.. warning:: A cycle is not allowed. The Dag is rejected when it is read, naming the tasks on the cycle.
348348

349+
Serialization
350+
~~~~~~~~~~~~~
351+
352+
A native Dag serializes into the same Dag JSON a Python Dag produces, so the scheduler reads it
353+
without knowing which language declared it.
354+
355+
``schedule`` accepts what maps to a stock timetable: unset, ``@once``, ``@continuous``, or a cron
356+
expression. A cron preset such as ``@daily`` is recorded as the expression it stands for. Anything
357+
else names a Python object a TypeScript bundle cannot point at, and is rejected.
358+
359+
Every task of a native Dag runs on the Node coordinator, so it needs the queue the deployment routes
360+
there. Set it once on the Dag and each task inherits it:
361+
362+
.. code-block:: typescript
363+
364+
const dag = new Dag("ts_etl", { schedule: "@daily", queue: "typescript" });
365+
366+
// ...and one task that needs its own.
367+
dag.task("heavy", heavyHandler, { queue: "typescript_large" })();
368+
369+
``queue`` on a task wins over the Dag's. See :ref:`typescript-sdk/coordinator-config` for the
370+
``queue_to_coordinator`` entry that sends that queue to the coordinator.
371+
349372
``new Dag`` and ``dag.task`` both take a trailing spec of Airflow options:
350373
``{ schedule: "@daily", tags: ["etl"] }`` for the Dag, ``{ retries: 2, retryDelay: 30 }`` for a task.
351374

‎ts-sdk/.prettierignore‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,3 +21,9 @@ node_modules/
2121

2222
# Byte-exact bundle artifacts: reformatting one invalidates its embedded digests.
2323
tests/cli/fixtures/*.min.mjs
24+
25+
# Written by the conformance harness, whose two sides have to agree byte for
26+
# byte: reformatting either would make serialize_python.py --check fail against
27+
# output it had just produced.
28+
tests/conformance/serialized_python.json
29+
tests/conformance/serialized_typescript.json
Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
#!/usr/bin/env python3
2+
# Licensed to the Apache Software Foundation (ASF) under one
3+
# or more contributor license agreements. See the NOTICE file
4+
# distributed with this work for additional information
5+
# regarding copyright ownership. The ASF licenses this file
6+
# to you under the Apache License, Version 2.0 (the
7+
# "License"); you may not use this file except in compliance
8+
# with the License. You may obtain a copy of the License at
9+
#
10+
# http://www.apache.org/licenses/LICENSE-2.0
11+
#
12+
# Unless required by applicable law or agreed to in writing,
13+
# software distributed under the License is distributed on an
14+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
# KIND, either express or implied. See the License for the
16+
# specific language governing permissions and limitations
17+
"""
18+
Keep the recorded Python serialization output the conformance suite compares against honest.
19+
20+
``ts-sdk/tests/conformance/serialized_python.json`` is what Airflow's own serializer
21+
produced for the fixtures in ``test_dags.json``. The TypeScript conformance suite diffs
22+
its output against that recording, so the recording going stale would leave the suite
23+
passing against an Airflow that no longer serializes the same way.
24+
"""
25+
26+
from __future__ import annotations
27+
28+
import subprocess
29+
import sys
30+
from pathlib import Path
31+
32+
sys.path.insert(0, str(Path(__file__).resolve().parents[4] / "scripts" / "ci" / "prek"))
33+
34+
from common_prek_utils import AIRFLOW_ROOT_PATH, console
35+
36+
if __name__ not in ("__main__", "__mp_main__"):
37+
raise SystemExit(
38+
"This file is intended to be executed as an executable program. You cannot use it as a module."
39+
f"To run this script, run the ./{__file__} command"
40+
)
41+
42+
SCRIPT = Path("ts-sdk/tests/conformance/serialize_python.py")
43+
44+
if __name__ == "__main__":
45+
result = subprocess.run(
46+
["uv", "run", "--project", "airflow-core", "python", str(SCRIPT), "--check"],
47+
cwd=AIRFLOW_ROOT_PATH,
48+
check=False,
49+
capture_output=True,
50+
text=True,
51+
)
52+
if result.returncode == 0:
53+
sys.exit(0)
54+
message = (
55+
f"{SCRIPT} --check failed, so the recorded Python output no longer matches what "
56+
"Airflow's serializer produces.\n"
57+
f"Re-record it with `uv run --project airflow-core python {SCRIPT}` and review the diff: "
58+
"a change there is a change in the parity contract.\n\n"
59+
f"{result.stdout}{result.stderr}"
60+
)
61+
if console:
62+
console.print(f"[yellow]{message}[/]")
63+
else:
64+
print(message, file=sys.stderr)
65+
raise SystemExit(result.returncode)

0 commit comments

Comments
 (0)