From 483d8aa271e207e9242a4d9d9390649565fe00b7 Mon Sep 17 00:00:00 2001 From: henry3260 Date: Thu, 1 Oct 2026 23:52:11 +0800 Subject: [PATCH 1/2] Go SDK: vendor the schemas it generates from MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit go-sdk reached across the monorepo for both schemas it generates from — airflow-core's Dag serialization schema and task-sdk's supervisor wire-schema snapshot — so a generated file was only explainable from a directory the published Go module does not contain, and every change to a Python schema was a change that had to carry regenerated Go with it, from whoever happened to be editing the Python. Both are now vendored under go-sdk/schema/, the way ts-sdk and java-sdk already vendor theirs, and the generators read the copies. That splits one question into two: the new sync-go-sdk-schemas hook copies an original over its copy when they differ and fails, since copying is mechanical and needs neither Go nor the network; and check-go-sdk-generated-drift regenerates from the copies, since what to do about a new schema construct is a decision. Regenerating from the copies produces byte-identical output, and the two Go tests that read a schema now read the copy, so they no longer skip or break outside the monorepo. --- .pre-commit-config.yaml | 17 +- go-sdk/README.md | 38 +- go-sdk/airflow/spec.go | 2 +- go-sdk/internal/genspec/main.go | 9 +- go-sdk/internal/genspec/normalize_test.go | 5 +- go-sdk/pkg/execution/genmodels/gen.go | 13 +- go-sdk/pkg/execution/messages_test.go | 23 +- go-sdk/schema/dag-schema.json | 474 ++ go-sdk/schema/supervisor-schema.json | 5292 +++++++++++++++++ .../ci/prek/check_go_sdk_generated_drift.py | 25 +- scripts/ci/prek/sync_go_sdk_schemas.py | 134 + .../prek/test_check_go_sdk_generated_drift.py | 13 +- .../tests/ci/prek/test_sync_go_sdk_schemas.py | 99 + 13 files changed, 6089 insertions(+), 55 deletions(-) create mode 100644 go-sdk/schema/dag-schema.json create mode 100644 go-sdk/schema/supervisor-schema.json create mode 100755 scripts/ci/prek/sync_go_sdk_schemas.py create mode 100644 scripts/tests/ci/prek/test_sync_go_sdk_schemas.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 1cfb2b430ea27..60a2bb945ffa0 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -348,6 +348,18 @@ repos: # prek run sync-java-sdk-dag-schema --hook-stage manual stages: ['manual'] files: ^java-sdk/sdk/schema/dag-schema\.json$ + - id: sync-go-sdk-schemas + name: Sync Go SDK vendored schemas with the distributions that own them + description: "Copy airflow-core's Dag schema and task-sdk's supervisor schema when go-sdk's vendored copies drift" + entry: ./scripts/ci/prek/sync_go_sdk_schemas.py + language: python + pass_filenames: false + files: > + (?x) + ^airflow-core/src/airflow/serialization/schema\.json$| + ^task-sdk/src/airflow/sdk/execution_time/schema/schema\.json$| + ^go-sdk/schema/.*\.json$| + ^scripts/ci/prek/sync_go_sdk_schemas\.py$ - id: check-go-sdk-generated-drift name: Check Go SDK generated files against the schemas they generate from description: "Regenerate the Dag specs and the coordinator-protocol models, and fail when a committed file differs" @@ -355,10 +367,11 @@ repos: # golang so prek provisions the toolchain the generators need, the way the # other checks that shell out to `go` in go-sdk get theirs. language: golang + # The vendored schemas, not the Python originals: sync-go-sdk-schemas above is + # what ties those to the copies, and it needs neither Go nor the network. files: > (?x) - ^airflow-core/src/airflow/serialization/schema\.json$| - ^task-sdk/src/airflow/sdk/execution_time/schema/schema\.json$| + ^go-sdk/schema/.*\.json$| ^go-sdk/airflow/spec\.gen\.go$| ^go-sdk/airflow/spec\.go$| ^go-sdk/internal/genspec/.*\.go$| diff --git a/go-sdk/README.md b/go-sdk/README.md index a6dedde08049b..eedf42a20050d 100644 --- a/go-sdk/README.md +++ b/go-sdk/README.md @@ -380,17 +380,34 @@ Python supervisor / task runner The Go side of the protocol is implemented in `pkg/execution/`. On the Python side it is the `ExecutableCoordinator` in `task-sdk/src/airflow/sdk/coordinators/executable/coordinator.py`. +## The vendored schemas + +`schema/dag-schema.json` and `schema/supervisor-schema.json` are byte-for-byte copies of schemas +airflow-core and task-sdk own. The generators read the copies, not the originals, so a generated file +is explainable from a file inside this module and a standalone checkout — a published Go module, an +ASF source release — can regenerate and test without the monorepo around it. `ts-sdk` and `java-sdk` +vendor theirs the same way. + +Two prek hooks split the work that keeps the copies honest: `sync-go-sdk-schemas` copies an original +over its copy when the two differ and fails, since copying is mechanical; and +`check-go-sdk-generated-drift` regenerates from the copies and fails when the committed Go differs, +since what to do about a new schema construct — a generator rule, an authoring exclusion — is a +decision. + ## Regenerating the coordinator-protocol models -The types in `pkg/execution/genmodels/` are generated from the in-tree supervisor schema snapshot -(`task-sdk/src/airflow/sdk/execution_time/schema/schema.json`); do not edit them by hand. To move the -SDK to a newer schema version: +The types in `pkg/execution/genmodels/` are generated from `schema/supervisor-schema.json`, this +module's vendored copy of the supervisor schema snapshot the Python Task SDK owns +(`task-sdk/src/airflow/sdk/execution_time/schema/schema.json`); do not edit either by hand. To move +the SDK to a newer schema version: -1. Set `SupervisorSchemaVersion` in [`pkg/execution/messages.go`](./pkg/execution/messages.go) to the - snapshot's `api_version` date. -2. Run `just generate-models`. +1. Refresh the copy: `prek run sync-go-sdk-schemas --all-files`, which overwrites + `schema/supervisor-schema.json` from task-sdk's and fails so the change lands in review. +2. Set `SupervisorSchemaVersion` in [`pkg/execution/messages.go`](./pkg/execution/messages.go) to the + copy's `api_version` date. +3. Run `just generate-models`, and commit the copy with what it generated. -`TestSupervisorSchemaVersionMatchesSnapshot` fails when the constant and the snapshot's `api_version` +`TestSupervisorSchemaVersionMatchesSnapshot` fails when the constant and the copy's `api_version` drift, so a missed bump is caught by `go test`. A snapshot can also grow a field without the `api_version` moving, which leaves the models behind with nothing failing — msgpack drops a field the struct does not declare — so the `check-go-sdk-generated-drift` prek hook regenerates them and fails @@ -399,9 +416,10 @@ when the committed files differ. ## Regenerating the Dag and task specs `airflow.DagSpec` and `airflow.TaskSpec` in [`airflow/spec.gen.go`](./airflow/spec.gen.go) are -generated from airflow-core's Dag serialization schema -(`airflow-core/src/airflow/serialization/schema.json`), which Python owns; do not edit them by hand. -Run `just generate-specs` after changing the schema or the generator. +generated from `schema/dag-schema.json`, this module's vendored copy of airflow-core's Dag +serialization schema (`airflow-core/src/airflow/serialization/schema.json`), which Python owns; do +not edit either by hand. Refresh the copy with `prek run sync-go-sdk-schemas --all-files`, then run +`just generate-specs` after changing the schema or the generator. The schema is the serialized shape rather than the authoring one, so [`internal/genspec/authoring.go`](./internal/genspec/authoring.go) holds the three tables that turn diff --git a/go-sdk/airflow/spec.go b/go-sdk/airflow/spec.go index 4b928999d4e4b..0d76198b8903a 100644 --- a/go-sdk/airflow/spec.go +++ b/go-sdk/airflow/spec.go @@ -35,7 +35,7 @@ import "reflect" // the zero value. Serializing a Dag converts the fields; a tag would invite // json.Marshal(spec) to skip that step and produce a shape core misreads. -//go:generate go run ../internal/genspec -schema ../../airflow-core/src/airflow/serialization/schema.json -out ../../.build/go-sdk/spec.schema.json +//go:generate go run ../internal/genspec -schema ../schema/dag-schema.json -out ../../.build/go-sdk/spec.schema.json //go:generate go run github.com/atombender/go-jsonschema@v0.23.1 --only-models --struct-name-from-title --tags "" --capitalization ID --capitalization JSON --capitalization MD --capitalization XCom -p airflow -o spec.gen.go ../../.build/go-sdk/spec.schema.json //go:generate go run ../internal/genspec -license spec.gen.go diff --git a/go-sdk/internal/genspec/main.go b/go-sdk/internal/genspec/main.go index 87f93b67a2601..208af46833d44 100644 --- a/go-sdk/internal/genspec/main.go +++ b/go-sdk/internal/genspec/main.go @@ -15,10 +15,9 @@ // specific language governing permissions and limitations // under the License. -// Command genspec rewrites Airflow core's Dag serialization schema -// (airflow-core/src/airflow/serialization/schema.json) into the schema the airflow -// package's DagSpec and TaskSpec generate from, so that the two structs are not -// hand-maintained. +// Command genspec rewrites Airflow core's Dag serialization schema (go-sdk's vendored +// copy of it, schema/dag-schema.json) into the schema the airflow package's DagSpec +// and TaskSpec generate from, so that the two structs are not hand-maintained. // // The schema is owned by Python and stays untouched; the rewritten copy is a build // artifact. genspec rewrites it in two passes: shapeForAuthoring turns the @@ -45,7 +44,7 @@ func main() { schemaPath := flag.String( "schema", "", - "path to airflow-core's Dag serialization schema.json", + "path to the vendored copy of airflow-core's Dag serialization schema", ) outPath := flag.String("out", "", "path to write the rewritten schema to") licensePath := flag.String( diff --git a/go-sdk/internal/genspec/normalize_test.go b/go-sdk/internal/genspec/normalize_test.go index b10bcb8fa186f..263b4918a82c7 100644 --- a/go-sdk/internal/genspec/normalize_test.go +++ b/go-sdk/internal/genspec/normalize_test.go @@ -26,8 +26,9 @@ import ( "github.com/stretchr/testify/require" ) -// coreSchemaPath is the schema genspec normalizes, reached from this package. -const coreSchemaPath = "../../../airflow-core/src/airflow/serialization/schema.json" +// coreSchemaPath is the schema genspec normalizes: go-sdk's vendored copy of +// airflow-core's, so this test runs in a standalone checkout as well. +const coreSchemaPath = "../../schema/dag-schema.json" func schemaFrom(t *testing.T, body string) map[string]any { t.Helper() diff --git a/go-sdk/pkg/execution/genmodels/gen.go b/go-sdk/pkg/execution/genmodels/gen.go index 0a121ea4efd5f..900207090c701 100644 --- a/go-sdk/pkg/execution/genmodels/gen.go +++ b/go-sdk/pkg/execution/genmodels/gen.go @@ -19,13 +19,14 @@ // messages (msgpack-over-socket) exchanged between the Airflow supervisor and a // Go SDK bundle in coordinator mode. // -// These files are generated from the supervisor wire-schema snapshot owned by the -// Python Task SDK (task-sdk/src/airflow/sdk/execution_time/schema/schema.json): +// These files are generated from ../../../schema/supervisor-schema.json, go-sdk's +// vendored copy of the supervisor wire-schema snapshot the Python Task SDK owns: // models.gen.go holds the struct types, discriminators.gen.go the Type // constants and EnsureType, defaults.gen.go the DecodeMsgpack methods that seed // non-zero schema defaults. Don't edit them by hand; change the Pydantic models, -// let the generate-supervisor-schemas-snapshot prek hook refresh the snapshot, -// then re-run `just generate-models` (go generate). +// let the generate-supervisor-schemas-snapshot prek hook refresh the snapshot and +// sync-go-sdk-schemas refresh the vendored copy, then re-run `just generate-models` +// (go generate). // // go generate runs go-jsonschema, then the local gen tool, which strips // go-jsonschema's dead anyOf-branch typedefs, widens concrete int/float/bool @@ -35,5 +36,5 @@ // discriminators.gen.go and defaults.gen.go. package genmodels -//go:generate go run github.com/atombender/go-jsonschema@v0.23.1 --only-models --struct-name-from-title --tags msgpack --capitalization ID --capitalization URI --capitalization TI -p genmodels -o models.gen.go ../../../../task-sdk/src/airflow/sdk/execution_time/schema/schema.json -//go:generate go run ./gen -schema ../../../../task-sdk/src/airflow/sdk/execution_time/schema/schema.json -models models.gen.go -out discriminators.gen.go -defaults defaults.gen.go +//go:generate go run github.com/atombender/go-jsonschema@v0.23.1 --only-models --struct-name-from-title --tags msgpack --capitalization ID --capitalization URI --capitalization TI -p genmodels -o models.gen.go ../../../schema/supervisor-schema.json +//go:generate go run ./gen -schema ../../../schema/supervisor-schema.json -models models.gen.go -out discriminators.gen.go -defaults defaults.gen.go diff --git a/go-sdk/pkg/execution/messages_test.go b/go-sdk/pkg/execution/messages_test.go index fa81df4ea66d2..32fe80d35083f 100644 --- a/go-sdk/pkg/execution/messages_test.go +++ b/go-sdk/pkg/execution/messages_test.go @@ -30,25 +30,18 @@ import ( "github.com/apache/airflow/go-sdk/pkg/execution/genmodels" ) -// sourceSchemaRelPath locates the canonical supervisor schema snapshot owned by -// the Python Task SDK, relative to this package directory within the monorepo. -// It is the same file genmodels is generated against (referenced by relative -// path rather than vendored into go-sdk). -const sourceSchemaRelPath = "../../../task-sdk/src/airflow/sdk/execution_time/schema/schema.json" +// vendoredSchemaRelPath locates go-sdk's vendored copy of the supervisor schema +// snapshot the Python Task SDK owns, relative to this package directory. It is +// the file genmodels is generated against, and it ships with go-sdk, so this +// test needs no monorepo around it. The sync-go-sdk-schemas prek hook is what +// keeps the copy equal to task-sdk's. +const vendoredSchemaRelPath = "../../schema/supervisor-schema.json" // TestSupervisorSchemaVersionMatchesSnapshot pins SupervisorSchemaVersion to the // api_version of the schema the models are generated from, so a schema bump that -// forgets to update the constant (or vice versa) fails loudly. It only runs when -// the task-sdk source is reachable from the checkout, so a standalone go-sdk -// build (without the rest of the monorepo) does not fail. +// forgets to update the constant (or vice versa) fails loudly. func TestSupervisorSchemaVersionMatchesSnapshot(t *testing.T) { - raw, err := os.ReadFile(sourceSchemaRelPath) - if os.IsNotExist(err) { - t.Skipf( - "task-sdk schema source not reachable at %s; skipping version check", - sourceSchemaRelPath, - ) - } + raw, err := os.ReadFile(vendoredSchemaRelPath) require.NoError(t, err) var snapshot struct { APIVersion string `json:"api_version"` diff --git a/go-sdk/schema/dag-schema.json b/go-sdk/schema/dag-schema.json new file mode 100644 index 0000000000000..086bccb34bb5b --- /dev/null +++ b/go-sdk/schema/dag-schema.json @@ -0,0 +1,474 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "$id": "https://airflow.apache.com/schemas/serialized-dags.json", + "definitions": { + "datetime": { + "description": "A date time, stored as fractional seconds since the epoch", + "type": "number" + }, + "timedelta": { + "type": "number", + "minimum": 0 + }, + "typed_timedelta": { + "type": "object", + "properties": { + "__type": { + "type": "string", + "const": "timedelta" + }, + "__var": { "$ref": "#/definitions/timedelta" } + }, + "required": [ + "__type", + "__var" + ], + "additionalProperties": false + }, + "typed_relativedelta": { + "type": "object", + "description": "A dateutil.relativedelta.relativedelta object", + "properties": { + "__type": { + "type": "string", + "const": "relativedelta" + }, + "__var": { + "type": "object", + "properties": { + "weekday": { + "type": "array", + "items": { "type": "integer" }, + "minItems": 1, + "maxItems": 2 + } + }, + "additionalProperties": { "type": "integer" } + } + } + }, + "timezone": { + "anyOf": [ + { "type": "string" }, + { "type": "integer" } + ] + }, + "asset_definition": { + "type": "object", + "properties": { + "uri": { "type": "string" }, + "name": { "type": "string" }, + "group": { "type": "string" }, + "extra": { + "anyOf": [ + {"type": "null"}, + { "$ref": "#/definitions/dict" } + ] + }, + "watchers": { + "type": "array", + "items": { "$ref": "#/definitions/trigger" } + } + }, + "required": [ "uri", "extra" ] + }, + "asset": { + "type": "object", + "properties": { + "uri": { "type": "string" }, + "extra": { + "anyOf": [ + {"type": "null"}, + { "$ref": "#/definitions/dict" } + ] + } + }, + "required": [ "uri", "extra" ] + }, + "typed_asset": { + "type": "object", + "properties": { + "__type": { + "type": "string", + "constant": "asset" + }, + "__var": { "$ref": "#/definitions/asset" } + }, + "required": [ + "__type", + "__var" + ], + "additionalProperties": false + }, + "typed_asset_cond": { + "type": "object", + "properties": { + "__type": { + "anyOf": [{ + "type": "string", + "constant": "asset_or" + }, + { + "type": "string", + "constant": "asset_and" + } + ] + }, + "__var": { + "type": "array", + "items": { + "anyOf": [ + {"$ref": "#/definitions/typed_asset"}, + { "$ref": "#/definitions/typed_asset_cond"} + ] + } + } + }, + "required": [ + "__type", + "__var" + ], + "additionalProperties": false + }, + "trigger": { + "type": "object", + "properties": { + "classpath": { "type": "string" }, + "kwargs": { "$ref": "#/definitions/dict" } + }, + "required": [ "classpath", "kwargs" ] + }, + "dict": { + "description": "A python dictionary containing values of any type", + "type": "object" + }, + "arg_binding": { + "$comment": "One captured TaskFlow call argument of a @task.stub task. Materialized directly by _serialize_node, so it stays plain JSON with no {__type, __var} encoding. The object stays open so future binding fields keep validating on older cores", + "type": "object", + "properties": { + "name": { "type": "string" }, + "kind": { "type": "string", "enum": [ "xcom", "literal" ] }, + "value_schema": { "type": "object" }, + "task_id": { "type": "string" }, + "value": {}, + "from_default": { "type": "boolean" } + }, + "required": [ "name", "kind" ] + }, + "color": { + "type": "string", + "pattern": "^#[a-fA-F0-9]{3,6}$" + }, + "extra_links": { + "type": "array", + "items": { + "type": "object", + "minProperties": 1, + "maxProperties": 1 + } + }, + "dag_dependencies": { + "type": "array", + "items": { + "type": "object" + } + }, + "dag": { + "type": "object", + "properties": { + "params": { "$ref": "#/definitions/params" }, + "dag_id": { "type": "string" }, + "tasks": { "$ref": "#/definitions/tasks" }, + "timezone": { "$ref": "#/definitions/timezone" }, + "owner_links": { "type": "object" }, + "timetable": { + "type": "object", + "properties": { + "type": { "type": "string" }, + "value": { "$ref": "#/definitions/dict" } + } + }, + "catchup": { "type": "boolean" }, + "allowed_run_types": { + "anyOf": [ + { "type": "array", "items": { "type": "string" } }, + { "type": "null" } + ] + }, + "fail_fast": { "type": "boolean", "default": false }, + "fileloc": { "type" : "string"}, + "relative_fileloc": { "type" : "string"}, + "bundle_name": { "anyOf": [{ "type": "null" }, { "type": "string" }] }, + "_processor_dags_folder": { + "anyOf": [ + { "type": "null" }, + { "type": "string" } + ] + }, + "dag_display_name": { "type" : "string"}, + "description": { "type" : "string"}, + "deadline": { + "anyOf": [ + { "$ref": "#/definitions/dict" }, + { + "type": "array", + "items": { "$ref": "#/definitions/dict" } + }, + { + "$comment": "Once persisted, a Dag's deadline alerts live as rows in the deadline_alert table and the serialized Dag keeps only a list of UUID strings referencing them (see SerializedDagModel._generate_deadline_uuids). This branch lets the stored form validate at any lifecycle stage, not only before the dict->UUID rewrite.", + "type": "array", + "items": { "type": "string", + "pattern": "^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$" + } + }, + { "type": "null" } + ] + }, + "_concurrency": { "type" : "number"}, + "max_active_tasks": { "type" : "number" }, + "max_active_runs": { "type" : "number" }, + "max_consecutive_failed_dag_runs": { "type" : "number" }, + "default_args": { "$ref": "#/definitions/dict" }, + "start_date": { "$ref": "#/definitions/datetime" }, + "end_date": { "$ref": "#/definitions/datetime" }, + "dagrun_timeout": { "$ref": "#/definitions/timedelta" }, + "doc_md": { "type" : "string"}, + "access_control": {"$ref": "#/definitions/dict" }, + "is_paused_upon_creation": { "type": "boolean" }, + "has_on_success_callback": { "type": "boolean", "default": false }, + "has_on_failure_callback": { "type": "boolean", "default": false }, + "render_template_as_native_obj": { "type": "boolean", "default": false }, + "tags": { "type": "array" }, + "task_group": {"anyOf": [ + { "type": "null" }, + { "$ref": "#/definitions/task_group" } + ]}, + "edge_info": { "$ref": "#/definitions/edge_info" }, + "dag_dependencies": { "$ref": "#/definitions/dag_dependencies" }, + "disable_bundle_versioning": {"type": "boolean" }, + "rerun_with_latest_version": {"type": ["boolean", "null"], "default": null} + }, + "required": [ + "dag_id", + "fileloc", + "tasks" + ], + "additionalProperties": false + }, + "tasks": { + "type": "array", + "additionalProperties": { "$ref": "#/definitions/operator" } + }, + "params": { + "type": "array", + "prefixItems": [ + { "type": "string" }, + { "$ref": "#/definitions/param" } + ], + "unevaluatedItems": false + }, + "param": { + "$comment": "A param for a dag / operator", + "type": "object", + "required": [ + "__class", + "default" + ], + "properties": { + "__class": { "type": "string" }, + "default": {}, + "description": {"anyOf": [{"type":"string"}, {"type":"null"}]}, + "schema": { "$ref": "#/definitions/dict" } + } + }, + "operator": { + "$comment": "A task/operator in a DAG", + "type": "object", + "required": [ + "task_type", + "_task_module", + "task_id", + "ui_color", + "ui_fgcolor", + "template_fields" + ], + "properties": { + "task_type": { "type": "string", "default": "BaseOperator"}, + "_task_module": { "type": "string" }, + "_operator_extra_links": { "$ref": "#/definitions/extra_links" }, + "task_id": { "type": "string" }, + "_task_display_name": { "type": "string" }, + "owner": { "type": "string", "default": "airflow" }, + "start_date": { "$ref": "#/definitions/datetime" }, + "end_date": { "$ref": "#/definitions/datetime" }, + "trigger_rule": { "type": "string", "default": "all_success" }, + "depends_on_past": { "type": "boolean", "default": false }, + "ignore_first_depends_on_past": { "type": "boolean", "default": false }, + "wait_for_past_depends_before_skipping": { "type": "boolean", "default": false }, + "wait_for_downstream": { "type": "boolean", "default": false }, + "retries": { "type": "number", "default": 0 }, + "queue": { "type": "string", "default": "default" }, + "pool": { "type": "string", "default": "default_pool" }, + "pool_slots": { "type": "number", "default": 1 }, + "execution_timeout": { "$ref": "#/definitions/timedelta" }, + "retry_delay": { "$ref": "#/definitions/timedelta", "default": 300.0 }, + "retry_exponential_backoff": { "type": "number", "default": 0 }, + "max_retry_delay": { "$ref": "#/definitions/timedelta" }, + "params": { "$ref": "#/definitions/params" }, + "priority_weight": { "type": "number", "default": 1 }, + "weight_rule": { "type": "string", "default": "downstream" }, + "executor": { "type": "string" }, + "executor_config": { "$ref": "#/definitions/dict" }, + "do_xcom_push": { "type": "boolean", "default": true }, + "email_on_failure": { "type": "boolean", "default": true }, + "email_on_retry": { "type": "boolean", "default": true }, + "ui_color": { "type": "string", "default": "#fff" }, + "ui_fgcolor": { "type": "string", "default": "#000" }, + "template_fields": { + "type": "array", + "items": { "type": "string" }, + "default": [] + }, + "template_ext": {"type": "array", "default": []}, + "template_fields_renderers": {"$ref": "#/definitions/dict", "default": {}}, + "downstream_task_ids": { + "type": "array", + "items": { "type": "string" }, + "default": [] + }, + "doc": { "type": "string" }, + "doc_md": { "type": "string" }, + "doc_json": { "type": "string" }, + "doc_yaml": { "type": "string" }, + "doc_rst": { "type": "string" }, + "_logger_name": { "type": "string" }, + "_needs_expansion": { "type": "boolean"}, + "_is_mapped": { "const": true, "$comment": "only present when True", "default": false }, + "_is_sensor": { "const": true, "$comment": "only present when True", "default": false }, + "partial_kwargs": { "type": "object" }, + "_disallow_kwargs_override": { "type": "boolean"}, + "_expand_input_attr": { "type": "string" }, + "map_index_template": { "type": "string" }, + "allow_nested_operators": { "type": "boolean", "default": true }, + "render_template_as_native_obj": { "anyOf": [{"type": "boolean"}, {"type": "null"}], "default": null }, + "inlets": {"type": "array", "default": []}, + "outlets": {"type": "array", "default": []}, + "has_on_execute_callback": {"type": "boolean", "default": false}, + "has_on_failure_callback": {"type": "boolean", "default": false}, + "has_on_skipped_callback": {"type": "boolean", "default": false}, + "has_on_success_callback": {"type": "boolean", "default": false}, + "has_on_retry_callback": {"type": "boolean", "default": false}, + "multiple_outputs": {"type": "boolean", "default": false}, + "start_from_trigger": {"type": "boolean", "default": false}, + "start_trigger_args": {"type": "object", "default": null}, + "is_setup": {"type": "boolean", "default": false}, + "is_teardown": {"type": "boolean", "default": false}, + "on_failure_fail_dagrun": {"type": "boolean", "default": false}, + "max_active_tis_per_dag": {"type": "integer"}, + "max_active_tis_per_dagrun": {"type": "integer"}, + "_arg_bindings": { + "$comment": "Used for mixed-language tasks (the @task.stub operator) or tasks defined in a non-Python SDK Dag", + "type": "array", + "items": { "$ref": "#/definitions/arg_binding" } + } + }, + "dependencies": { + "expand_input": ["partial_kwargs", "_is_mapped"], + "partial_kwargs": ["expand_input", "_is_mapped"], + "_is_mapped": ["expand_input", "partial_kwargs"] + }, + "additionalProperties": true + }, + "task_group": { + "$comment": "A TaskGroup containing tasks", + "type": "object", + "required": [ + "_group_id", + "group_display_name", + "prefix_group_id", + "children", + "tooltip", + "ui_color", + "ui_fgcolor", + "upstream_group_ids", + "downstream_group_ids", + "upstream_task_ids", + "downstream_task_ids" + ], + "properties": { + "_group_id": {"anyOf": [{"type": "null"}, { "type": "string" }]}, + "group_display_name": {"type": "string" }, + "is_mapped": { "type": "boolean" }, + "prefix_group_id": { "type": "boolean" }, + "children": { "$ref": "#/definitions/dict" }, + "tooltip": { "type": "string" }, + "doc_md": { + "anyOf": [ + { "type": "string" }, + { "type": "null" } + ]}, + "ui_color": { "type": "string" }, + "ui_fgcolor": { "type": "string" }, + "upstream_group_ids": { + "type": "array", + "items": { "type": "string" } + }, + "downstream_group_ids": { + "type": "array", + "items": { "type": "string" } + }, + "upstream_task_ids": { + "type": "array", + "items": { "type": "string" } + }, + "downstream_task_ids": { + "type": "array", + "items": { "type": "string" } + } + }, + "additionalProperties": false + }, + "edge_info": { + "$comment": "Metadata about DAG edges", + "type": "object", + "additionalProperties": { + "type": "object", + "additionalProperties": { + "type": "object", + "properties": { + "label": { "type": "string" } + }, + "required": ["label"], + "additionalProperties": false + } + } + } + }, + + "type": "object", + "allOf": [ + { + "type": "object", + "properties": { + "__version": { + "type": "integer", + "exclusiveMinimum": 0 + }, + "dag": { "$ref": "#/definitions/dag" }, + "client_defaults": { + "type": "object", + "description": "SDK-specific default values that differ from schema defaults", + "properties": { + "tasks": { + "type": "object", + "description": "Task-level default overrides" + } + }, + "additionalProperties": false + } + }, + "additionalProperties": false, + "required": [ "__version", "dag" ] + } + ] +} diff --git a/go-sdk/schema/supervisor-schema.json b/go-sdk/schema/supervisor-schema.json new file mode 100644 index 0000000000000..d6ee95927efee --- /dev/null +++ b/go-sdk/schema/supervisor-schema.json @@ -0,0 +1,5292 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "api_version": "2026-10-30", + "description": "Apache Airflow SDK Supervisor Schema", + "$defs": { + "AssetAliasReferenceAssetEventDagRun": { + "additionalProperties": false, + "description": "Schema for AssetAliasModel used in AssetEventDagRunReference.", + "properties": { + "name": { + "title": "Name", + "type": "string" + } + }, + "required": [ + "name" + ], + "title": "AssetAliasReferenceAssetEventDagRun", + "type": "object" + }, + "AssetEventResponse": { + "description": "Asset event schema with fields that are needed for Runtime.", + "properties": { + "id": { + "title": "Id", + "type": "integer" + }, + "timestamp": { + "format": "date-time", + "title": "Timestamp", + "type": "string" + }, + "extra": { + "anyOf": [ + { + "additionalProperties": { + "anyOf": [ + { + "$ref": "#/$defs/JsonValue" + }, + { + "type": "null" + } + ] + }, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Extra" + }, + "asset": { + "$ref": "#/$defs/AssetResponse" + }, + "created_dagruns": { + "items": { + "$ref": "#/$defs/DagRunAssetReference" + }, + "title": "Created Dagruns", + "type": "array" + }, + "source_task_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Source Task Id" + }, + "source_dag_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Source Dag Id" + }, + "source_run_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Source Run Id" + }, + "source_map_index": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Source Map Index" + }, + "partition_key": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Partition Key" + } + }, + "required": [ + "id", + "timestamp", + "asset", + "created_dagruns" + ], + "title": "AssetEventResponse", + "type": "object" + }, + "AssetEventsResult": { + "description": "Response to GetAssetEvent request.", + "properties": { + "asset_events": { + "items": { + "$ref": "#/$defs/AssetEventResponse" + }, + "title": "Asset Events", + "type": "array" + }, + "type": { + "const": "AssetEventsResult", + "default": "AssetEventsResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "asset_events" + ], + "title": "AssetEventsResult", + "type": "object" + }, + "AssetProfile": { + "additionalProperties": false, + "description": "Profile of an asset-like object.\n\nAsset will have name, uri defined, with type set to 'Asset'.\nAssetNameRef will have name defined, type set to 'AssetNameRef'.\nAssetUriRef will have uri defined, type set to 'AssetUriRef'.\nAssetAlias will have name defined, type set to 'AssetAlias'.\n\nNote that 'type' here is distinct from 'asset_type' the user declares on an\nAsset (or subclass). This field is for distinguishing between different\nasset-related types (Asset, AssetRef, or AssetAlias).", + "properties": { + "name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Name" + }, + "uri": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Uri" + }, + "type": { + "title": "Type", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "AssetProfile", + "type": "object" + }, + "AssetResponse": { + "description": "Asset schema for responses with fields that are needed for Runtime.", + "properties": { + "name": { + "title": "Name", + "type": "string" + }, + "uri": { + "title": "Uri", + "type": "string" + }, + "group": { + "title": "Group", + "type": "string" + }, + "extra": { + "anyOf": [ + { + "additionalProperties": { + "anyOf": [ + { + "$ref": "#/$defs/JsonValue" + }, + { + "type": "null" + } + ] + }, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Extra" + } + }, + "required": [ + "name", + "uri", + "group" + ], + "title": "AssetResponse", + "type": "object" + }, + "AssetResult": { + "description": "Response to ReadXCom request.", + "properties": { + "name": { + "title": "Name", + "type": "string" + }, + "uri": { + "title": "Uri", + "type": "string" + }, + "group": { + "title": "Group", + "type": "string" + }, + "extra": { + "anyOf": [ + { + "additionalProperties": { + "anyOf": [ + { + "$ref": "#/$defs/JsonValue" + }, + { + "type": "null" + } + ] + }, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Extra" + }, + "type": { + "const": "AssetResult", + "default": "AssetResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "name", + "uri", + "group" + ], + "title": "AssetResult", + "type": "object" + }, + "AssetStateStoreResult": { + "additionalProperties": false, + "description": "Response to GetAssetStateStore; wraps the generated API response for supervisor to worker comms.", + "properties": { + "value": { + "anyOf": [ + { + "$ref": "#/$defs/JsonValue" + }, + { + "type": "null" + } + ] + }, + "type": { + "const": "AssetStateStoreResult", + "default": "AssetStateStoreResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "value" + ], + "title": "AssetStateStoreResult", + "type": "object" + }, + "AssetsByAliasResult": { + "description": "Response to GetAssetsByAlias; list of concrete assets resolved from an alias.", + "properties": { + "assets": { + "items": { + "$ref": "#/$defs/AssetResult" + }, + "title": "Assets", + "type": "array" + }, + "type": { + "const": "AssetsByAliasResult", + "default": "AssetsByAliasResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "assets" + ], + "title": "AssetsByAliasResult", + "type": "object" + }, + "AwaitInputTask": { + "additionalProperties": false, + "description": "Park a task instance awaiting human input (Human-in-the-loop), without a trigger.", + "properties": { + "state": { + "const": "awaiting_input", + "default": "awaiting_input", + "title": "State", + "type": "string" + }, + "timeout": { + "anyOf": [ + { + "format": "duration", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Timeout" + }, + "next_method": { + "title": "Next Method", + "type": "string" + }, + "next_kwargs": { + "anyOf": [ + { + "additionalProperties": { + "anyOf": [ + { + "$ref": "#/$defs/JsonValue" + }, + { + "type": "null" + } + ] + }, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Next Kwargs" + }, + "rendered_map_index": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Rendered Map Index" + }, + "type": { + "const": "AwaitInputTask", + "default": "AwaitInputTask", + "title": "Type", + "type": "string" + } + }, + "required": [ + "next_method" + ], + "title": "AwaitInputTask", + "type": "object" + }, + "BundleInfo": { + "description": "Schema for telling task which bundle to run with.", + "properties": { + "name": { + "title": "Name", + "type": "string" + }, + "version": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Version" + }, + "version_data": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Version Data" + } + }, + "required": [ + "name" + ], + "title": "BundleInfo", + "type": "object" + }, + "ClearAssetStateStoreByName": { + "properties": { + "name": { + "title": "Name", + "type": "string" + }, + "type": { + "const": "ClearAssetStateStoreByName", + "default": "ClearAssetStateStoreByName", + "title": "Type", + "type": "string" + } + }, + "required": [ + "name" + ], + "title": "ClearAssetStateStoreByName", + "type": "object" + }, + "ClearAssetStateStoreByUri": { + "properties": { + "uri": { + "title": "Uri", + "type": "string" + }, + "type": { + "const": "ClearAssetStateStoreByUri", + "default": "ClearAssetStateStoreByUri", + "title": "Type", + "type": "string" + } + }, + "required": [ + "uri" + ], + "title": "ClearAssetStateStoreByUri", + "type": "object" + }, + "ClearTaskStateStore": { + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "type": { + "const": "ClearTaskStateStore", + "default": "ClearTaskStateStore", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id" + ], + "title": "ClearTaskStateStore", + "type": "object" + }, + "ConnectionResponse": { + "description": "Connection schema for responses with fields that are needed for Runtime.", + "properties": { + "conn_id": { + "title": "Conn Id", + "type": "string" + }, + "conn_type": { + "title": "Conn Type", + "type": "string" + }, + "host": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Host" + }, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Schema" + }, + "login": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Login" + }, + "password": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Password" + }, + "port": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Port" + }, + "extra": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Extra" + } + }, + "required": [ + "conn_id", + "conn_type", + "host", + "schema", + "login", + "password", + "port", + "extra" + ], + "title": "ConnectionResponse", + "type": "object" + }, + "ConnectionResult": { + "properties": { + "conn_id": { + "title": "Conn Id", + "type": "string" + }, + "conn_type": { + "title": "Conn Type", + "type": "string" + }, + "host": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Host" + }, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Schema" + }, + "login": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Login" + }, + "password": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Password" + }, + "port": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Port" + }, + "extra": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Extra" + }, + "type": { + "const": "ConnectionResult", + "default": "ConnectionResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "conn_id", + "conn_type", + "host", + "schema", + "login", + "password", + "port", + "extra" + ], + "title": "ConnectionResult", + "type": "object" + }, + "CreateHITLDetailPayload": { + "description": "Add the input request part of a Human-in-the-loop response.", + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "options": { + "items": { + "type": "string" + }, + "minItems": 1, + "title": "Options", + "type": "array" + }, + "subject": { + "title": "Subject", + "type": "string" + }, + "body": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Body" + }, + "defaults": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Defaults" + }, + "multiple": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "default": false, + "title": "Multiple" + }, + "params": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Params" + }, + "assigned_users": { + "anyOf": [ + { + "items": { + "$ref": "#/$defs/HITLUser" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Assigned Users" + }, + "type": { + "const": "CreateHITLDetailPayload", + "default": "CreateHITLDetailPayload", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id", + "options", + "subject" + ], + "title": "CreateHITLDetailPayload", + "type": "object" + }, + "DRCount": { + "description": "Response containing count of Dag Runs matching certain filters.", + "properties": { + "count": { + "title": "Count", + "type": "integer" + }, + "type": { + "const": "DRCount", + "default": "DRCount", + "title": "Type", + "type": "string" + } + }, + "required": [ + "count" + ], + "title": "DRCount", + "type": "object" + }, + "DagCallbackRequest": { + "description": "A Class with information about the success/failure DAG callback to be executed.", + "properties": { + "filepath": { + "title": "Filepath", + "type": "string" + }, + "bundle_name": { + "title": "Bundle Name", + "type": "string" + }, + "bundle_version": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Bundle Version" + }, + "version_data": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Version Data" + }, + "msg": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Msg" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "context_from_server": { + "anyOf": [ + { + "$ref": "#/$defs/DagRunContext" + }, + { + "type": "null" + } + ], + "default": null + }, + "is_failure_callback": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "default": true, + "title": "Is Failure Callback" + }, + "type": { + "const": "DagCallbackRequest", + "default": "DagCallbackRequest", + "title": "Type", + "type": "string" + } + }, + "required": [ + "filepath", + "bundle_name", + "bundle_version", + "dag_id", + "run_id" + ], + "title": "DagCallbackRequest", + "type": "object" + }, + "DagFileParseRequest": { + "description": "Request for DAG File Parsing.\n\nThis is the request that the manager will send to the DAG parser with the dag file and\nany other necessary metadata.", + "properties": { + "file": { + "title": "File", + "type": "string" + }, + "bundle_path": { + "format": "path", + "title": "Bundle Path", + "type": "string" + }, + "bundle_name": { + "title": "Bundle Name", + "type": "string" + }, + "callback_requests": { + "items": { + "discriminator": { + "mapping": { + "DagCallbackRequest": "#/$defs/DagCallbackRequest", + "EmailRequest": "#/$defs/EmailRequest", + "TaskCallbackRequest": "#/$defs/TaskCallbackRequest" + }, + "propertyName": "type" + }, + "oneOf": [ + { + "$ref": "#/$defs/DagCallbackRequest" + }, + { + "$ref": "#/$defs/TaskCallbackRequest" + }, + { + "$ref": "#/$defs/EmailRequest" + } + ] + }, + "title": "Callback Requests", + "type": "array" + }, + "type": { + "const": "DagFileParseRequest", + "default": "DagFileParseRequest", + "title": "Type", + "type": "string" + } + }, + "required": [ + "file", + "bundle_path", + "bundle_name" + ], + "title": "DagFileParseRequest", + "type": "object" + }, + "DagFileParsingResult": { + "description": "Result of DAG File Parsing.\n\nThis is the result of a successful DAG parse, in this class, we gather all serialized DAGs,\nimport errors and warnings to send back to the scheduler to store in the DB.", + "properties": { + "fileloc": { + "title": "Fileloc", + "type": "string" + }, + "serialized_dags": { + "items": { + "$ref": "#/$defs/LazyDeserializedDAG" + }, + "title": "Serialized Dags", + "type": "array" + }, + "warnings": { + "anyOf": [ + { + "items": {}, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Warnings" + }, + "import_errors": { + "anyOf": [ + { + "additionalProperties": { + "type": "string" + }, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Import Errors" + }, + "parsed_definitions": { + "items": { + "type": "string" + }, + "title": "Parsed Definitions", + "type": "array" + }, + "dag_source_codes": { + "additionalProperties": { + "$ref": "#/$defs/DagSourceCode" + }, + "title": "Dag Source Codes", + "type": "object" + }, + "type": { + "const": "DagFileParsingResult", + "default": "DagFileParsingResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "fileloc", + "serialized_dags" + ], + "title": "DagFileParsingResult", + "type": "object" + }, + "DagResult": { + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "is_paused": { + "title": "Is Paused", + "type": "boolean" + }, + "bundle_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Bundle Name" + }, + "bundle_version": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Bundle Version" + }, + "relative_fileloc": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Relative Fileloc" + }, + "owners": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Owners" + }, + "tags": { + "items": { + "type": "string" + }, + "title": "Tags", + "type": "array" + }, + "next_dagrun": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Next Dagrun" + }, + "type": { + "const": "DagResult", + "default": "DagResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id", + "is_paused", + "bundle_name", + "bundle_version", + "relative_fileloc", + "owners", + "tags", + "next_dagrun" + ], + "title": "DagResult", + "type": "object" + }, + "DagRunAssetReference": { + "additionalProperties": false, + "description": "DagRun serializer for asset responses.", + "properties": { + "run_id": { + "title": "Run Id", + "type": "string" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "logical_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Logical Date" + }, + "start_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Start Date" + }, + "end_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "End Date" + }, + "state": { + "title": "State", + "type": "string" + }, + "data_interval_start": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Data Interval Start" + }, + "data_interval_end": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Data Interval End" + }, + "partition_key": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Partition Key" + } + }, + "required": [ + "run_id", + "dag_id", + "logical_date", + "start_date", + "end_date", + "state", + "data_interval_start", + "data_interval_end", + "partition_key" + ], + "title": "DagRunAssetReference", + "type": "object" + }, + "DagRunContext": { + "description": "Class to pass context info from the server to build a Execution context object.", + "properties": { + "dag_run": { + "anyOf": [ + { + "$ref": "#/$defs/DagRun" + }, + { + "type": "null" + } + ], + "default": null + }, + "last_ti": { + "anyOf": [ + { + "$ref": "#/$defs/TaskInstance" + }, + { + "type": "null" + } + ], + "default": null + } + }, + "title": "DagRunContext", + "type": "object" + }, + "DagRunResult": { + "additionalProperties": false, + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "logical_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Logical Date" + }, + "data_interval_start": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Data Interval Start" + }, + "data_interval_end": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Data Interval End" + }, + "run_after": { + "format": "date-time", + "title": "Run After", + "type": "string" + }, + "start_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Start Date" + }, + "end_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "End Date" + }, + "clear_number": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": 0, + "title": "Clear Number" + }, + "run_type": { + "$ref": "#/$defs/DagRunType" + }, + "state": { + "$ref": "#/$defs/DagRunState" + }, + "conf": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Conf" + }, + "triggering_user_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Triggering User Name" + }, + "consumed_asset_events": { + "items": { + "$ref": "#/$defs/AssetEventDagRunReference" + }, + "title": "Consumed Asset Events", + "type": "array" + }, + "partition_key": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Partition Key" + }, + "partition_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Partition Date" + }, + "note": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Note" + }, + "team_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Team Name" + }, + "type": { + "const": "DagRunResult", + "default": "DagRunResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id", + "run_id", + "logical_date", + "data_interval_start", + "data_interval_end", + "run_after", + "start_date", + "end_date", + "run_type", + "state", + "consumed_asset_events", + "partition_key" + ], + "title": "DagRunResult", + "type": "object" + }, + "DagRunState": { + "description": "All possible states that a DagRun can be in.\n\nThese are \"shared\" with TaskInstanceState in some parts of the code,\nso please ensure that their values always match the ones with the\nsame name in TaskInstanceState.", + "enum": [ + "queued", + "running", + "success", + "failed" + ], + "title": "DagRunState", + "type": "string" + }, + "DagRunStateResult": { + "properties": { + "state": { + "$ref": "#/$defs/DagRunState" + }, + "type": { + "const": "DagRunStateResult", + "default": "DagRunStateResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "state" + ], + "title": "DagRunStateResult", + "type": "object" + }, + "DagRunType": { + "description": "Class with DagRun types.", + "enum": [ + "backfill", + "scheduled", + "manual", + "operator_triggered", + "asset_triggered", + "asset_materialization" + ], + "title": "DagRunType", + "type": "string" + }, + "DagSourceCode": { + "description": "Raw source code and its language identifier for a DAG definition.", + "properties": { + "source_code": { + "title": "Source Code", + "type": "string" + }, + "language": { + "title": "Language", + "type": "string" + } + }, + "required": [ + "source_code", + "language" + ], + "title": "DagSourceCode", + "type": "object" + }, + "DeferTask": { + "additionalProperties": false, + "description": "Update a task instance state to deferred.", + "properties": { + "state": { + "const": "deferred", + "default": "deferred", + "title": "State", + "type": "string" + }, + "classpath": { + "title": "Classpath", + "type": "string" + }, + "trigger_kwargs": { + "anyOf": [ + { + "additionalProperties": { + "anyOf": [ + { + "$ref": "#/$defs/JsonValue" + }, + { + "type": "null" + } + ] + }, + "type": "object" + }, + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Trigger Kwargs" + }, + "trigger_timeout": { + "anyOf": [ + { + "format": "duration", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Trigger Timeout" + }, + "queue": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Queue" + }, + "next_method": { + "title": "Next Method", + "type": "string" + }, + "next_kwargs": { + "anyOf": [ + { + "additionalProperties": { + "anyOf": [ + { + "$ref": "#/$defs/JsonValue" + }, + { + "type": "null" + } + ] + }, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Next Kwargs" + }, + "rendered_map_index": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Rendered Map Index" + }, + "type": { + "const": "DeferTask", + "default": "DeferTask", + "title": "Type", + "type": "string" + } + }, + "required": [ + "classpath", + "next_method" + ], + "title": "DeferTask", + "type": "object" + }, + "DeleteAssetStateStoreByName": { + "properties": { + "name": { + "title": "Name", + "type": "string" + }, + "key": { + "title": "Key", + "type": "string" + }, + "type": { + "const": "DeleteAssetStateStoreByName", + "default": "DeleteAssetStateStoreByName", + "title": "Type", + "type": "string" + } + }, + "required": [ + "name", + "key" + ], + "title": "DeleteAssetStateStoreByName", + "type": "object" + }, + "DeleteAssetStateStoreByUri": { + "properties": { + "uri": { + "title": "Uri", + "type": "string" + }, + "key": { + "title": "Key", + "type": "string" + }, + "type": { + "const": "DeleteAssetStateStoreByUri", + "default": "DeleteAssetStateStoreByUri", + "title": "Type", + "type": "string" + } + }, + "required": [ + "uri", + "key" + ], + "title": "DeleteAssetStateStoreByUri", + "type": "object" + }, + "DeleteTaskStateStore": { + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "key": { + "title": "Key", + "type": "string" + }, + "type": { + "const": "DeleteTaskStateStore", + "default": "DeleteTaskStateStore", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id", + "key" + ], + "title": "DeleteTaskStateStore", + "type": "object" + }, + "DeleteVariable": { + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "type": { + "const": "DeleteVariable", + "default": "DeleteVariable", + "title": "Type", + "type": "string" + } + }, + "required": [ + "key" + ], + "title": "DeleteVariable", + "type": "object" + }, + "DeleteXCom": { + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "task_id": { + "title": "Task Id", + "type": "string" + }, + "map_index": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Map Index" + }, + "type": { + "const": "DeleteXCom", + "default": "DeleteXCom", + "title": "Type", + "type": "string" + } + }, + "required": [ + "key", + "dag_id", + "run_id", + "task_id" + ], + "title": "DeleteXCom", + "type": "object" + }, + "EmailRequest": { + "description": "Email notification request for task failures/retries.", + "properties": { + "filepath": { + "title": "Filepath", + "type": "string" + }, + "bundle_name": { + "title": "Bundle Name", + "type": "string" + }, + "bundle_version": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Bundle Version" + }, + "version_data": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Version Data" + }, + "msg": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Msg" + }, + "ti": { + "$ref": "#/$defs/TaskInstance" + }, + "email_type": { + "default": "failure", + "enum": [ + "failure", + "retry" + ], + "title": "Email Type", + "type": "string" + }, + "context_from_server": { + "$ref": "#/$defs/TIRunContext" + }, + "type": { + "const": "EmailRequest", + "default": "EmailRequest", + "title": "Type", + "type": "string" + } + }, + "required": [ + "filepath", + "bundle_name", + "bundle_version", + "ti", + "context_from_server" + ], + "title": "EmailRequest", + "type": "object" + }, + "ErrorResponse": { + "properties": { + "error": { + "$ref": "#/$defs/ErrorType", + "default": "GENERIC_ERROR" + }, + "detail": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Detail" + }, + "type": { + "const": "ErrorResponse", + "default": "ErrorResponse", + "title": "Type", + "type": "string" + } + }, + "title": "ErrorResponse", + "type": "object" + }, + "ErrorType": { + "description": "Error types used in the API client.", + "enum": [ + "CONNECTION_NOT_FOUND", + "VARIABLE_NOT_FOUND", + "XCOM_NOT_FOUND", + "ASSET_NOT_FOUND", + "TASK_STORE_NOT_FOUND", + "ASSET_STORE_NOT_FOUND", + "DAGRUN_ALREADY_EXISTS", + "PERMISSION_DENIED", + "GENERIC_ERROR", + "API_SERVER_ERROR" + ], + "title": "ErrorType", + "type": "string" + }, + "GetAssetByName": { + "properties": { + "name": { + "title": "Name", + "type": "string" + }, + "type": { + "const": "GetAssetByName", + "default": "GetAssetByName", + "title": "Type", + "type": "string" + } + }, + "required": [ + "name" + ], + "title": "GetAssetByName", + "type": "object" + }, + "GetAssetByUri": { + "properties": { + "uri": { + "title": "Uri", + "type": "string" + }, + "type": { + "const": "GetAssetByUri", + "default": "GetAssetByUri", + "title": "Type", + "type": "string" + } + }, + "required": [ + "uri" + ], + "title": "GetAssetByUri", + "type": "object" + }, + "GetAssetEventByAsset": { + "properties": { + "name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Name" + }, + "uri": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Uri" + }, + "after": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "After" + }, + "before": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Before" + }, + "limit": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Limit" + }, + "ascending": { + "default": true, + "title": "Ascending", + "type": "boolean" + }, + "partition_key": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Partition Key" + }, + "partition_key_regexp_pattern": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Partition Key Regexp Pattern" + }, + "extra": { + "anyOf": [ + { + "additionalProperties": { + "type": "string" + }, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Extra" + }, + "type": { + "const": "GetAssetEventByAsset", + "default": "GetAssetEventByAsset", + "title": "Type", + "type": "string" + } + }, + "required": [ + "name", + "uri" + ], + "title": "GetAssetEventByAsset", + "type": "object" + }, + "GetAssetEventByAssetAlias": { + "properties": { + "alias_name": { + "title": "Alias Name", + "type": "string" + }, + "after": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "After" + }, + "before": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Before" + }, + "limit": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Limit" + }, + "ascending": { + "default": true, + "title": "Ascending", + "type": "boolean" + }, + "partition_key": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Partition Key" + }, + "partition_key_regexp_pattern": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Partition Key Regexp Pattern" + }, + "extra": { + "anyOf": [ + { + "additionalProperties": { + "type": "string" + }, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Extra" + }, + "type": { + "const": "GetAssetEventByAssetAlias", + "default": "GetAssetEventByAssetAlias", + "title": "Type", + "type": "string" + } + }, + "required": [ + "alias_name" + ], + "title": "GetAssetEventByAssetAlias", + "type": "object" + }, + "GetAssetStateStoreByName": { + "properties": { + "name": { + "title": "Name", + "type": "string" + }, + "key": { + "title": "Key", + "type": "string" + }, + "type": { + "const": "GetAssetStateStoreByName", + "default": "GetAssetStateStoreByName", + "title": "Type", + "type": "string" + } + }, + "required": [ + "name", + "key" + ], + "title": "GetAssetStateStoreByName", + "type": "object" + }, + "GetAssetStateStoreByUri": { + "properties": { + "uri": { + "title": "Uri", + "type": "string" + }, + "key": { + "title": "Key", + "type": "string" + }, + "type": { + "const": "GetAssetStateStoreByUri", + "default": "GetAssetStateStoreByUri", + "title": "Type", + "type": "string" + } + }, + "required": [ + "uri", + "key" + ], + "title": "GetAssetStateStoreByUri", + "type": "object" + }, + "GetAssetsByAlias": { + "properties": { + "alias_name": { + "title": "Alias Name", + "type": "string" + }, + "type": { + "const": "GetAssetsByAlias", + "default": "GetAssetsByAlias", + "title": "Type", + "type": "string" + } + }, + "required": [ + "alias_name" + ], + "title": "GetAssetsByAlias", + "type": "object" + }, + "GetConnection": { + "properties": { + "conn_id": { + "title": "Conn Id", + "type": "string" + }, + "type": { + "const": "GetConnection", + "default": "GetConnection", + "title": "Type", + "type": "string" + } + }, + "required": [ + "conn_id" + ], + "title": "GetConnection", + "type": "object" + }, + "GetDRCount": { + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "logical_dates": { + "anyOf": [ + { + "items": { + "format": "date-time", + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Logical Dates" + }, + "run_ids": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Run Ids" + }, + "states": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "States" + }, + "type": { + "const": "GetDRCount", + "default": "GetDRCount", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id" + ], + "title": "GetDRCount", + "type": "object" + }, + "GetDag": { + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "type": { + "const": "GetDag", + "default": "GetDag", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id" + ], + "title": "GetDag", + "type": "object" + }, + "GetDagRun": { + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "type": { + "const": "GetDagRun", + "default": "GetDagRun", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id", + "run_id" + ], + "title": "GetDagRun", + "type": "object" + }, + "GetDagRunState": { + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "type": { + "const": "GetDagRunState", + "default": "GetDagRunState", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id", + "run_id" + ], + "title": "GetDagRunState", + "type": "object" + }, + "GetHITLDetailResponse": { + "description": "Get the response content part of a Human-in-the-loop response.", + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "type": { + "const": "GetHITLDetailResponse", + "default": "GetHITLDetailResponse", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id" + ], + "title": "GetHITLDetailResponse", + "type": "object" + }, + "GetPrevSuccessfulDagRun": { + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "type": { + "const": "GetPrevSuccessfulDagRun", + "default": "GetPrevSuccessfulDagRun", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id" + ], + "title": "GetPrevSuccessfulDagRun", + "type": "object" + }, + "GetPreviousDagRun": { + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "logical_date": { + "format": "date-time", + "title": "Logical Date", + "type": "string" + }, + "state": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "State" + }, + "type": { + "const": "GetPreviousDagRun", + "default": "GetPreviousDagRun", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id", + "logical_date" + ], + "title": "GetPreviousDagRun", + "type": "object" + }, + "GetPreviousTI": { + "description": "Request to get previous task instance.", + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "task_id": { + "title": "Task Id", + "type": "string" + }, + "logical_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Logical Date" + }, + "map_index": { + "default": -1, + "title": "Map Index", + "type": "integer" + }, + "state": { + "anyOf": [ + { + "$ref": "#/$defs/TaskInstanceState" + }, + { + "type": "null" + } + ], + "default": null + }, + "type": { + "const": "GetPreviousTI", + "default": "GetPreviousTI", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id", + "task_id" + ], + "title": "GetPreviousTI", + "type": "object" + }, + "GetTICount": { + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "map_index": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Map Index" + }, + "task_ids": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Task Ids" + }, + "task_group_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Task Group Id" + }, + "logical_dates": { + "anyOf": [ + { + "items": { + "format": "date-time", + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Logical Dates" + }, + "run_ids": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Run Ids" + }, + "states": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "States" + }, + "type": { + "const": "GetTICount", + "default": "GetTICount", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id" + ], + "title": "GetTICount", + "type": "object" + }, + "GetTaskBreadcrumbs": { + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "type": { + "const": "GetTaskBreadcrumbs", + "default": "GetTaskBreadcrumbs", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id", + "run_id" + ], + "title": "GetTaskBreadcrumbs", + "type": "object" + }, + "GetTaskRescheduleStartDate": { + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "try_number": { + "default": 1, + "title": "Try Number", + "type": "integer" + }, + "type": { + "const": "GetTaskRescheduleStartDate", + "default": "GetTaskRescheduleStartDate", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id" + ], + "title": "GetTaskRescheduleStartDate", + "type": "object" + }, + "GetTaskStateStore": { + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "key": { + "title": "Key", + "type": "string" + }, + "type": { + "const": "GetTaskStateStore", + "default": "GetTaskStateStore", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id", + "key" + ], + "title": "GetTaskStateStore", + "type": "object" + }, + "GetTaskStates": { + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "map_index": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Map Index" + }, + "task_ids": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Task Ids" + }, + "task_group_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Task Group Id" + }, + "logical_dates": { + "anyOf": [ + { + "items": { + "format": "date-time", + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Logical Dates" + }, + "run_ids": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Run Ids" + }, + "type": { + "const": "GetTaskStates", + "default": "GetTaskStates", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id" + ], + "title": "GetTaskStates", + "type": "object" + }, + "GetVariable": { + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "type": { + "const": "GetVariable", + "default": "GetVariable", + "title": "Type", + "type": "string" + } + }, + "required": [ + "key" + ], + "title": "GetVariable", + "type": "object" + }, + "GetVariableKeys": { + "properties": { + "prefix": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Prefix" + }, + "limit": { + "default": 1000, + "title": "Limit", + "type": "integer" + }, + "offset": { + "default": 0, + "title": "Offset", + "type": "integer" + }, + "type": { + "const": "GetVariableKeys", + "default": "GetVariableKeys", + "title": "Type", + "type": "string" + } + }, + "title": "GetVariableKeys", + "type": "object" + }, + "GetXCom": { + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "task_id": { + "title": "Task Id", + "type": "string" + }, + "map_index": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Map Index" + }, + "include_prior_dates": { + "default": false, + "title": "Include Prior Dates", + "type": "boolean" + }, + "type": { + "const": "GetXCom", + "default": "GetXCom", + "title": "Type", + "type": "string" + } + }, + "required": [ + "key", + "dag_id", + "run_id", + "task_id" + ], + "title": "GetXCom", + "type": "object" + }, + "GetXComCount": { + "description": "Get the number of (mapped) XCom values available.", + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "task_id": { + "title": "Task Id", + "type": "string" + }, + "type": { + "const": "GetXComCount", + "default": "GetXComCount", + "title": "Type", + "type": "string" + } + }, + "required": [ + "key", + "dag_id", + "run_id", + "task_id" + ], + "title": "GetXComCount", + "type": "object" + }, + "GetXComSequenceItem": { + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "task_id": { + "title": "Task Id", + "type": "string" + }, + "offset": { + "title": "Offset", + "type": "integer" + }, + "type": { + "const": "GetXComSequenceItem", + "default": "GetXComSequenceItem", + "title": "Type", + "type": "string" + } + }, + "required": [ + "key", + "dag_id", + "run_id", + "task_id", + "offset" + ], + "title": "GetXComSequenceItem", + "type": "object" + }, + "GetXComSequenceSlice": { + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "task_id": { + "title": "Task Id", + "type": "string" + }, + "start": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Start" + }, + "stop": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Stop" + }, + "step": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Step" + }, + "include_prior_dates": { + "default": false, + "title": "Include Prior Dates", + "type": "boolean" + }, + "type": { + "const": "GetXComSequenceSlice", + "default": "GetXComSequenceSlice", + "title": "Type", + "type": "string" + } + }, + "required": [ + "key", + "dag_id", + "run_id", + "task_id", + "start", + "stop", + "step" + ], + "title": "GetXComSequenceSlice", + "type": "object" + }, + "HITLDetailRequestResult": { + "description": "Response to CreateHITLDetailPayload request.", + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "options": { + "items": { + "type": "string" + }, + "minItems": 1, + "title": "Options", + "type": "array" + }, + "subject": { + "title": "Subject", + "type": "string" + }, + "body": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Body" + }, + "defaults": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Defaults" + }, + "multiple": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "default": false, + "title": "Multiple" + }, + "params": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Params" + }, + "assigned_users": { + "anyOf": [ + { + "items": { + "$ref": "#/$defs/HITLUser" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Assigned Users" + }, + "type": { + "const": "HITLDetailRequestResult", + "default": "HITLDetailRequestResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id", + "options", + "subject" + ], + "title": "HITLDetailRequestResult", + "type": "object" + }, + "HITLUser": { + "description": "Schema for a Human-in-the-loop users.", + "properties": { + "id": { + "title": "Id", + "type": "string" + }, + "name": { + "title": "Name", + "type": "string" + } + }, + "required": [ + "id", + "name" + ], + "title": "HITLUser", + "type": "object" + }, + "InactiveAssetsResult": { + "description": "Response of InactiveAssets requests.", + "properties": { + "inactive_assets": { + "anyOf": [ + { + "items": { + "$ref": "#/$defs/AssetProfile" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Inactive Assets" + }, + "type": { + "const": "InactiveAssetsResult", + "default": "InactiveAssetsResult", + "title": "Type", + "type": "string" + } + }, + "title": "InactiveAssetsResult", + "type": "object" + }, + "JsonValue": { + "title": "JsonValue", + "type": [ + "object", + "array", + "string", + "number", + "boolean", + "null" + ] + }, + "LazyDeserializedDAG": { + "description": "Lazily build information from the serialized DAG structure.\n\nAn object that will present \"enough\" of the DAG like interface to update DAG db models etc, without having\nto deserialize the full DAG and Task hierarchy.", + "properties": { + "data": { + "additionalProperties": true, + "title": "Data", + "type": "object" + }, + "last_loaded": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Last Loaded" + } + }, + "required": [ + "data" + ], + "title": "LazyDeserializedDAG", + "type": "object" + }, + "MaskSecret": { + "description": "Add a new value to be redacted in task logs.", + "properties": { + "value": { + "$ref": "#/$defs/JsonValue" + }, + "name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Name" + }, + "type": { + "const": "MaskSecret", + "default": "MaskSecret", + "title": "Type", + "type": "string" + } + }, + "required": [ + "value" + ], + "title": "MaskSecret", + "type": "object" + }, + "OKResponse": { + "properties": { + "ok": { + "title": "Ok", + "type": "boolean" + }, + "type": { + "const": "OKResponse", + "default": "OKResponse", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ok" + ], + "title": "OKResponse", + "type": "object" + }, + "PrevSuccessfulDagRunResult": { + "properties": { + "data_interval_start": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Data Interval Start" + }, + "data_interval_end": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Data Interval End" + }, + "start_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Start Date" + }, + "end_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "End Date" + }, + "type": { + "const": "PrevSuccessfulDagRunResult", + "default": "PrevSuccessfulDagRunResult", + "title": "Type", + "type": "string" + } + }, + "title": "PrevSuccessfulDagRunResult", + "type": "object" + }, + "PreviousDagRunResult": { + "description": "Response containing previous Dag run information.", + "properties": { + "dag_run": { + "anyOf": [ + { + "$ref": "#/$defs/DagRun" + }, + { + "type": "null" + } + ], + "default": null + }, + "type": { + "const": "PreviousDagRunResult", + "default": "PreviousDagRunResult", + "title": "Type", + "type": "string" + } + }, + "title": "PreviousDagRunResult", + "type": "object" + }, + "PreviousTIResponse": { + "description": "Schema for response with previous TaskInstance information.", + "properties": { + "task_id": { + "title": "Task Id", + "type": "string" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "logical_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Logical Date" + }, + "start_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Start Date" + }, + "end_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "End Date" + }, + "state": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "State" + }, + "try_number": { + "title": "Try Number", + "type": "integer" + }, + "map_index": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": -1, + "title": "Map Index" + }, + "duration": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Duration" + } + }, + "required": [ + "task_id", + "dag_id", + "run_id", + "try_number" + ], + "title": "PreviousTIResponse", + "type": "object" + }, + "PreviousTIResult": { + "description": "Response containing previous task instance data.", + "properties": { + "task_instance": { + "anyOf": [ + { + "$ref": "#/$defs/PreviousTIResponse" + }, + { + "type": "null" + } + ], + "default": null + }, + "type": { + "const": "PreviousTIResult", + "default": "PreviousTIResult", + "title": "Type", + "type": "string" + } + }, + "title": "PreviousTIResult", + "type": "object" + }, + "PutVariable": { + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "value": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Value" + }, + "description": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Description" + }, + "type": { + "const": "PutVariable", + "default": "PutVariable", + "title": "Type", + "type": "string" + } + }, + "required": [ + "key", + "value", + "description" + ], + "title": "PutVariable", + "type": "object" + }, + "RescheduleTask": { + "additionalProperties": false, + "description": "Update a task instance state to reschedule/up_for_reschedule.", + "properties": { + "state": { + "const": "up_for_reschedule", + "default": "up_for_reschedule", + "title": "State", + "type": "string" + }, + "reschedule_date": { + "format": "date-time", + "title": "Reschedule Date", + "type": "string" + }, + "end_date": { + "format": "date-time", + "title": "End Date", + "type": "string" + }, + "type": { + "const": "RescheduleTask", + "default": "RescheduleTask", + "title": "Type", + "type": "string" + } + }, + "required": [ + "reschedule_date", + "end_date" + ], + "title": "RescheduleTask", + "type": "object" + }, + "ResendLoggingFD": { + "properties": { + "type": { + "const": "ResendLoggingFD", + "default": "ResendLoggingFD", + "title": "Type", + "type": "string" + } + }, + "title": "ResendLoggingFD", + "type": "object" + }, + "RetryTask": { + "additionalProperties": false, + "description": "Update a task instance state to up_for_retry.", + "properties": { + "state": { + "const": "up_for_retry", + "default": "up_for_retry", + "title": "State", + "type": "string" + }, + "end_date": { + "format": "date-time", + "title": "End Date", + "type": "string" + }, + "rendered_map_index": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Rendered Map Index" + }, + "retry_delay_seconds": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Retry Delay Seconds" + }, + "retry_reason": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Retry Reason" + }, + "type": { + "const": "RetryTask", + "default": "RetryTask", + "title": "Type", + "type": "string" + } + }, + "required": [ + "end_date" + ], + "title": "RetryTask", + "type": "object" + }, + "SentFDs": { + "properties": { + "type": { + "const": "SentFDs", + "default": "SentFDs", + "title": "Type", + "type": "string" + }, + "fds": { + "items": { + "type": "integer" + }, + "title": "Fds", + "type": "array" + } + }, + "required": [ + "fds" + ], + "title": "SentFDs", + "type": "object" + }, + "SetAssetStateStoreByName": { + "properties": { + "name": { + "title": "Name", + "type": "string" + }, + "key": { + "title": "Key", + "type": "string" + }, + "value": { + "$ref": "#/$defs/JsonValue" + }, + "type": { + "const": "SetAssetStateStoreByName", + "default": "SetAssetStateStoreByName", + "title": "Type", + "type": "string" + } + }, + "required": [ + "name", + "key", + "value" + ], + "title": "SetAssetStateStoreByName", + "type": "object" + }, + "SetAssetStateStoreByUri": { + "properties": { + "uri": { + "title": "Uri", + "type": "string" + }, + "key": { + "title": "Key", + "type": "string" + }, + "value": { + "$ref": "#/$defs/JsonValue" + }, + "type": { + "const": "SetAssetStateStoreByUri", + "default": "SetAssetStateStoreByUri", + "title": "Type", + "type": "string" + } + }, + "required": [ + "uri", + "key", + "value" + ], + "title": "SetAssetStateStoreByUri", + "type": "object" + }, + "SetRenderedFields": { + "description": "Payload for setting RTIF for a task instance.", + "properties": { + "rendered_fields": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "title": "Rendered Fields", + "type": "object" + }, + "type": { + "const": "SetRenderedFields", + "default": "SetRenderedFields", + "title": "Type", + "type": "string" + } + }, + "required": [ + "rendered_fields" + ], + "title": "SetRenderedFields", + "type": "object" + }, + "SetRenderedMapIndex": { + "description": "Payload for setting rendered_map_index for a task instance.", + "properties": { + "rendered_map_index": { + "title": "Rendered Map Index", + "type": "string" + }, + "type": { + "const": "SetRenderedMapIndex", + "default": "SetRenderedMapIndex", + "title": "Type", + "type": "string" + } + }, + "required": [ + "rendered_map_index" + ], + "title": "SetRenderedMapIndex", + "type": "object" + }, + "SetTaskStateStore": { + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "key": { + "title": "Key", + "type": "string" + }, + "value": { + "$ref": "#/$defs/JsonValue" + }, + "expires_at": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Expires At" + }, + "type": { + "const": "SetTaskStateStore", + "default": "SetTaskStateStore", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id", + "key", + "value", + "expires_at" + ], + "title": "SetTaskStateStore", + "type": "object" + }, + "SetXCom": { + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "value": { + "$ref": "#/$defs/JsonValue" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "task_id": { + "title": "Task Id", + "type": "string" + }, + "map_index": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Map Index" + }, + "dag_result": { + "default": false, + "title": "Dag Result", + "type": "boolean" + }, + "mapped_length": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Mapped Length" + }, + "type": { + "const": "SetXCom", + "default": "SetXCom", + "title": "Type", + "type": "string" + } + }, + "required": [ + "key", + "value", + "dag_id", + "run_id", + "task_id" + ], + "title": "SetXCom", + "type": "object" + }, + "SkipDownstreamTasks": { + "additionalProperties": false, + "description": "Update state of downstream tasks within a task instance to 'skipped', while updating current task to success state.", + "properties": { + "tasks": { + "items": { + "anyOf": [ + { + "type": "string" + }, + { + "maxItems": 2, + "minItems": 2, + "prefixItems": [ + { + "type": "string" + }, + { + "type": "integer" + } + ], + "type": "array" + } + ] + }, + "title": "Tasks", + "type": "array" + }, + "type": { + "const": "SkipDownstreamTasks", + "default": "SkipDownstreamTasks", + "title": "Type", + "type": "string" + } + }, + "required": [ + "tasks" + ], + "title": "SkipDownstreamTasks", + "type": "object" + }, + "StartupDetails": { + "properties": { + "ti": { + "$ref": "#/$defs/TaskInstance" + }, + "dag_rel_path": { + "title": "Dag Rel Path", + "type": "string" + }, + "bundle_info": { + "$ref": "#/$defs/BundleInfo" + }, + "start_date": { + "format": "date-time", + "title": "Start Date", + "type": "string" + }, + "ti_context": { + "$ref": "#/$defs/TIRunContext" + }, + "sentry_integration": { + "title": "Sentry Integration", + "type": "string" + }, + "type": { + "const": "StartupDetails", + "default": "StartupDetails", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti", + "dag_rel_path", + "bundle_info", + "start_date", + "ti_context", + "sentry_integration" + ], + "title": "StartupDetails", + "type": "object" + }, + "SucceedTask": { + "additionalProperties": false, + "description": "Update a task's state to success. Includes task_outlets and outlet_events for registering asset events.", + "properties": { + "state": { + "const": "success", + "default": "success", + "title": "State", + "type": "string" + }, + "end_date": { + "format": "date-time", + "title": "End Date", + "type": "string" + }, + "task_outlets": { + "anyOf": [ + { + "items": { + "$ref": "#/$defs/AssetProfile" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Task Outlets" + }, + "outlet_events": { + "anyOf": [ + { + "items": { + "additionalProperties": true, + "type": "object" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Outlet Events" + }, + "rendered_map_index": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Rendered Map Index" + }, + "type": { + "const": "SucceedTask", + "default": "SucceedTask", + "title": "Type", + "type": "string" + } + }, + "required": [ + "end_date" + ], + "title": "SucceedTask", + "type": "object" + }, + "TICount": { + "description": "Response containing count of Task Instances matching certain filters.", + "properties": { + "count": { + "title": "Count", + "type": "integer" + }, + "type": { + "const": "TICount", + "default": "TICount", + "title": "Type", + "type": "string" + } + }, + "required": [ + "count" + ], + "title": "TICount", + "type": "object" + }, + "TaskBreadcrumbsResult": { + "properties": { + "breadcrumbs": { + "items": { + "additionalProperties": true, + "type": "object" + }, + "title": "Breadcrumbs", + "type": "array" + }, + "type": { + "const": "TaskBreadcrumbsResult", + "default": "TaskBreadcrumbsResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "breadcrumbs" + ], + "title": "TaskBreadcrumbsResult", + "type": "object" + }, + "TaskCallbackRequest": { + "description": "Task callback status information.\n\nA Class with information about the success/failure TI callback to be executed. Currently, only failure\ncallbacks when tasks are externally killed or experience heartbeat timeouts are run via DagFileProcessorProcess.", + "properties": { + "filepath": { + "title": "Filepath", + "type": "string" + }, + "bundle_name": { + "title": "Bundle Name", + "type": "string" + }, + "bundle_version": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Bundle Version" + }, + "version_data": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Version Data" + }, + "msg": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Msg" + }, + "ti": { + "$ref": "#/$defs/TaskInstance" + }, + "task_callback_type": { + "anyOf": [ + { + "$ref": "#/$defs/TaskInstanceState" + }, + { + "type": "null" + } + ], + "default": null + }, + "context_from_server": { + "anyOf": [ + { + "$ref": "#/$defs/TIRunContext" + }, + { + "type": "null" + } + ], + "default": null + }, + "type": { + "const": "TaskCallbackRequest", + "default": "TaskCallbackRequest", + "title": "Type", + "type": "string" + } + }, + "required": [ + "filepath", + "bundle_name", + "bundle_version", + "ti" + ], + "title": "TaskCallbackRequest", + "type": "object" + }, + "TaskInstanceState": { + "description": "All possible states that a Task Instance can be in.\n\nNote that None is also allowed, so always use this in a type hint with Optional.", + "enum": [ + "removed", + "scheduled", + "queued", + "running", + "success", + "restarting", + "failed", + "up_for_retry", + "up_for_reschedule", + "upstream_failed", + "skipped", + "deferred", + "awaiting_input" + ], + "title": "TaskInstanceState", + "type": "string" + }, + "TaskRescheduleStartDate": { + "description": "Response containing the first reschedule date for a task instance.", + "properties": { + "start_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Start Date" + }, + "type": { + "const": "TaskRescheduleStartDate", + "default": "TaskRescheduleStartDate", + "title": "Type", + "type": "string" + } + }, + "required": [ + "start_date" + ], + "title": "TaskRescheduleStartDate", + "type": "object" + }, + "TaskState": { + "description": "Update a task's state.\n\nIf a process exits without sending one of these the state will be derived from the exit code:\n- 0 = SUCCESS\n- anything else = FAILED", + "properties": { + "state": { + "enum": [ + "failed", + "skipped", + "removed" + ], + "title": "State", + "type": "string" + }, + "end_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "End Date" + }, + "type": { + "const": "TaskState", + "default": "TaskState", + "title": "Type", + "type": "string" + }, + "rendered_map_index": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Rendered Map Index" + }, + "retry_reason": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Retry Reason" + } + }, + "required": [ + "state" + ], + "title": "TaskState", + "type": "object" + }, + "TaskStateStoreResult": { + "additionalProperties": false, + "description": "Response to GetTaskStateStore; wraps the generated API response for supervisor to worker comms.", + "properties": { + "value": { + "anyOf": [ + { + "$ref": "#/$defs/JsonValue" + }, + { + "type": "null" + } + ] + }, + "type": { + "const": "TaskStateStoreResult", + "default": "TaskStateStoreResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "value" + ], + "title": "TaskStateStoreResult", + "type": "object" + }, + "TaskStatesResult": { + "properties": { + "task_states": { + "additionalProperties": true, + "title": "Task States", + "type": "object" + }, + "type": { + "const": "TaskStatesResult", + "default": "TaskStatesResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "task_states" + ], + "title": "TaskStatesResult", + "type": "object" + }, + "TriggerDagRun": { + "additionalProperties": false, + "properties": { + "logical_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Logical Date" + }, + "run_after": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Run After" + }, + "conf": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Conf" + }, + "reset_dag_run": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "default": false, + "title": "Reset Dag Run" + }, + "partition_key": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Partition Key" + }, + "note": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Note" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Dag Run Id", + "type": "string" + }, + "type": { + "const": "TriggerDagRun", + "default": "TriggerDagRun", + "title": "Type", + "type": "string" + } + }, + "required": [ + "dag_id", + "run_id" + ], + "title": "TriggerDagRun", + "type": "object" + }, + "UpdateDagRunNote": { + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "note": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Note" + }, + "type": { + "const": "UpdateDagRunNote", + "default": "UpdateDagRunNote", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id", + "note" + ], + "title": "UpdateDagRunNote", + "type": "object" + }, + "UpdateHITLDetail": { + "description": "Update the response content part of an existing Human-in-the-loop response.", + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "chosen_options": { + "items": { + "type": "string" + }, + "minItems": 1, + "title": "Chosen Options", + "type": "array" + }, + "params_input": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Params Input" + }, + "type": { + "const": "UpdateHITLDetail", + "default": "UpdateHITLDetail", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id", + "chosen_options" + ], + "title": "UpdateHITLDetail", + "type": "object" + }, + "ValidateInletsAndOutlets": { + "properties": { + "ti_id": { + "format": "uuid", + "title": "Ti Id", + "type": "string" + }, + "type": { + "const": "ValidateInletsAndOutlets", + "default": "ValidateInletsAndOutlets", + "title": "Type", + "type": "string" + } + }, + "required": [ + "ti_id" + ], + "title": "ValidateInletsAndOutlets", + "type": "object" + }, + "VariableKeysResult": { + "properties": { + "keys": { + "items": { + "type": "string" + }, + "title": "Keys", + "type": "array" + }, + "total_entries": { + "title": "Total Entries", + "type": "integer" + }, + "type": { + "const": "VariableKeysResult", + "default": "VariableKeysResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "keys", + "total_entries" + ], + "title": "VariableKeysResult", + "type": "object" + }, + "VariableResponse": { + "additionalProperties": false, + "description": "Variable schema for responses with fields that are needed for Runtime.", + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "value": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Value" + } + }, + "required": [ + "key", + "value" + ], + "title": "VariableResponse", + "type": "object" + }, + "VariableResult": { + "additionalProperties": false, + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "value": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Value" + }, + "type": { + "const": "VariableResult", + "default": "VariableResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "key", + "value" + ], + "title": "VariableResult", + "type": "object" + }, + "XComCountResponse": { + "properties": { + "len": { + "title": "Len", + "type": "integer" + }, + "type": { + "const": "XComCountResponse", + "default": "XComCountResponse", + "title": "Type", + "type": "string" + } + }, + "required": [ + "len" + ], + "title": "XComCountResponse", + "type": "object" + }, + "XComResult": { + "description": "Response to ReadXCom request.", + "properties": { + "key": { + "title": "Key", + "type": "string" + }, + "value": { + "anyOf": [ + { + "$ref": "#/$defs/JsonValue" + }, + { + "type": "null" + } + ] + }, + "type": { + "const": "XComResult", + "default": "XComResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "key", + "value" + ], + "title": "XComResult", + "type": "object" + }, + "XComSequenceIndexResult": { + "properties": { + "root": { + "$ref": "#/$defs/JsonValue" + }, + "type": { + "const": "XComSequenceIndexResult", + "default": "XComSequenceIndexResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "root" + ], + "title": "XComSequenceIndexResult", + "type": "object" + }, + "XComSequenceSliceResult": { + "properties": { + "root": { + "items": { + "$ref": "#/$defs/JsonValue" + }, + "title": "Root", + "type": "array" + }, + "type": { + "const": "XComSequenceSliceResult", + "default": "XComSequenceSliceResult", + "title": "Type", + "type": "string" + } + }, + "required": [ + "root" + ], + "title": "XComSequenceSliceResult", + "type": "object" + }, + "ArgValueSchema": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "title": "ArgValueSchema", + "type": "object" + }, + "LiteralArgBinding": { + "description": "One positional stub-task argument carrying an inline literal from the Dag file.", + "properties": { + "name": { + "title": "Name", + "type": "string" + }, + "value_schema": { + "anyOf": [ + { + "$ref": "#/$defs/ArgValueSchema" + }, + { + "type": "null" + } + ], + "default": null + }, + "kind": { + "const": "literal", + "title": "Kind", + "type": "string" + }, + "value": { + "anyOf": [ + { + "$ref": "#/$defs/JsonValue" + }, + { + "type": "null" + } + ], + "default": null + }, + "from_default": { + "default": false, + "title": "From Default", + "type": "boolean" + } + }, + "required": [ + "name", + "kind" + ], + "title": "LiteralArgBinding", + "type": "object" + }, + "TaskArgBinding": { + "discriminator": { + "mapping": { + "literal": "#/$defs/LiteralArgBinding", + "xcom": "#/$defs/XComArgBinding" + }, + "propertyName": "kind" + }, + "oneOf": [ + { + "$ref": "#/$defs/XComArgBinding" + }, + { + "$ref": "#/$defs/LiteralArgBinding" + } + ], + "title": "TaskArgBinding" + }, + "XComArgBinding": { + "description": "One positional stub-task argument pulled from an upstream task's XCom.", + "properties": { + "name": { + "title": "Name", + "type": "string" + }, + "value_schema": { + "anyOf": [ + { + "$ref": "#/$defs/ArgValueSchema" + }, + { + "type": "null" + } + ], + "default": null + }, + "kind": { + "const": "xcom", + "title": "Kind", + "type": "string" + }, + "task_id": { + "title": "Task Id", + "type": "string" + } + }, + "required": [ + "name", + "kind", + "task_id" + ], + "title": "XComArgBinding", + "type": "object" + }, + "AssetEventDagRunReference": { + "additionalProperties": false, + "description": "Schema for AssetEvent model used in DagRun.", + "properties": { + "asset": { + "$ref": "#/$defs/AssetReferenceAssetEventDagRun" + }, + "extra": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "title": "Extra", + "type": "object" + }, + "source_task_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Source Task Id" + }, + "source_dag_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Source Dag Id" + }, + "source_run_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Source Run Id" + }, + "source_map_index": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Source Map Index" + }, + "source_aliases": { + "items": { + "$ref": "#/$defs/AssetAliasReferenceAssetEventDagRun" + }, + "title": "Source Aliases", + "type": "array" + }, + "timestamp": { + "format": "date-time", + "title": "Timestamp", + "type": "string" + }, + "partition_key": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Partition Key" + } + }, + "required": [ + "asset", + "extra", + "source_task_id", + "source_dag_id", + "source_run_id", + "source_map_index", + "source_aliases", + "timestamp" + ], + "title": "AssetEventDagRunReference", + "type": "object" + }, + "AssetReferenceAssetEventDagRun": { + "additionalProperties": false, + "description": "Schema for AssetModel used in AssetEventDagRunReference.", + "properties": { + "name": { + "title": "Name", + "type": "string" + }, + "uri": { + "title": "Uri", + "type": "string" + }, + "extra": { + "additionalProperties": { + "$ref": "#/$defs/JsonValue" + }, + "title": "Extra", + "type": "object" + } + }, + "required": [ + "name", + "uri", + "extra" + ], + "title": "AssetReferenceAssetEventDagRun", + "type": "object" + }, + "DagRun": { + "additionalProperties": false, + "description": "Schema for DagRun model with minimal required fields needed for Runtime.", + "properties": { + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "logical_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Logical Date" + }, + "data_interval_start": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Data Interval Start" + }, + "data_interval_end": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Data Interval End" + }, + "run_after": { + "format": "date-time", + "title": "Run After", + "type": "string" + }, + "start_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Start Date" + }, + "end_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "End Date" + }, + "clear_number": { + "default": 0, + "title": "Clear Number", + "type": "integer" + }, + "run_type": { + "$ref": "#/$defs/DagRunType" + }, + "state": { + "$ref": "#/$defs/DagRunState" + }, + "conf": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Conf" + }, + "triggering_user_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Triggering User Name" + }, + "consumed_asset_events": { + "items": { + "$ref": "#/$defs/AssetEventDagRunReference" + }, + "title": "Consumed Asset Events", + "type": "array" + }, + "partition_key": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Partition Key" + }, + "partition_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Partition Date" + }, + "note": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Note" + }, + "team_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Team Name" + } + }, + "required": [ + "dag_id", + "run_id", + "logical_date", + "data_interval_start", + "data_interval_end", + "run_after", + "start_date", + "end_date", + "run_type", + "state", + "consumed_asset_events", + "partition_key" + ], + "title": "DagRun", + "type": "object" + }, + "TIRunContext": { + "description": "Response schema for TaskInstance run context.", + "properties": { + "dag_run": { + "$ref": "#/$defs/DagRun" + }, + "task_reschedule_count": { + "default": 0, + "title": "Task Reschedule Count", + "type": "integer" + }, + "max_tries": { + "title": "Max Tries", + "type": "integer" + }, + "variables": { + "items": { + "$ref": "#/$defs/VariableResponse" + }, + "title": "Variables", + "type": "array" + }, + "connections": { + "items": { + "$ref": "#/$defs/ConnectionResponse" + }, + "title": "Connections", + "type": "array" + }, + "next_method": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Next Method" + }, + "next_kwargs": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Next Kwargs" + }, + "xcom_keys_to_clear": { + "items": { + "type": "string" + }, + "title": "Xcom Keys To Clear", + "type": "array" + }, + "should_retry": { + "default": false, + "title": "Should Retry", + "type": "boolean" + }, + "start_date": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Start Date" + }, + "arg_bindings": { + "anyOf": [ + { + "items": { + "$ref": "#/$defs/TaskArgBinding" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Arg Bindings" + }, + "multi_team": { + "default": false, + "title": "Multi Team", + "type": "boolean" + } + }, + "required": [ + "dag_run", + "max_tries" + ], + "title": "TIRunContext", + "type": "object" + }, + "TaskInstance": { + "description": "Schema for TaskInstance model with minimal required fields needed for Runtime.", + "properties": { + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "task_id": { + "title": "Task Id", + "type": "string" + }, + "dag_id": { + "title": "Dag Id", + "type": "string" + }, + "run_id": { + "title": "Run Id", + "type": "string" + }, + "try_number": { + "title": "Try Number", + "type": "integer" + }, + "dag_version_id": { + "format": "uuid", + "title": "Dag Version Id", + "type": "string" + }, + "map_index": { + "default": -1, + "title": "Map Index", + "type": "integer" + }, + "hostname": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Hostname" + }, + "context_carrier": { + "anyOf": [ + { + "additionalProperties": true, + "type": "object" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Context Carrier" + }, + "queue": { + "default": "default", + "title": "Queue", + "type": "string" + } + }, + "required": [ + "id", + "task_id", + "dag_id", + "run_id", + "try_number", + "dag_version_id" + ], + "title": "TaskInstance", + "type": "object" + } + } +} diff --git a/scripts/ci/prek/check_go_sdk_generated_drift.py b/scripts/ci/prek/check_go_sdk_generated_drift.py index f66dcae71b9ee..b0f5b80191afc 100755 --- a/scripts/ci/prek/check_go_sdk_generated_drift.py +++ b/scripts/ci/prek/check_go_sdk_generated_drift.py @@ -19,18 +19,21 @@ Keep the Go SDK's generated files in step with the schemas they generate from. Two of the Go SDK's surfaces are generated from schemas Python owns, and both are -committed, so nothing regenerates them when the schema on the Python side moves: +committed, so nothing regenerates them when the schema moves: * ``go-sdk/airflow/spec.gen.go`` — ``airflow.DagSpec`` and ``airflow.TaskSpec``, the - structs a Dag author fills in, from ``airflow-core``'s Dag serialization schema. + structs a Dag author fills in, from ``go-sdk/schema/dag-schema.json``. * ``go-sdk/pkg/execution/genmodels/*.gen.go`` — the coordinator-protocol messages, - from the supervisor wire-schema snapshot the Python Task SDK owns. + from ``go-sdk/schema/supervisor-schema.json``. -Without this check a property added, renamed or retyped on the Python side leaves the -Go side silently behind: a Dag authored in Go keeps serializing the old shape, and -msgpack drops a message field the Go struct does not declare. That is not theoretical -— ``models.gen.go`` sat two fields behind its snapshot across two releases, which is -what #73954 is about. +Both of those are go-sdk's vendored copies of a schema another distribution owns; +``sync-go-sdk-schemas`` is what keeps a copy equal to its original. This check takes +the copy as given and asks only whether the committed Go is what the copy generates. + +Without it a property added, renamed or retyped upstream leaves the Go side silently +behind: a Dag authored in Go keeps serializing the old shape, and msgpack drops a +message field the Go struct does not declare. That is not theoretical — ``models.gen.go`` +sat two fields behind its snapshot across two releases, which is what #73954 is about. The check regenerates each target and asks Git whether it changed. A drifted file is left regenerated in the working tree, so the fix is to commit it. @@ -70,7 +73,7 @@ class Target(NamedTuple): Target( package="./airflow/...", committed=(GO_SDK_MODULE / "airflow" / "spec.gen.go",), - schema="airflow-core/src/airflow/serialization/schema.json", + schema="go-sdk/schema/dag-schema.json", remedy=( "Review it — a property that should not reach an author belongs in the " "exclusion list in go-sdk/internal/genspec/authoring.go — then commit it:" @@ -83,7 +86,7 @@ class Target(NamedTuple): GO_SDK_MODULE / "pkg" / "execution" / "genmodels" / "discriminators.gen.go", GO_SDK_MODULE / "pkg" / "execution" / "genmodels" / "defaults.gen.go", ), - schema="task-sdk/src/airflow/sdk/execution_time/schema/schema.json", + schema="go-sdk/schema/supervisor-schema.json", remedy="Commit it:", ), ) @@ -127,7 +130,7 @@ def format_report( f"ERROR: regenerating {written} failed.", "", "A generator fails on a schema construct it has no rule for, which is how a", - "change on the Python side that needs a new rule surfaces; the generators also", + "newly vendored schema that needs a new rule surfaces; the generators also", "need the Go toolchain and the network to fetch go-jsonschema.", "`go generate` reported:", "", diff --git a/scripts/ci/prek/sync_go_sdk_schemas.py b/scripts/ci/prek/sync_go_sdk_schemas.py new file mode 100755 index 0000000000000..e0305921bfe2f --- /dev/null +++ b/scripts/ci/prek/sync_go_sdk_schemas.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +# 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. +""" +Refresh the Go SDK's vendored copies of the two schemas Python owns. + +``go-sdk`` generates its Dag-authoring structs from ``airflow-core``'s Dag serialization +schema and its coordinator-protocol models from the supervisor wire-schema snapshot the +Python Task SDK owns. It used to reach both by relative path across the monorepo, which +made a generated file only explainable from a directory the published Go module does not +contain, and made every change to a Python schema a change that has to carry regenerated +Go with it. It now vendors them under ``go-sdk/schema/``, the way ``ts-sdk`` and +``java-sdk`` already vendor theirs. + +Vendoring splits the one question ("is the Go behind Python?") into two: + +* this hook — is the copy equal to the source? Copying is mechanical, so it copies for + you and fails, the way ``sync-ts-sdk-dag-schema`` does. +* ``check-go-sdk-generated-drift`` — are the generated files what the copy generates? + That one needs the Go toolchain, and a new schema construct may need a generator rule + or an authoring exclusion, so what to do about it is a decision, not a copy. + +Run from the repo root: + + uv run --project scripts python scripts/ci/prek/sync_go_sdk_schemas.py + +Exits 0 when both copies already matched their source, 1 when it refreshed one. +""" + +from __future__ import annotations + +import pathlib +import shutil +import sys +from typing import NamedTuple + +REPO_ROOT = pathlib.Path(__file__).resolve().parents[3] + + +class VendoredSchema(NamedTuple): + """One schema Python owns and the copy ``go generate`` reads instead.""" + + source: pathlib.Path + vendored: pathlib.Path + # generated is what the copy feeds, and regenerate the recipe that rewrites it. + generated: str + regenerate: str + # also is what else has to move when this schema does, or "" when nothing does. + also: str + + +VENDORED_SCHEMAS = ( + VendoredSchema( + source=pathlib.Path("airflow-core/src/airflow/serialization/schema.json"), + vendored=pathlib.Path("go-sdk/schema/dag-schema.json"), + generated="go-sdk/airflow/spec.gen.go", + regenerate="just generate-specs", + also=( + "A property that should not reach a Dag author belongs in the exclusion list " + "in go-sdk/internal/genspec/authoring.go." + ), + ), + VendoredSchema( + source=pathlib.Path("task-sdk/src/airflow/sdk/execution_time/schema/schema.json"), + vendored=pathlib.Path("go-sdk/schema/supervisor-schema.json"), + generated="go-sdk/pkg/execution/genmodels/*.gen.go", + regenerate="just generate-models", + also=( + "When api_version moved, SupervisorSchemaVersion in " + "go-sdk/pkg/execution/messages.go has to move with it; " + "TestSupervisorSchemaVersionMatchesSnapshot fails until it does." + ), + ), +) + + +def refresh(schema: VendoredSchema, repo_root: pathlib.Path = REPO_ROOT) -> bool: + """Copy ``source`` over ``vendored`` when the two differ. Returns whether it copied.""" + source_path = repo_root / schema.source + vendored_path = repo_root / schema.vendored + if not source_path.is_file(): + raise SystemExit(f"{schema.source} is missing; cannot refresh {schema.vendored}") + if vendored_path.is_file() and vendored_path.read_bytes() == source_path.read_bytes(): + return False + vendored_path.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(source_path, vendored_path) + return True + + +def format_report(refreshed: tuple[VendoredSchema, ...]) -> tuple[int, str]: + """Turn what was refreshed into ``(exit_code, report)``.""" + if not refreshed: + copies = ", ".join(str(schema.vendored) for schema in VENDORED_SCHEMAS) + return 0, f"{copies} match the schemas they are vendored from." + lines = [] + for schema in refreshed: + lines.extend( + [ + f"Refreshed {schema.vendored} from {schema.source}.", + "", + f"Review the diff, regenerate what it feeds, and commit both it and {schema.generated}:", + "", + f" (cd go-sdk && {schema.regenerate})", + "", + ] + ) + if schema.also: + lines.extend([schema.also, ""]) + return 1, "\n".join(lines).rstrip() + + +def main() -> int: + refreshed = tuple(schema for schema in VENDORED_SCHEMAS if refresh(schema)) + exit_code, report = format_report(refreshed) + print(report, file=sys.stderr if exit_code else sys.stdout) + return exit_code + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/tests/ci/prek/test_check_go_sdk_generated_drift.py b/scripts/tests/ci/prek/test_check_go_sdk_generated_drift.py index de4efb557885a..44f03a182ff86 100644 --- a/scripts/tests/ci/prek/test_check_go_sdk_generated_drift.py +++ b/scripts/tests/ci/prek/test_check_go_sdk_generated_drift.py @@ -56,11 +56,18 @@ def test_both_targets_are_checked_and_name_generated_files_not_a_package_directo assert all(path.name.endswith(".gen.go") for path in target.committed), target.committed +def test_every_target_generates_from_a_schema_vendored_inside_the_go_module(): + # Reaching across the monorepo for a schema is what sync-go-sdk-schemas replaced: + # a path outside go-sdk/ does not exist in the published module or a source release. + for target in checker.TARGETS: + assert target.schema.startswith("go-sdk/schema/"), target.schema + + def test_current_files_pass(): exit_code, report = checker.format_report(SPECS, 0, "", 0, "") assert exit_code == 0 - assert "up to date with airflow-core/src/airflow/serialization/schema.json" in report + assert "up to date with go-sdk/schema/dag-schema.json" in report def test_drifted_specs_fail_with_the_diff_and_where_to_decide_about_a_property(): @@ -73,11 +80,11 @@ def test_drifted_specs_fail_with_the_diff_and_where_to_decide_about_a_property() assert "Deadline string" in report -def test_drifted_models_name_the_supervisor_snapshot_and_have_nothing_to_decide(): +def test_drifted_models_name_the_vendored_snapshot_and_have_nothing_to_decide(): exit_code, report = checker.format_report(MODELS, 0, "", 0, MODELS_DRIFT_DIFF) assert exit_code == 1 - assert "task-sdk/src/airflow/sdk/execution_time/schema/schema.json" in report + assert "go-sdk/schema/supervisor-schema.json" in report assert ( "git add go-sdk/pkg/execution/genmodels/models.gen.go " "go-sdk/pkg/execution/genmodels/discriminators.gen.go " diff --git a/scripts/tests/ci/prek/test_sync_go_sdk_schemas.py b/scripts/tests/ci/prek/test_sync_go_sdk_schemas.py new file mode 100644 index 0000000000000..3460d52d686f1 --- /dev/null +++ b/scripts/tests/ci/prek/test_sync_go_sdk_schemas.py @@ -0,0 +1,99 @@ +# 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. +from __future__ import annotations + +import pytest +import sync_go_sdk_schemas as syncer + +DAG_SCHEMA, SUPERVISOR_SCHEMA = syncer.VENDORED_SCHEMAS + + +@pytest.fixture +def repo(tmp_path): + """A checkout holding both sources and both vendored copies, all in step.""" + for schema in syncer.VENDORED_SCHEMAS: + for relative_path in (schema.source, schema.vendored): + path = tmp_path / relative_path + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(f'{{"title": "{schema.source.name}"}}') + return tmp_path + + +def test_both_schemas_are_vendored_inside_the_go_module(): + # A copy outside go-sdk/ would leave `go generate` reaching across the monorepo, + # which is the coupling vendoring removes. + for schema in syncer.VENDORED_SCHEMAS: + assert str(schema.vendored).startswith("go-sdk/schema/"), schema.vendored + assert not str(schema.source).startswith("go-sdk/"), schema.source + + +def test_a_copy_in_step_is_left_alone(repo): + assert syncer.refresh(DAG_SCHEMA, repo) is False + + +def test_a_stale_copy_is_overwritten_from_its_source(repo): + vendored_path = repo / SUPERVISOR_SCHEMA.vendored + vendored_path.write_text('{"title": "stale"}') + + assert syncer.refresh(SUPERVISOR_SCHEMA, repo) is True + assert vendored_path.read_bytes() == (repo / SUPERVISOR_SCHEMA.source).read_bytes() + + +def test_a_missing_copy_is_created(repo): + vendored_path = repo / DAG_SCHEMA.vendored + vendored_path.unlink() + + assert syncer.refresh(DAG_SCHEMA, repo) is True + assert vendored_path.read_bytes() == (repo / DAG_SCHEMA.source).read_bytes() + + +def test_a_missing_source_is_an_error_not_a_silent_pass(repo): + # Overwriting the copy with nothing, or reporting success, would both hide that the + # schema moved on the Python side. + (repo / DAG_SCHEMA.source).unlink() + + with pytest.raises(SystemExit, match=str(DAG_SCHEMA.source)): + syncer.refresh(DAG_SCHEMA, repo) + + +def test_copies_in_step_pass_and_name_what_was_compared(): + exit_code, report = syncer.format_report(()) + + assert exit_code == 0 + assert "go-sdk/schema/dag-schema.json" in report + assert "go-sdk/schema/supervisor-schema.json" in report + + +def test_a_refreshed_dag_schema_fails_and_says_what_to_regenerate(): + exit_code, report = syncer.format_report((DAG_SCHEMA,)) + + assert exit_code == 1 + assert "Refreshed go-sdk/schema/dag-schema.json" in report + assert "just generate-specs" in report + assert "go-sdk/airflow/spec.gen.go" in report + # A property that should not reach a Dag author is a decision, not a copy. + assert "go-sdk/internal/genspec/authoring.go" in report + + +def test_a_refreshed_supervisor_schema_points_at_the_version_constant(): + exit_code, report = syncer.format_report((SUPERVISOR_SCHEMA,)) + + assert exit_code == 1 + assert "just generate-models" in report + # api_version and SupervisorSchemaVersion have to move together. + assert "go-sdk/pkg/execution/messages.go" in report + assert "just generate-specs" not in report From 782d109f5fde91de57531e957a835b3cb48b445e Mon Sep 17 00:00:00 2001 From: henry3260 Date: Fri, 2 Oct 2026 13:43:25 +0800 Subject: [PATCH 2/2] Make the Go SDK schema sync a manual step, and assert its reports whole Review feedback: running the sync hook on every commit still fails the PR of whoever changed a Python schema, which is the coupling vendoring is meant to remove, so it becomes manual-only the way sync-java-sdk-dag-schema is. The trade-off is that nothing now announces a stale copy on its own. The three format_report tests assert the whole report rather than fragments of it, and a both-refreshed case covers how two refreshes are separated. --- .pre-commit-config.yaml | 10 +++- go-sdk/README.md | 18 ++++--- scripts/ci/prek/sync_go_sdk_schemas.py | 16 ++++-- .../tests/ci/prek/test_sync_go_sdk_schemas.py | 50 +++++++++++++++---- 4 files changed, 70 insertions(+), 24 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 60a2bb945ffa0..54b29e9cdd6c8 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -354,6 +354,13 @@ repos: entry: ./scripts/ci/prek/sync_go_sdk_schemas.py language: python pass_filenames: false + # Re-vendoring is a go-sdk maintainer's deliberate step after a schema change on + # the Python side, not something every commit should do, so this is manual only, + # the way sync-java-sdk-dag-schema above is: + # prek run sync-go-sdk-schemas --hook-stage manual + # Running it on every commit would make a change to a Python schema fail the PR + # of whoever made it, which is the coupling vendoring is here to remove. + stages: ['manual'] files: > (?x) ^airflow-core/src/airflow/serialization/schema\.json$| @@ -368,7 +375,8 @@ repos: # other checks that shell out to `go` in go-sdk get theirs. language: golang # The vendored schemas, not the Python originals: sync-go-sdk-schemas above is - # what ties those to the copies, and it needs neither Go nor the network. + # what ties those to the copies, as a manual step. This one runs on every commit, + # because a refreshed copy that nothing regenerated from is the drift that matters. files: > (?x) ^go-sdk/schema/.*\.json$| diff --git a/go-sdk/README.md b/go-sdk/README.md index eedf42a20050d..00ad8125e7374 100644 --- a/go-sdk/README.md +++ b/go-sdk/README.md @@ -388,11 +388,13 @@ is explainable from a file inside this module and a standalone checkout — a pu ASF source release — can regenerate and test without the monorepo around it. `ts-sdk` and `java-sdk` vendor theirs the same way. -Two prek hooks split the work that keeps the copies honest: `sync-go-sdk-schemas` copies an original -over its copy when the two differ and fails, since copying is mechanical; and -`check-go-sdk-generated-drift` regenerates from the copies and fails when the committed Go differs, -since what to do about a new schema construct — a generator rule, an authoring exclusion — is a -decision. +Two prek hooks split the work that keeps the copies honest. `sync-go-sdk-schemas` copies an original +over its copy when the two differ and fails, since copying is mechanical; it is a **manual** hook, so +re-vendoring is a deliberate step here rather than something a change to a Python schema forces on +whoever made it — which also means nothing announces on its own that a copy has gone stale. +`check-go-sdk-generated-drift` runs on every commit and regenerates from the copies, failing when the +committed Go differs, since what to do about a new schema construct — a generator rule, an authoring +exclusion — is a decision. ## Regenerating the coordinator-protocol models @@ -401,7 +403,7 @@ module's vendored copy of the supervisor schema snapshot the Python Task SDK own (`task-sdk/src/airflow/sdk/execution_time/schema/schema.json`); do not edit either by hand. To move the SDK to a newer schema version: -1. Refresh the copy: `prek run sync-go-sdk-schemas --all-files`, which overwrites +1. Refresh the copy: `prek run sync-go-sdk-schemas --hook-stage manual`, which overwrites `schema/supervisor-schema.json` from task-sdk's and fails so the change lands in review. 2. Set `SupervisorSchemaVersion` in [`pkg/execution/messages.go`](./pkg/execution/messages.go) to the copy's `api_version` date. @@ -418,8 +420,8 @@ when the committed files differ. `airflow.DagSpec` and `airflow.TaskSpec` in [`airflow/spec.gen.go`](./airflow/spec.gen.go) are generated from `schema/dag-schema.json`, this module's vendored copy of airflow-core's Dag serialization schema (`airflow-core/src/airflow/serialization/schema.json`), which Python owns; do -not edit either by hand. Refresh the copy with `prek run sync-go-sdk-schemas --all-files`, then run -`just generate-specs` after changing the schema or the generator. +not edit either by hand. Refresh the copy with `prek run sync-go-sdk-schemas --hook-stage manual`, +then run `just generate-specs` after changing the schema or the generator. The schema is the serialized shape rather than the authoring one, so [`internal/genspec/authoring.go`](./internal/genspec/authoring.go) holds the three tables that turn diff --git a/scripts/ci/prek/sync_go_sdk_schemas.py b/scripts/ci/prek/sync_go_sdk_schemas.py index e0305921bfe2f..38171925eb4d5 100755 --- a/scripts/ci/prek/sync_go_sdk_schemas.py +++ b/scripts/ci/prek/sync_go_sdk_schemas.py @@ -29,12 +29,20 @@ Vendoring splits the one question ("is the Go behind Python?") into two: * this hook — is the copy equal to the source? Copying is mechanical, so it copies for - you and fails, the way ``sync-ts-sdk-dag-schema`` does. + you and fails. It is a **manual** hook, the way ``sync-java-sdk-dag-schema`` is: + re-vendoring is a go-sdk maintainer's deliberate step, and running it on every commit + would fail the PR of whoever changed a Python schema, which is the coupling vendoring + is here to remove. Nothing therefore tells you on its own that a copy went stale. * ``check-go-sdk-generated-drift`` — are the generated files what the copy generates? - That one needs the Go toolchain, and a new schema construct may need a generator rule - or an authoring exclusion, so what to do about it is a decision, not a copy. + That one runs on every commit, because a refreshed copy nothing regenerated from is + the drift that matters, and a new schema construct may need a generator rule or an + authoring exclusion, so what to do about it is a decision, not a copy. -Run from the repo root: +Run it from the repo root, through prek: + + prek run sync-go-sdk-schemas --hook-stage manual + +or directly: uv run --project scripts python scripts/ci/prek/sync_go_sdk_schemas.py diff --git a/scripts/tests/ci/prek/test_sync_go_sdk_schemas.py b/scripts/tests/ci/prek/test_sync_go_sdk_schemas.py index 3460d52d686f1..95ac79c6ed3c1 100644 --- a/scripts/tests/ci/prek/test_sync_go_sdk_schemas.py +++ b/scripts/tests/ci/prek/test_sync_go_sdk_schemas.py @@ -70,30 +70,58 @@ def test_a_missing_source_is_an_error_not_a_silent_pass(repo): syncer.refresh(DAG_SCHEMA, repo) +# The reports are asserted whole rather than by fragment: this is the hook's entire +# output to whoever tripped it, so a recipe going missing or two refreshes running +# together should fail the test, not pass it quietly. +DAG_SCHEMA_REPORT = """\ +Refreshed go-sdk/schema/dag-schema.json from airflow-core/src/airflow/serialization/schema.json. + +Review the diff, regenerate what it feeds, and commit both it and go-sdk/airflow/spec.gen.go: + + (cd go-sdk && just generate-specs) + +A property that should not reach a Dag author belongs in the exclusion list in \ +go-sdk/internal/genspec/authoring.go.""" + +SUPERVISOR_SCHEMA_REPORT = """\ +Refreshed go-sdk/schema/supervisor-schema.json from \ +task-sdk/src/airflow/sdk/execution_time/schema/schema.json. + +Review the diff, regenerate what it feeds, and commit both it and \ +go-sdk/pkg/execution/genmodels/*.gen.go: + + (cd go-sdk && just generate-models) + +When api_version moved, SupervisorSchemaVersion in go-sdk/pkg/execution/messages.go has to \ +move with it; TestSupervisorSchemaVersionMatchesSnapshot fails until it does.""" + + def test_copies_in_step_pass_and_name_what_was_compared(): exit_code, report = syncer.format_report(()) assert exit_code == 0 - assert "go-sdk/schema/dag-schema.json" in report - assert "go-sdk/schema/supervisor-schema.json" in report + assert report == ( + "go-sdk/schema/dag-schema.json, go-sdk/schema/supervisor-schema.json " + "match the schemas they are vendored from." + ) def test_a_refreshed_dag_schema_fails_and_says_what_to_regenerate(): exit_code, report = syncer.format_report((DAG_SCHEMA,)) assert exit_code == 1 - assert "Refreshed go-sdk/schema/dag-schema.json" in report - assert "just generate-specs" in report - assert "go-sdk/airflow/spec.gen.go" in report - # A property that should not reach a Dag author is a decision, not a copy. - assert "go-sdk/internal/genspec/authoring.go" in report + assert report == DAG_SCHEMA_REPORT def test_a_refreshed_supervisor_schema_points_at_the_version_constant(): exit_code, report = syncer.format_report((SUPERVISOR_SCHEMA,)) assert exit_code == 1 - assert "just generate-models" in report - # api_version and SupervisorSchemaVersion have to move together. - assert "go-sdk/pkg/execution/messages.go" in report - assert "just generate-specs" not in report + assert report == SUPERVISOR_SCHEMA_REPORT + + +def test_both_refreshed_reports_each_one_separately(): + exit_code, report = syncer.format_report(syncer.VENDORED_SCHEMAS) + + assert exit_code == 1 + assert report == f"{DAG_SCHEMA_REPORT}\n\n{SUPERVISOR_SCHEMA_REPORT}"