This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 6317c63ae16 Go SDK: generate DagSpec and TaskSpec from the
serialization schema (#73936)
6317c63ae16 is described below
commit 6317c63ae16901a1fba5b17969e9ff12e35e94e2
Author: Henry Chen <[email protected]>
AuthorDate: Wed Sep 30 22:56:38 2026 +0800
Go SDK: generate DagSpec and TaskSpec from the serialization schema (#73936)
The two structs a Dag author fills in describe the same attributes
airflow-core's Dag serialization schema declares, and Python owns that
schema. Hand-maintained, they drift from it in silence: a property added,
renamed or retyped there leaves a Go-authored Dag serializing a shape
core no longer means, and nothing says so.
The schema is the serialized shape rather than the authoring one, so
generation cannot take it whole. It requires the paths and task lists the
bundle fills in, types a moment in time and a duration as a number of
seconds, carries a serialized timetable where an author writes a
schedule, and names none of the values a trigger rule may take. Those are
decisions about the SDK's surface, and the shape tables record each one
with the reason it was made, so that the next property to appear on the
Python side is a review question rather than an accident. Anything not
listed generates, which is what keeps a new property from vanishing.
---
.pre-commit-config.yaml | 17 +
go-sdk/Justfile | 4 +
go-sdk/README.md | 27 ++
go-sdk/airflow/dag.go | 9 +-
go-sdk/airflow/dag_test.go | 65 +++
go-sdk/airflow/spec.gen.go | 169 ++++++++
go-sdk/airflow/spec.go | 115 ++++-
go-sdk/airflow/spec_test.go | 71 ++++
go-sdk/internal/genlicense/genlicense.go | 71 ++++
go-sdk/internal/genlicense/genlicense_test.go | 83 ++++
go-sdk/internal/genspec/authoring.go | 470 +++++++++++++++++++++
go-sdk/internal/genspec/authoring_test.go | 329 +++++++++++++++
go-sdk/internal/genspec/main.go | 110 +++++
.../spec.go => internal/genspec/main_test.go} | 25 +-
go-sdk/internal/genspec/normalize.go | 176 ++++++++
go-sdk/internal/genspec/normalize_test.go | 256 +++++++++++
go-sdk/pkg/execution/genmodels/gen/main.go | 49 +--
scripts/ci/prek/check_go_sdk_spec_drift.py | 140 ++++++
.../tests/ci/prek/test_check_go_sdk_spec_drift.py | 133 ++++++
19 files changed, 2252 insertions(+), 67 deletions(-)
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 7fc98f8f3ab..3b6fe5200e1 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -348,6 +348,23 @@ repos:
# prek run sync-java-sdk-dag-schema --hook-stage manual
stages: ['manual']
files: ^java-sdk/sdk/schema/dag-schema\.json$
+ - id: check-go-sdk-spec-drift
+ name: Check Go SDK Dag spec structs against the serialization schema
+ description: "Regenerate go-sdk/airflow/spec.gen.go and fail when the
committed file differs"
+ entry: ./scripts/ci/prek/check_go_sdk_spec_drift.py
+ # 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
+ files: >
+ (?x)
+ ^airflow-core/src/airflow/serialization/schema\.json$|
+ ^go-sdk/airflow/spec\.gen\.go$|
+ ^go-sdk/airflow/spec\.go$|
+ ^go-sdk/internal/genspec/.*\.go$|
+ ^go-sdk/internal/genlicense/.*\.go$|
+ ^scripts/ci/prek/check_go_sdk_spec_drift\.py$
+ pass_filenames: false
+ require_serial: true
- id: check-go-version-in-sync
name: Check Go toolchain version is consistent across build files
entry: ./scripts/ci/prek/check_go_version_in_sync.py
diff --git a/go-sdk/Justfile b/go-sdk/Justfile
index 7094a394835..8445b9e1b92 100644
--- a/go-sdk/Justfile
+++ b/go-sdk/Justfile
@@ -42,3 +42,7 @@ docs port="6060":
# Regenerate the coordinator-protocol data models from the task-sdk supervisor
schema
generate-models:
go generate ./pkg/execution/genmodels/...
+
+# Regenerate the Dag and task spec structs from the core Dag serialization
schema
+generate-specs:
+ go generate ./airflow/...
diff --git a/go-sdk/README.md b/go-sdk/README.md
index e8fa38cf3e1..c4b32712b26 100644
--- a/go-sdk/README.md
+++ b/go-sdk/README.md
@@ -393,6 +393,33 @@ SDK to a newer schema version:
`TestSupervisorSchemaVersionMatchesSnapshot` fails when the constant and the
snapshot's `api_version`
drift, so a missed bump is caught by `go test` instead of needing a dedicated
prek hook.
+## 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.
+
+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
+it into the authoring shape, each entry carrying the reason it exists:
+
+- **exclusions** — properties an author never sets, such as the paths the
bundle fills in and the
+ template fields of a Python operator class. A property *not* excluded
generates, so one added on
+ the Python side surfaces in review instead of vanishing.
+- **type overrides** — the schema types a moment in time and a duration as a
number of seconds, and
+ an integral count as a JSON number.
+- **injections** — `Schedule`, which stands in for the serialized `timetable`.
Injecting into the
+ schema rather than hand-writing the field keeps every field in one struct
declaration, which is
+ what lets `TaskSpec` implement the sealed `TaskOption`.
+
+`trigger_rule` is typed as a plain string with no values named, so
`TriggerRule` and its constants
+are hand-written in [`airflow/spec.go`](./airflow/spec.go);
+`TestTriggerRuleConstantsMatchPython` is their tripwire against Airflow's own
enum.
+
+The `check-go-sdk-spec-drift` prek hook regenerates the file and fails when
the committed one
+differs, so a schema change that never reached Go cannot merge.
+
## Architectural decisions
The [`adr/`](./adr) directory records the design decisions behind the SDK:
diff --git a/go-sdk/airflow/dag.go b/go-sdk/airflow/dag.go
index 7870eb3fee1..2d46fbd886b 100644
--- a/go-sdk/airflow/dag.go
+++ b/go-sdk/airflow/dag.go
@@ -28,9 +28,8 @@ import (
// DagRef is a Dag authored in Go. [Dag] returns a new one.
type DagRef struct {
dagID string
- // Dag and Task copy the specs they are given, so a caller cannot
change a registered Dag
- // through a spec it still holds. A slice or map field in DagSpec or
TaskSpec would share its
- // contents with the caller, so Dag and Task would have to copy that
field too.
+ // Dag and Task copy the specs they are given with copySpec, so a
caller cannot change a
+ // registered Dag through a spec it still holds.
spec DagSpec
mu sync.Mutex
@@ -63,7 +62,7 @@ func Dag(dagID string, spec ...DagSpec) *DagRef {
}
d := &DagRef{dagID: dagID}
if len(spec) == 1 {
- d.spec = spec[0]
+ d.spec = copySpec(spec[0])
}
return d
}
@@ -175,7 +174,7 @@ func (d *DagRef) Task(fn any, opts ...TaskOption) *TaskRef {
}
d.taskIDs[taskID] = struct{}{}
- task := &TaskRef{taskID: taskID, spec: spec}
+ task := &TaskRef{taskID: taskID, spec: copySpec(spec)}
d.tasks = append(d.tasks, task)
return task
}
diff --git a/go-sdk/airflow/dag_test.go b/go-sdk/airflow/dag_test.go
index 431638f8e83..0ed707ab663 100644
--- a/go-sdk/airflow/dag_test.go
+++ b/go-sdk/airflow/dag_test.go
@@ -459,3 +459,68 @@ func TestTaskOptionRejectsForeignTypes(t *testing.T) {
assert.Contains(t, string(out), "foreignOption")
assert.Contains(t, string(out), "airflow.TaskOption")
}
+
+// withStorage returns a value of type t that points at something, which is
what makes a
+// copy observable: a copy points at storage of its own to compare against.
+func withStorage(t reflect.Type) reflect.Value {
+ switch t.Kind() {
+ case reflect.Pointer:
+ return reflect.New(t.Elem())
+ case reflect.Map:
+ m := reflect.MakeMap(t)
+ m.SetMapIndex(reflect.Zero(t.Key()), reflect.Zero(t.Elem()))
+ return m
+ default:
+ return reflect.MakeSlice(t, 1, 1)
+ }
+}
+
+// TestDagAndTaskCopyTheReferenceFieldsOfTheirSpecs pins the copy invariant
DagRef
+// documents. Assigning a spec copies a pointer, or a slice or map header, and
not what it
+// points at, so a caller that keeps what it passed could otherwise change a
registered Dag
+// through it. The fields are read from the spec types rather than named, so a
field added to
+// a generated spec is covered without this test being edited.
+func TestDagAndTaskCopyTheReferenceFieldsOfTheirSpecs(t *testing.T) {
+ for _, tt := range []struct {
+ spec any
+ store func(spec reflect.Value) reflect.Value
+ }{
+ {
+ spec: DagSpec{},
+ store: func(spec reflect.Value) reflect.Value {
+ return reflect.ValueOf(Dag("etl",
spec.Interface().(DagSpec)).spec)
+ },
+ },
+ {
+ spec: TaskSpec{},
+ store: func(spec reflect.Value) reflect.Value {
+ return reflect.ValueOf(Dag("etl").Task(extract,
spec.Interface().(TaskSpec)).spec)
+ },
+ },
+ } {
+ specType := reflect.TypeOf(tt.spec)
+ for i := range specType.NumField() {
+ field := specType.Field(i)
+ switch field.Type.Kind() {
+ case reflect.Pointer, reflect.Slice, reflect.Map:
+ default:
+ continue
+ }
+ t.Run(specType.Name()+"."+field.Name, func(t
*testing.T) {
+ given := reflect.New(specType).Elem()
+ given.Field(i).Set(withStorage(field.Type))
+
+ stored := tt.store(given).Field(i)
+
+ require.False(t, stored.IsNil(), "the field did
not reach the registered spec")
+ assert.NotEqual(
+ t,
+ given.Field(i).Pointer(),
+ stored.Pointer(),
+ "%s shares its contents with the
caller; copy it in Dag or Task",
+ field.Name,
+ )
+ })
+ }
+ }
+}
diff --git a/go-sdk/airflow/spec.gen.go b/go-sdk/airflow/spec.gen.go
new file mode 100644
index 00000000000..f1889915b58
--- /dev/null
+++ b/go-sdk/airflow/spec.gen.go
@@ -0,0 +1,169 @@
+// Code generated by github.com/atombender/go-jsonschema, DO NOT EDIT.
+// 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.
+
+package airflow
+
+import "time"
+
+// DagSpec holds the attributes of a Dag other than its dag_id. Dag takes one.
+type DagSpec struct {
+ // Catchup corresponds to the JSON schema field "catchup".
+ Catchup *bool
+
+ // DagDisplayName corresponds to the JSON schema field
"dag_display_name".
+ DagDisplayName string
+
+ // DagrunTimeout corresponds to the JSON schema field "dagrun_timeout".
+ DagrunTimeout time.Duration
+
+ // Description corresponds to the JSON schema field "description".
+ Description string
+
+ // DisableBundleVersioning corresponds to the JSON schema field
+ // "disable_bundle_versioning".
+ DisableBundleVersioning *bool
+
+ // DocMD corresponds to the JSON schema field "doc_md".
+ DocMD string
+
+ // EndDate corresponds to the JSON schema field "end_date".
+ EndDate time.Time
+
+ // FailFast corresponds to the JSON schema field "fail_fast".
+ FailFast bool
+
+ // IsPausedUponCreation corresponds to the JSON schema field
+ // "is_paused_upon_creation".
+ IsPausedUponCreation *bool
+
+ // MaxActiveRuns corresponds to the JSON schema field "max_active_runs".
+ MaxActiveRuns int
+
+ // MaxActiveTasks corresponds to the JSON schema field
"max_active_tasks".
+ MaxActiveTasks int
+
+ // MaxConsecutiveFailedDagRuns corresponds to the JSON schema field
+ // "max_consecutive_failed_dag_runs".
+ MaxConsecutiveFailedDagRuns *int
+
+ // RenderTemplateAsNativeObj corresponds to the JSON schema field
+ // "render_template_as_native_obj".
+ RenderTemplateAsNativeObj bool
+
+ // Schedule is the cron expression or preset the Dag runs on, such as
"@daily".
+ Schedule string
+
+ // StartDate corresponds to the JSON schema field "start_date".
+ StartDate time.Time
+
+ // Tags corresponds to the JSON schema field "tags".
+ Tags []string
+}
+
+// TaskSpec holds the attributes of a task. DagRef.Task takes at most one per
task.
+type TaskSpec struct {
+ // TaskDisplayName corresponds to the JSON schema field
"_task_display_name".
+ TaskDisplayName string
+
+ // DependsOnPast corresponds to the JSON schema field "depends_on_past".
+ DependsOnPast bool
+
+ // DoXComPush corresponds to the JSON schema field "do_xcom_push".
+ DoXComPush *bool
+
+ // DocMD corresponds to the JSON schema field "doc_md".
+ DocMD string
+
+ // EmailOnFailure corresponds to the JSON schema field
"email_on_failure".
+ EmailOnFailure *bool
+
+ // EmailOnRetry corresponds to the JSON schema field "email_on_retry".
+ EmailOnRetry *bool
+
+ // EndDate corresponds to the JSON schema field "end_date".
+ EndDate time.Time
+
+ // ExecutionTimeout corresponds to the JSON schema field
"execution_timeout".
+ ExecutionTimeout time.Duration
+
+ // Executor corresponds to the JSON schema field "executor".
+ Executor string
+
+ // IgnoreFirstDependsOnPast corresponds to the JSON schema field
+ // "ignore_first_depends_on_past".
+ IgnoreFirstDependsOnPast bool
+
+ // MapIndexTemplate corresponds to the JSON schema field
"map_index_template".
+ MapIndexTemplate string
+
+ // MaxActiveTisPerDag corresponds to the JSON schema field
+ // "max_active_tis_per_dag".
+ MaxActiveTisPerDag int
+
+ // MaxActiveTisPerDagrun corresponds to the JSON schema field
+ // "max_active_tis_per_dagrun".
+ MaxActiveTisPerDagrun int
+
+ // MaxRetryDelay corresponds to the JSON schema field "max_retry_delay".
+ MaxRetryDelay time.Duration
+
+ // Owner corresponds to the JSON schema field "owner".
+ Owner string
+
+ // Pool corresponds to the JSON schema field "pool".
+ Pool string
+
+ // PoolSlots corresponds to the JSON schema field "pool_slots".
+ PoolSlots *int
+
+ // PriorityWeight corresponds to the JSON schema field
"priority_weight".
+ PriorityWeight *int
+
+ // Queue corresponds to the JSON schema field "queue".
+ Queue string
+
+ // Retries corresponds to the JSON schema field "retries".
+ Retries int
+
+ // RetryDelay corresponds to the JSON schema field "retry_delay".
+ RetryDelay *time.Duration
+
+ // RetryExponentialBackoff corresponds to the JSON schema field
+ // "retry_exponential_backoff".
+ RetryExponentialBackoff float64
+
+ // StartDate corresponds to the JSON schema field "start_date".
+ StartDate time.Time
+
+ // TaskID is the task_id of the task. When TaskID is empty, the task_id
is the
+ // name of the Go function that the task runs.
+ TaskID string
+
+ // TriggerRule corresponds to the JSON schema field "trigger_rule".
+ TriggerRule TriggerRule
+
+ // WaitForDownstream corresponds to the JSON schema field
"wait_for_downstream".
+ WaitForDownstream bool
+
+ // WaitForPastDependsBeforeSkipping corresponds to the JSON schema field
+ // "wait_for_past_depends_before_skipping".
+ WaitForPastDependsBeforeSkipping bool
+
+ // WeightRule corresponds to the JSON schema field "weight_rule".
+ WeightRule WeightRule
+}
diff --git a/go-sdk/airflow/spec.go b/go-sdk/airflow/spec.go
index 20621ded56f..4b928999d4e 100644
--- a/go-sdk/airflow/spec.go
+++ b/go-sdk/airflow/spec.go
@@ -17,12 +17,111 @@
package airflow
-// DagSpec holds the attributes of a Dag other than its dag_id. [Dag] takes
one.
-type DagSpec struct{}
-
-// TaskSpec holds the attributes of a task. [DagRef.Task] takes at most one
per task.
-type TaskSpec struct {
- // TaskID is the task_id of the task. When TaskID is empty, the task_id
is the name of the
- // Go function that the task runs.
- TaskID string
+import "reflect"
+
+// DagSpec and TaskSpec are generated from Airflow core's Dag serialization
schema,
+// which Python owns, so that neither struct drifts from it silently. genspec
+// rewrites the schema into the authoring shape, go-jsonschema writes the
structs,
+// and genspec puts the license header back on what it wrote. The rewritten
schema
+// is a build artifact under .build; spec.gen.go is committed.
+//
+// To change a field, change the schema on the Python side, or the exclusions,
type
+// overrides and injected properties in internal/genspec/authoring.go, and run
+// `just generate-specs`.
+//
+// The structs carry no struct tags, because they are the authoring shape and
not the
+// wire format: encoding/json would write a time.Duration as the nanoseconds
Go counts it
+// in where the schema means seconds, and omitempty would drop a value an
author set to
+// 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 github.com/atombender/[email protected] --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
+
+// TriggerRule is when a task runs, given the state of the tasks upstream of
it, and
+// WeightRule is how a task's priority weight is worked out from the graph
around it.
+// The serialization schema types both as a plain string and names none of the
values
+// either takes, so the constants are written here and the generated fields
are given
+// these types by internal/genspec/authoring.go.
+type TriggerRule string
+
+const (
+ TriggerRuleAllSuccess TriggerRule = "all_success"
+ TriggerRuleAllFailed TriggerRule = "all_failed"
+ TriggerRuleAllDone TriggerRule = "all_done"
+ TriggerRuleAllDoneMinOneSuccess TriggerRule =
"all_done_min_one_success"
+ TriggerRuleAllDoneSetupSuccess TriggerRule =
"all_done_setup_success"
+ TriggerRuleOneSuccess TriggerRule = "one_success"
+ TriggerRuleOneFailed TriggerRule = "one_failed"
+ TriggerRuleOneDone TriggerRule = "one_done"
+ TriggerRuleNoneFailed TriggerRule = "none_failed"
+ TriggerRuleNoneFailedMinOneSuccess TriggerRule =
"none_failed_min_one_success"
+ TriggerRuleNoneSkipped TriggerRule = "none_skipped"
+ TriggerRuleAllSkipped TriggerRule = "all_skipped"
+ TriggerRuleAlways TriggerRule = "always"
+)
+
+type WeightRule string
+
+const (
+ WeightRuleDownstream WeightRule = "downstream"
+ WeightRuleUpstream WeightRule = "upstream"
+ WeightRuleAbsolute WeightRule = "absolute"
+)
+
+// copySpec returns a copy of spec that shares no slice, map or pointer with
it, so that a
+// caller changing what it still holds cannot change a registered Dag.
Assigning a spec
+// copies a slice header or a pointer and not what it points at, which is the
sharing this
+// undoes.
+//
+// It walks the spec with reflection rather than naming the fields because the
spec types
+// are generated: a field named here would have to be added again every time
the
+// serialization schema grows one, and the copy would go quietly missing until
someone
+// noticed a registered Dag changing underneath them.
+func copySpec[T any](spec T) T {
+ return copyValue(reflect.ValueOf(spec)).Interface().(T)
+}
+
+func copyValue(v reflect.Value) reflect.Value {
+ switch v.Kind() {
+ case reflect.Pointer:
+ if v.IsNil() {
+ return v
+ }
+ out := reflect.New(v.Type().Elem())
+ out.Elem().Set(copyValue(v.Elem()))
+ return out
+ case reflect.Slice:
+ if v.IsNil() {
+ return v
+ }
+ out := reflect.MakeSlice(v.Type(), v.Len(), v.Len())
+ for i := range v.Len() {
+ out.Index(i).Set(copyValue(v.Index(i)))
+ }
+ return out
+ case reflect.Map:
+ if v.IsNil() {
+ return v
+ }
+ out := reflect.MakeMapWithSize(v.Type(), v.Len())
+ for iter := v.MapRange(); iter.Next(); {
+ out.SetMapIndex(copyValue(iter.Key()),
copyValue(iter.Value()))
+ }
+ return out
+ case reflect.Struct:
+ out := reflect.New(v.Type()).Elem()
+ // Assigning first leaves an unexported field the shallow copy
Go itself would
+ // make; only the fields reflection can set are replaced by a
copy of their own.
+ out.Set(v)
+ for i := range v.NumField() {
+ if field := out.Field(i); field.CanSet() {
+ field.Set(copyValue(v.Field(i)))
+ }
+ }
+ return out
+ default:
+ return v
+ }
}
diff --git a/go-sdk/airflow/spec_test.go b/go-sdk/airflow/spec_test.go
new file mode 100644
index 00000000000..2221bd56e91
--- /dev/null
+++ b/go-sdk/airflow/spec_test.go
@@ -0,0 +1,71 @@
+// 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.
+
+package airflow
+
+import (
+ "os"
+ "path/filepath"
+ "regexp"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// specPath is where the Go half of each value set is declared, reached from
this package.
+const specPath = "spec.go"
+
+func readLines(t *testing.T, path, pattern string) []string {
+ t.Helper()
+
+ body, err := os.ReadFile(filepath.FromSlash(path))
+ require.NoError(t, err)
+ matches :=
regexp.MustCompile(pattern).FindAllStringSubmatch(string(body), -1)
+ require.NotEmpty(t, matches, "%s matched nothing, so the file's shape
has changed", path)
+ values := make([]string, 0, len(matches))
+ for _, match := range matches {
+ values = append(values, match[1])
+ }
+ return values
+}
+
+// TestRuleConstantsMatchPython is the tripwire for a trigger rule or a weight
rule
+// added or renamed on the Python side. The serialization schema types both
fields as a
+// plain string and names none of their values, so the constants here are
hand-written
+// and the spec drift check cannot see them; Airflow rejects a rule it does
not know, so
+// a missing constant is a Dag that fails to register.
+func TestRuleConstantsMatchPython(t *testing.T) {
+ for _, tt := range []struct {
+ goType string
+ pythonPath string
+ }{
+ {goType: "TriggerRule", pythonPath:
"../../airflow-core/src/airflow/task/trigger_rule.py"},
+ {goType: "WeightRule", pythonPath:
"../../airflow-core/src/airflow/task/weight_rule.py"},
+ } {
+ t.Run(tt.goType, func(t *testing.T) {
+ python := readLines(t, tt.pythonPath, `(?m)^ [A-Z_]+
= "([a-z_]+)"$`)
+ spelled := readLines(
+ t,
+ specPath,
+ `(?m)^\t`+tt.goType+`\w+ +`+tt.goType+` =
"([a-z_]+)"$`,
+ )
+
+ assert.ElementsMatch(t, python, spelled)
+ })
+ }
+}
diff --git a/go-sdk/internal/genlicense/genlicense.go
b/go-sdk/internal/genlicense/genlicense.go
new file mode 100644
index 00000000000..b7bc6e42e04
--- /dev/null
+++ b/go-sdk/internal/genlicense/genlicense.go
@@ -0,0 +1,71 @@
+// 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.
+
+// Package genlicense inserts the Apache license header into a generated Go
file.
+// A generator that writes a file itself carries the header in its template;
one
+// that shells out to go-jsonschema, which emits none, calls EnsureHeader so
that
+// the committed file is reproducible from go generate alone and a drift check
sees
+// no difference.
+package genlicense
+
+import (
+ "bytes"
+ "os"
+ "strings"
+)
+
+const header = `// 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.`
+
+// EnsureHeader inserts the header into the file at path, after the "Code
+// generated" line that has to stay first for golines to leave it alone. It is
+// idempotent, so a generator can call it on every run.
+func EnsureHeader(path string) error {
+ src, err := os.ReadFile(path)
+ if err != nil {
+ return err
+ }
+ if bytes.Contains(src, []byte("Licensed to the Apache Software
Foundation")) {
+ return nil
+ }
+ lines := strings.Split(string(src), "\n")
+ insertAt := 0
+ for i, line := range lines {
+ if strings.HasPrefix(line, "// Code generated") {
+ insertAt = i + 1
+ break
+ }
+ }
+ out := append([]string{}, lines[:insertAt]...)
+ out = append(out, strings.Split(header, "\n")...)
+ out = append(out, lines[insertAt:]...)
+ return os.WriteFile(path, []byte(strings.Join(out, "\n")), 0o644)
+}
diff --git a/go-sdk/internal/genlicense/genlicense_test.go
b/go-sdk/internal/genlicense/genlicense_test.go
new file mode 100644
index 00000000000..d2961e92b65
--- /dev/null
+++ b/go-sdk/internal/genlicense/genlicense_test.go
@@ -0,0 +1,83 @@
+// 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.
+
+package genlicense_test
+
+import (
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/airflow/go-sdk/internal/genlicense"
+)
+
+func writeFile(t *testing.T, body string) string {
+ t.Helper()
+
+ path := filepath.Join(t.TempDir(), "out.gen.go")
+ require.NoError(t, os.WriteFile(path, []byte(body), 0o644))
+ return path
+}
+
+func readFile(t *testing.T, path string) string {
+ t.Helper()
+
+ body, err := os.ReadFile(path)
+ require.NoError(t, err)
+ return string(body)
+}
+
+func TestEnsureHeaderInsertsAfterTheCodeGeneratedLine(t *testing.T) {
+ path := writeFile(t, "// Code generated by a tool, DO NOT
EDIT.\n\npackage out\n")
+
+ require.NoError(t, genlicense.EnsureHeader(path))
+
+ lines := strings.Split(readFile(t, path), "\n")
+ assert.Equal(t, "// Code generated by a tool, DO NOT EDIT.", lines[0],
+ "golines leaves the file alone only while this line is first")
+ assert.Equal(t, "// Licensed to the Apache Software Foundation (ASF)
under one", lines[1])
+ assert.Contains(t, readFile(t, path), "\npackage out\n")
+}
+
+func TestEnsureHeaderIsIdempotent(t *testing.T) {
+ path := writeFile(t, "// Code generated by a tool, DO NOT
EDIT.\n\npackage out\n")
+
+ require.NoError(t, genlicense.EnsureHeader(path))
+ once := readFile(t, path)
+ require.NoError(t, genlicense.EnsureHeader(path))
+
+ assert.Equal(t, once, readFile(t, path), "a generator calls it on every
run")
+}
+
+func TestEnsureHeaderPrependsWhenThereIsNoCodeGeneratedLine(t *testing.T) {
+ path := writeFile(t, "package out\n")
+
+ require.NoError(t, genlicense.EnsureHeader(path))
+
+ assert.True(
+ t,
+ strings.HasPrefix(readFile(t, path), "// Licensed to the Apache
Software Foundation"),
+ )
+}
+
+func TestEnsureHeaderReportsAMissingFile(t *testing.T) {
+ assert.Error(t, genlicense.EnsureHeader(filepath.Join(t.TempDir(),
"absent.go")))
+}
diff --git a/go-sdk/internal/genspec/authoring.go
b/go-sdk/internal/genspec/authoring.go
new file mode 100644
index 00000000000..db0b707e582
--- /dev/null
+++ b/go-sdk/internal/genspec/authoring.go
@@ -0,0 +1,470 @@
+// 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.
+
+package main
+
+import (
+ "encoding/json"
+ "errors"
+ "fmt"
+)
+
+// authoringShape rewrites the two definitions the airflow package generates
from
+// into the shape a Dag author writes, rather than the shape Airflow
serializes.
+// Each definition drops the properties in exclude, rewrites the properties in
+// override, and gains the properties in inject.
+type authoringShape struct {
+ // doc becomes the description of the definition, which go-jsonschema
writes as
+ // the doc comment of the generated type.
+ doc string
+ // exclude names each property that must not reach the generated
struct, mapped
+ // to why. A property absent from the list generates, so that a
property added
+ // on the Python side surfaces in review rather than vanishing; the
reason is
+ // what a reviewer reads when deciding whether a new one belongs here.
+ exclude map[string]string
+ // override rewrites a property that generates as the wrong Go type. The
+ // serialization schema types a moment in time and a duration as a
number of
+ // seconds and an integral count as a JSON number, none of which is the
type an
+ // author sets.
+ override map[string]propertyOverride
+ // inject adds a property the schema has no counterpart for, so that
every field
+ // of the generated struct comes from generation and the struct stays
one
+ // declaration.
+ inject map[string]map[string]any
+}
+
+// propertyOverride is the part of a property genspec rewrites. goType and
imports
+// become go-jsonschema's goJSONSchema extension, which it reads before a
$ref, so
+// an override applies to a property written as a reference too.
+type propertyOverride struct {
+ goType string
+ imports []string
+ // doc replaces the description, and so the doc comment of the
generated field,
+ // where the serialized property has nothing to say about how an author
sets it.
+ doc string
+ // pointer forces a pointer where setPointers cannot tell that the Go
zero value is
+ // a setting an author could mean. Whether 0 is a legal value for a
count is not
+ // something the schema says, so it is stated here for the counts where
it is.
+ pointer bool
+}
+
+var authoringShapes = map[string]authoringShape{
+ "dag": dagShape,
+ "operator": taskShape,
+}
+
+var dagShape = authoringShape{
+ doc: "DagSpec holds the attributes of a Dag other than its dag_id. Dag
takes one.",
+ exclude: map[string]string{
+ "dag_id": "a positional parameter of
airflow.Dag, not a spec field",
+ "fileloc": "the path of the Dag file, which
the bundle fills in",
+ "relative_fileloc": "the path of the Dag file, which
the bundle fills in",
+ "_processor_dags_folder": "the Dag processor's own folder,
filled in at parse time",
+ "bundle_name": "the name of the bundle that
carries the Dag, not the Dag's",
+ "tasks": "the tasks dag.Task registers",
+ "task_group": "the groups dag.TaskGroup
registers",
+ "edge_info": "the labels airflow.Label carries
into an edge verb",
+ "dag_dependencies": "derived from the edges and the
assets a Dag declares",
+ "timezone": "carried by the time.Time an
author sets on StartDate",
+ "timetable": "the serialized form of Schedule,
which is injected instead",
+ "allowed_run_types": "a union the author expresses by
setting Schedule",
+ "_concurrency": "the pre-2.2 spelling of
MaxActiveTasks",
+ "has_on_success_callback": "derived from whether a callback
is registered",
+ "has_on_failure_callback": "derived from whether a callback
is registered",
+ "params": "no Go authoring type yet: a param
carries a schema of its own",
+ "default_args": "no Go authoring type yet: the
values are arbitrary and untyped",
+ "access_control": "no Go authoring type yet, and it
is deprecated in Airflow 3",
+ "owner_links": "no Go authoring type yet: an
object of arbitrary link targets",
+ "deadline": "no Go authoring type yet: a
serialized deadline reference",
+ "rerun_with_latest_version": "no Go authoring type yet: the
tri-state a null allows",
+ },
+ override: map[string]propertyOverride{
+ "start_date": {goType: "time.Time", imports:
[]string{"time"}},
+ "end_date": {goType: "time.Time", imports:
[]string{"time"}},
+ "dagrun_timeout": {goType: "time.Duration", imports:
[]string{"time"}},
+ "max_active_tasks": {goType: "int"},
+ "max_active_runs": {goType: "int"},
+ // 0 means "never pause this Dag", and the default comes from
+ // [core] max_consecutive_failed_dag_runs_per_dag, which a
deployment can set
+ // above 0, so an author has to be able to say 0 and be heard.
+ "max_consecutive_failed_dag_runs": {goType: "int", pointer:
true},
+ "tags": {goType: "[]string"},
+ },
+ inject: map[string]map[string]any{
+ // The schema carries the serialized timetable this resolves
to, never the
+ // expression an author writes.
+ "schedule": {
+ "type": "string",
+ "description": "Schedule is the cron expression or
preset the Dag runs on, such as \"@daily\".",
+ },
+ },
+}
+
+var taskShape = authoringShape{
+ doc: "TaskSpec holds the attributes of a task. DagRef.Task takes at
most one per task.",
+ exclude: map[string]string{
+ "task_type": "the operator class name,
which the SDK fills in",
+ "_task_module": "the operator's Python module,
which the SDK fills in",
+ "_operator_extra_links": "links a Python operator class
declares, which a Go task has none of",
+ "ui_color": "the grid colour, which the
SDK fills in",
+ "ui_fgcolor": "the grid colour, which the
SDK fills in",
+ "template_fields": "the templated attributes of a
Python operator class",
+ "template_ext": "the templated attributes of a
Python operator class",
+ "template_fields_renderers": "the templated attributes of a
Python operator class",
+ "downstream_task_ids": "the edges Before, After and
Inputs declare",
+ "partial_kwargs": "the serialized form of a
mapped task's partial arguments",
+ "_logger_name": "the logger the task runner
names",
+ "_needs_expansion": "derived from whether the task
is mapped",
+ "_is_mapped": "derived from whether the task
is mapped",
+ "_is_sensor": "derived from the task's own
kind",
+ "_disallow_kwargs_override": "a mapped-task serialization
detail",
+ "_expand_input_attr": "a mapped-task serialization
detail",
+ "_arg_bindings": "the bindings airflow.Inputs
records",
+ "has_on_execute_callback": "derived from whether a
callback is registered",
+ "has_on_failure_callback": "derived from whether a
callback is registered",
+ "has_on_skipped_callback": "derived from whether a
callback is registered",
+ "has_on_success_callback": "derived from whether a
callback is registered",
+ "has_on_retry_callback": "derived from whether a
callback is registered",
+ "start_from_trigger": "deferral is Python's, per
decision 11 of ADR 8",
+ "start_trigger_args": "deferral is Python's, per
decision 11 of ADR 8",
+ "multiple_outputs": "derived from the Go
function's return type",
+ "params": "no Go authoring type yet: a
param carries a schema of its own",
+ "executor_config": "no Go authoring type yet: the
keys are executor-specific",
+ "inlets": "no Go authoring type yet: an
asset needs its own spec",
+ "outlets": "no Go authoring type yet: an
asset needs its own spec",
+ "render_template_as_native_obj": "set on the Dag, where the
schema types it without a null",
+ // 4 and 5: an attribute an author can set but the SDK cannot
yet honour, and one
+ // that only Python has, are worse than a missing field: a
field is easy to add
+ // later and hard to take away.
+ "is_setup": "setup/teardown needs trigger-rule
handling the SDK does not model yet",
+ "is_teardown": "setup/teardown needs trigger-rule
handling the SDK does not model yet",
+ "on_failure_fail_dagrun": "only meaningful on a teardown task",
+ "allow_nested_operators": "Python-only: it warns when an
operator executes inside another",
+ "doc": "a legacy rendering of the task's
docs; only doc_md is exposed",
+ "doc_json": "a legacy rendering of the task's
docs; only doc_md is exposed",
+ "doc_rst": "a legacy rendering of the task's
docs; only doc_md is exposed",
+ "doc_yaml": "a legacy rendering of the task's
docs; only doc_md is exposed",
+ },
+ override: map[string]propertyOverride{
+ "start_date": {goType: "time.Time", imports:
[]string{"time"}},
+ "end_date": {goType: "time.Time", imports:
[]string{"time"}},
+ "execution_timeout": {goType: "time.Duration", imports:
[]string{"time"}},
+ "retry_delay": {goType: "time.Duration", imports:
[]string{"time"}},
+ "max_retry_delay": {goType: "time.Duration", imports:
[]string{"time"}},
+ "retries": {goType: "int"},
+ "pool_slots": {goType: "int"},
+ "priority_weight": {goType: "int"},
+ "max_active_tis_per_dag": {goType: "int"},
+ "max_active_tis_per_dagrun": {goType: "int"},
+ // TriggerRule, WeightRule and their constants are hand-written
in the airflow
+ // package: the schema types both fields as a plain string and
names none of
+ // the values either one takes.
+ "trigger_rule": {goType: "TriggerRule"},
+ "weight_rule": {goType: "WeightRule"},
+ // A multiplier, not a switch: 0 keeps the delay constant, 2.0
doubles it each
+ // retry. The schema's number is right, and the float is what
carries the 2.0.
+ "retry_exponential_backoff": {goType: "float64"},
+ "task_id": {
+ goType: "string",
+ doc: "TaskID is the task_id of the task. When TaskID
is empty, the task_id is the name of the Go function that the task runs.",
+ },
+ },
+}
+
+// shapeForAuthoring rewrites doc into the schema the spec structs generate
from:
+// each definition in shapes takes its authoring shape, and everything the
airflow
+// package does not generate is dropped.
+func shapeForAuthoring(doc map[string]any, shapes map[string]authoringShape)
error {
+ if err := applyAuthoringShapes(doc, shapes); err != nil {
+ return err
+ }
+ return keepOnlySpecGeneratingSchema(doc, shapes)
+}
+
+// applyAuthoringShapes rewrites each definition in shapes. It reports a list
entry
+// that no longer matches the schema — an excluded or overridden property that
has
+// gone, an injected property the schema has grown — because each of those
means the
+// list here decides nothing and the generated struct would silently change
shape.
+func applyAuthoringShapes(doc map[string]any, shapes
map[string]authoringShape) error {
+ definitions, ok := doc["definitions"].(map[string]any)
+ if !ok {
+ return errNoDefinitions
+ }
+ for _, name := range sortedKeys(shapes) {
+ definition, ok := definitions[name].(map[string]any)
+ if !ok {
+ return fmt.Errorf(
+ "definitions/%s is missing, and the airflow
package generates from it",
+ name,
+ )
+ }
+ properties, ok := definition["properties"].(map[string]any)
+ if !ok {
+ return fmt.Errorf("definitions/%s has no properties to
generate a struct from", name)
+ }
+ shape := shapes[name]
+ if err := excludeProperties(name, definition, properties,
shape.exclude); err != nil {
+ return err
+ }
+ if err := overrideProperties(name, properties, shape.override);
err != nil {
+ return err
+ }
+ if err := injectProperties(name, properties, shape.inject); err
!= nil {
+ return err
+ }
+ if err := rejectCombinators(name, definition); err != nil {
+ return err
+ }
+ definition["description"] = shape.doc
+ // An authoring struct holds the fields it declares and no
others, and every
+ // field is optional: airflow.Dag takes the dag_id positionally
and a task_id
+ // defaults to the name of the Go function.
+ definition["additionalProperties"] = false
+ delete(definition, "required")
+ setPointers(properties)
+ }
+ return nil
+}
+
+func excludeProperties(
+ name string, definition, properties map[string]any, exclude
map[string]string,
+) error {
+ for _, property := range sortedKeys(exclude) {
+ if _, ok := properties[property]; !ok {
+ return fmt.Errorf(
+ "definitions/%s/properties/%s is excluded as
%q, but the schema no longer has it",
+ name, property, exclude[property],
+ )
+ }
+ delete(properties, property)
+ }
+ return nil
+}
+
+// overrideProperties replaces the Go type of a property with goJSONSchema, the
+// extension go-jsonschema reads before it reads a type or a $ref.
+func overrideProperties(
+ name string,
+ properties map[string]any,
+ override map[string]propertyOverride,
+) error {
+ for _, property := range sortedKeys(override) {
+ node, ok := properties[property].(map[string]any)
+ if !ok {
+ return fmt.Errorf(
+ "definitions/%s/properties/%s is overridden to
%s, but the schema no longer has it",
+ name, property, override[property].goType,
+ )
+ }
+ extension := map[string]any{"type": override[property].goType}
+ if imports := override[property].imports; len(imports) > 0 {
+ extension["imports"] = anySlice(imports)
+ }
+ if doc := override[property].doc; doc != "" {
+ node["description"] = doc
+ }
+ if override[property].pointer {
+ extension["pointer"] = true
+ }
+ node["goJSONSchema"] = extension
+ // The override replaces whatever the reference resolves to,
and dropping it
+ // keeps the pruned schema free of references to definitions
that are gone.
+ delete(node, "$ref")
+ }
+ return nil
+}
+
+// rejectCombinators reports an anyOf, oneOf or allOf left in a definition the
specs
+// generate from. genspec has no rule for one, and go-jsonschema does not fail
on it
+// either: it degrades the property to interface{} or gives it a typedef of
its own,
+// so the field's meaning is lost in a diff that still looks like a field. The
+// nullable pair the schema writes as anyOf [boolean, null] is the shape to
expect,
+// and resolveNullableTypes only handles the type-list spelling of it.
+//
+// A property whose type an override replaces is exempt: the override stands
for
+// whatever the schema says the property is.
+func rejectCombinators(name string, definition map[string]any) error {
+ return walkFrom("/definitions/"+name, definition, func(path string,
node map[string]any) error {
+ if extension, ok := node["goJSONSchema"].(map[string]any); ok {
+ if _, overridden := extension["type"]; overridden {
+ return nil
+ }
+ }
+ for _, keyword := range []string{"allOf", "anyOf", "oneOf"} {
+ if _, ok := node[keyword]; ok {
+ return fmt.Errorf(
+ "%s has %s, which genspec has no rule
for; exclude the property or give it "+
+ "a type override",
+ path, keyword,
+ )
+ }
+ }
+ return nil
+ })
+}
+
+// setPointers decides, for every property left, whether its field is a
pointer. A
+// field needs one wherever the Go zero value is something an author could
mean and
+// the schema does not assert that it is already the default: a concrete field
would
+// make that setting indistinguishable from an unset one, and omitempty would
drop
+// it on the way out.
+//
+// Two shapes qualify. A scalar whose schema default is not the Go zero value
is the
+// rule pkg/execution/genmodels applies to the supervisor schema. A boolean
with no
+// schema default at all is the second: false is always one of its two legal
values,
+// and the absence of a default does not mean the default is false — catchup
and
+// is_paused_upon_creation take theirs from [scheduler] catchup_by_default and
+// [core] dags_are_paused_at_creation, both of which can be true. A count
keeps its
+// concrete type where 0 is not a value it can take, as for max_active_runs,
so the
+// zero value can only mean unset; the counts where 0 does mean something say
so with
+// a pointer override, which the schema gives no way to work out.
+func setPointers(properties map[string]any) {
+ for _, name := range sortedKeys(properties) {
+ node, ok := properties[name].(map[string]any)
+ if !ok {
+ continue
+ }
+ extension, ok := node["goJSONSchema"].(map[string]any)
+ if !ok {
+ extension = map[string]any{}
+ node["goJSONSchema"] = extension
+ }
+ if _, forced := extension["pointer"]; forced {
+ continue
+ }
+ extension["pointer"] = needsPointer(node)
+ }
+}
+
+func needsPointer(node map[string]any) bool {
+ if node["type"] == "boolean" {
+ _, declared := node["default"]
+ return !declared || hasNonZeroDefault(node)
+ }
+ return hasNonZeroDefault(node)
+}
+
+// hasNonZeroDefault reports whether the property carries a scalar default
that a
+// Go zero value does not satisfy. A string default is not one: an author
leaves a
+// string empty to mean unset, and no default the schema declares is the empty
+// string.
+func hasNonZeroDefault(node map[string]any) bool {
+ value, ok := node["default"]
+ if !ok {
+ return false
+ }
+ switch value := value.(type) {
+ case bool:
+ return value
+ // A number arrives as a json.Number from readSchema, which decodes with
+ // UseNumber, and as a float64 from any plain decoder.
+ case json.Number:
+ number, err := value.Float64()
+ return err == nil && number != 0
+ case float64:
+ return value != 0
+ default:
+ return false
+ }
+}
+
+func injectProperties(
+ name string,
+ properties map[string]any,
+ inject map[string]map[string]any,
+) error {
+ for _, property := range sortedKeys(inject) {
+ if _, ok := properties[property]; ok {
+ return fmt.Errorf(
+ "definitions/%s/properties/%s is injected, but
the schema now declares it too",
+ name, property,
+ )
+ }
+ properties[property] = cloneProperty(inject[property])
+ }
+ return nil
+}
+
+// keepOnlySpecGeneratingSchema leaves the definitions in shapes and drops
+// everything else, so that the generated file holds the spec structs alone.
The
+// root schema describes a serialized Dag file rather than a spec, and every
other
+// definition is reachable only from a property the shapes exclude.
+//
+// It reports a definition that still references a dropped one, which would
+// otherwise generate against a $ref that resolves to nothing.
+func keepOnlySpecGeneratingSchema(doc map[string]any, shapes
map[string]authoringShape) error {
+ definitions, ok := doc["definitions"].(map[string]any)
+ if !ok {
+ return errNoDefinitions
+ }
+ for _, name := range sortedKeys(definitions) {
+ if _, kept := shapes[name]; !kept {
+ delete(definitions, name)
+ }
+ }
+ for _, key := range sortedKeys(doc) {
+ switch key {
+ case "$schema", "$id", "definitions":
+ default:
+ delete(doc, key)
+ }
+ }
+ return walkObjects(doc, func(path string, node map[string]any) error {
+ ref, ok := node["$ref"].(string)
+ if !ok {
+ return nil
+ }
+ target, ok := definitionName(ref)
+ if !ok {
+ return fmt.Errorf("%s references %s, which is not a
definition", path, ref)
+ }
+ if _, ok := definitions[target]; !ok {
+ return fmt.Errorf(
+ "%s references definitions/%s, which the spec
structs do not generate; "+
+ "exclude the property or give it a type
override",
+ path, target,
+ )
+ }
+ return nil
+ })
+}
+
+func definitionName(ref string) (string, bool) {
+ const prefix = "#/definitions/"
+ if len(ref) <= len(prefix) || ref[:len(prefix)] != prefix {
+ return "", false
+ }
+ return ref[len(prefix):], true
+}
+
+func cloneProperty(property map[string]any) map[string]any {
+ clone := make(map[string]any, len(property))
+ for key, value := range property {
+ clone[key] = value
+ }
+ return clone
+}
+
+func anySlice(values []string) []any {
+ out := make([]any, 0, len(values))
+ for _, value := range values {
+ out = append(out, value)
+ }
+ return out
+}
+
+var errNoDefinitions = errors.New(`the schema has no "definitions" object`)
diff --git a/go-sdk/internal/genspec/authoring_test.go
b/go-sdk/internal/genspec/authoring_test.go
new file mode 100644
index 00000000000..bb5f9403faf
--- /dev/null
+++ b/go-sdk/internal/genspec/authoring_test.go
@@ -0,0 +1,329 @@
+// 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.
+
+package main
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// shapes is one definition's worth of rules, named as the schema names it.
+func shapes(shape authoringShape) map[string]authoringShape {
+ return map[string]authoringShape{"dag": shape}
+}
+
+func TestShapeForAuthoringDropsAnExcludedPropertyAndItsRequiredEntry(t
*testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {
+ "type": "object",
+ "required": ["fileloc", "description"],
+ "properties": {
+ "fileloc": {"type": "string"},
+ "description": {"type": "string"}
+ }
+ }
+ }
+ }`)
+
+ require.NoError(t, shapeForAuthoring(doc, shapes(authoringShape{
+ exclude: map[string]string{"fileloc": "the bundle fills it in"},
+ })))
+
+ dag := doc["definitions"].(map[string]any)["dag"].(map[string]any)
+ assert.NotContains(t, dag["properties"], "fileloc")
+ assert.Contains(t, dag["properties"], "description")
+ assert.NotContains(t, dag, "required", "an authoring struct has no
required field")
+}
+
+func TestShapeForAuthoringReportsAnExclusionTheSchemaNoLongerHas(t *testing.T)
{
+ doc := schemaFrom(t, `{
+ "definitions": {"dag": {"type": "object", "properties":
{"description": {"type": "string"}}}}
+ }`)
+
+ err := shapeForAuthoring(doc, shapes(authoringShape{
+ exclude: map[string]string{"fileloc": "the bundle fills it in"},
+ }))
+
+ assert.EqualError(
+ t,
+ err,
+ `definitions/dag/properties/fileloc is excluded as "the bundle
fills it in", `+
+ "but the schema no longer has it",
+ )
+}
+
+func TestShapeForAuthoringOverridesTheGoTypeOfAReferencedProperty(t
*testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {"type": "object", "properties": {"start_date":
{"$ref": "#/definitions/datetime"}}},
+ "datetime": {"type": "number"}
+ }
+ }`)
+
+ require.NoError(t, shapeForAuthoring(doc, shapes(authoringShape{
+ override: map[string]propertyOverride{
+ "start_date": {goType: "time.Time", imports:
[]string{"time"}},
+ },
+ })))
+
+ properties :=
doc["definitions"].(map[string]any)["dag"].(map[string]any)["properties"].(map[string]any)
+ startDate := properties["start_date"].(map[string]any)
+ assert.NotContains(t, startDate, "$ref", "the reference would outlive
the definition it names")
+ assert.Equal(t, "time.Time",
startDate["goJSONSchema"].(map[string]any)["type"])
+ assert.Equal(t, []any{"time"},
startDate["goJSONSchema"].(map[string]any)["imports"])
+}
+
+func TestShapeForAuthoringOverridesTheDocCommentOfAProperty(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {"dag": {"type": "object", "properties":
{"dag_id": {"type": "string"}}}}
+ }`)
+
+ require.NoError(t, shapeForAuthoring(doc, shapes(authoringShape{
+ override: map[string]propertyOverride{
+ "dag_id": {goType: "string", doc: "DagID is what the
constructor takes."},
+ },
+ })))
+
+ properties :=
doc["definitions"].(map[string]any)["dag"].(map[string]any)["properties"].(map[string]any)
+ assert.Equal(
+ t,
+ "DagID is what the constructor takes.",
+ properties["dag_id"].(map[string]any)["description"],
+ )
+}
+
+func TestShapeForAuthoringLetsAnOverrideForceAPointer(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {
+ "type": "object",
+ "properties": {
+ "max_consecutive_failed_dag_runs":
{"type": "number"},
+ "max_active_runs": {"type": "number"}
+ }
+ }
+ }
+ }`)
+
+ require.NoError(t, shapeForAuthoring(doc, shapes(authoringShape{
+ override: map[string]propertyOverride{
+ "max_consecutive_failed_dag_runs": {goType: "int",
pointer: true},
+ "max_active_runs": {goType: "int"},
+ },
+ })))
+
+ properties :=
doc["definitions"].(map[string]any)["dag"].(map[string]any)["properties"].(map[string]any)
+ forced := properties["max_consecutive_failed_dag_runs"].(map[string]any)
+ assert.Equal(t, true,
forced["goJSONSchema"].(map[string]any)["pointer"],
+ "0 is a setting for this count, which the schema gives no way
to work out")
+ ruled := properties["max_active_runs"].(map[string]any)
+ assert.Equal(t, false,
ruled["goJSONSchema"].(map[string]any)["pointer"])
+}
+
+func TestShapeForAuthoringReportsAnOverrideTheSchemaNoLongerHas(t *testing.T) {
+ doc := schemaFrom(t, `{"definitions": {"dag": {"type": "object",
"properties": {}}}}`)
+
+ err := shapeForAuthoring(doc, shapes(authoringShape{
+ override: map[string]propertyOverride{"start_date": {goType:
"time.Time"}},
+ }))
+
+ assert.EqualError(
+ t,
+ err,
+ "definitions/dag/properties/start_date is overridden to
time.Time, "+
+ "but the schema no longer has it",
+ )
+}
+
+func TestShapeForAuthoringInjectsAPropertyTheSchemaHasNoCounterpartFor(t
*testing.T) {
+ doc := schemaFrom(t, `{"definitions": {"dag": {"type": "object",
"properties": {}}}}`)
+
+ require.NoError(t, shapeForAuthoring(doc, shapes(authoringShape{
+ inject: map[string]map[string]any{"schedule": {"type":
"string"}},
+ })))
+
+ properties :=
doc["definitions"].(map[string]any)["dag"].(map[string]any)["properties"].(map[string]any)
+ assert.Equal(t, "string",
properties["schedule"].(map[string]any)["type"])
+}
+
+func TestShapeForAuthoringReportsAnInjectionTheSchemaNowDeclares(t *testing.T)
{
+ doc := schemaFrom(t, `{
+ "definitions": {"dag": {"type": "object", "properties":
{"schedule": {"type": "object"}}}}
+ }`)
+
+ err := shapeForAuthoring(doc, shapes(authoringShape{
+ inject: map[string]map[string]any{"schedule": {"type":
"string"}},
+ }))
+
+ assert.EqualError(
+ t,
+ err,
+ "definitions/dag/properties/schedule is injected, but the
schema now declares it too",
+ )
+}
+
+func TestShapeForAuthoringReportsAReferenceToAPrunedDefinition(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {"type": "object", "properties": {"params":
{"$ref": "#/definitions/params"}}},
+ "params": {"type": "object"}
+ }
+ }`)
+
+ err := shapeForAuthoring(doc, shapes(authoringShape{}))
+
+ assert.EqualError(
+ t,
+ err,
+ "/definitions/dag/properties/params references
definitions/params, which the spec structs "+
+ "do not generate; exclude the property or give it a
type override",
+ )
+}
+
+func TestShapeForAuthoringKeepsOnlyTheDefinitionsTheSpecsGenerateFrom(t
*testing.T) {
+ doc := schemaFrom(t, `{
+ "$schema": "http://json-schema.org/draft-07/schema#";,
+ "type": "object",
+ "allOf": [{"properties": {"dag": {"$ref":
"#/definitions/dag"}}}],
+ "definitions": {
+ "dag": {"type": "object", "properties": {}},
+ "asset": {"type": "object"}
+ }
+ }`)
+
+ require.NoError(t, shapeForAuthoring(doc, shapes(authoringShape{})))
+
+ assert.Equal(t, []string{"$schema", "definitions"}, sortedKeys(doc),
+ "the root describes a serialized Dag file, which is not a spec")
+ assert.Equal(t, []string{"dag"},
sortedKeys(doc["definitions"].(map[string]any)))
+}
+
+func TestShapeForAuthoringRejectsACombinatorItHasNoRuleFor(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {
+ "type": "object",
+ "properties": {
+ "render_template_as_native_obj": {
+ "anyOf": [{"type": "boolean"},
{"type": "null"}]
+ }
+ }
+ }
+ }
+ }`)
+
+ err := shapeForAuthoring(doc, shapes(authoringShape{}))
+
+ assert.EqualError(
+ t,
+ err,
+ "/definitions/dag/properties/render_template_as_native_obj has
anyOf, which genspec "+
+ "has no rule for; exclude the property or give it a
type override",
+ )
+}
+
+func TestShapeForAuthoringKeepsACombinatorAnOverrideReplaces(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {
+ "type": "object",
+ "properties": {"deadline": {"anyOf": [{"type":
"number"}, {"type": "null"}]}}
+ }
+ }
+ }`)
+
+ assert.NoError(t, shapeForAuthoring(doc, shapes(authoringShape{
+ override: map[string]propertyOverride{
+ "deadline": {goType: "time.Duration", imports:
[]string{"time"}},
+ },
+ })))
+}
+
+func TestShapeForAuthoringPointsOnlyAtAScalarWhoseDefaultIsNotTheGoZeroValue(t
*testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {
+ "type": "object",
+ "properties": {
+ "do_xcom_push": {"type": "boolean",
"default": true},
+ "fail_fast": {"type": "boolean",
"default": false},
+ "catchup": {"type": "boolean"},
+ "retries": {"type": "number",
"default": 0},
+ "pool_slots": {"type": "number",
"default": 1},
+ "max_active_runs": {"type": "number"},
+ "owner": {"type": "string", "default":
"airflow"},
+ "tags": {"type": "array"}
+ }
+ }
+ }
+ }`)
+
+ require.NoError(t, shapeForAuthoring(doc, shapes(authoringShape{})))
+
+ properties :=
doc["definitions"].(map[string]any)["dag"].(map[string]any)["properties"].(map[string]any)
+ for name, pointer := range map[string]bool{
+ "do_xcom_push": true,
+ "pool_slots": true,
+ // A boolean with no schema default takes its default from
Airflow config, which
+ // can be true, so false has to be expressible.
+ "catchup": true,
+ "fail_fast": false,
+ "retries": false,
+ // 0 is not a value max_active_runs can take, so its zero value
can only mean unset.
+ "max_active_runs": false,
+ "owner": false,
+ "tags": false,
+ } {
+ node := properties[name].(map[string]any)
+ assert.Equal(t, pointer,
node["goJSONSchema"].(map[string]any)["pointer"], name)
+ }
+}
+
+// TestShapeForAuthoringShapesTheCoreSchema is the tripwire for a property
added,
+// renamed or retyped on the Python side that the shape tables have no rule
for.
+func TestShapeForAuthoringShapesTheCoreSchema(t *testing.T) {
+ doc := readCoreSchema(t)
+
+ require.NoError(t, shapeForAuthoring(doc, authoringShapes))
+ require.NoError(t, normalize(doc, specTitles))
+
+ definitions := doc["definitions"].(map[string]any)
+ assert.Equal(t, []string{"dag", "operator"}, sortedKeys(definitions))
+ dag :=
definitions["dag"].(map[string]any)["properties"].(map[string]any)
+ assert.Contains(t, dag, "schedule", "Schedule is injected over the
serialized timetable")
+ task :=
definitions["operator"].(map[string]any)["properties"].(map[string]any)
+ assert.Equal(
+ t,
+ "TriggerRule",
+
task["trigger_rule"].(map[string]any)["goJSONSchema"].(map[string]any)["type"],
+ )
+ assert.Equal(
+ t,
+ "WeightRule",
+
task["weight_rule"].(map[string]any)["goJSONSchema"].(map[string]any)["type"],
+ )
+ assert.Contains(
+ t,
+ task["task_id"].(map[string]any)["description"],
+ "the name of the Go function",
+ "the serialized property says nothing about how a task_id is
defaulted",
+ )
+}
diff --git a/go-sdk/internal/genspec/main.go b/go-sdk/internal/genspec/main.go
new file mode 100644
index 00000000000..87f93b67a26
--- /dev/null
+++ b/go-sdk/internal/genspec/main.go
@@ -0,0 +1,110 @@
+// 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.
+
+// 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.
+//
+// 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
+// serialized shape into the authoring shape, and normalize makes what is left
+// something go-jsonschema can read. Each change is documented on the function
that
+// makes it.
+//
+// With -license it inserts the Apache header into an already generated file
+// instead, which is what go-jsonschema leaves out.
+package main
+
+import (
+ "bytes"
+ "encoding/json"
+ "flag"
+ "log"
+ "os"
+ "path/filepath"
+
+ "github.com/apache/airflow/go-sdk/internal/genlicense"
+)
+
+func main() {
+ schemaPath := flag.String(
+ "schema",
+ "",
+ "path to airflow-core's Dag serialization schema.json",
+ )
+ outPath := flag.String("out", "", "path to write the rewritten schema
to")
+ licensePath := flag.String(
+ "license",
+ "",
+ "path to a generated Go file to insert the Apache license
header into, instead of rewriting a schema",
+ )
+ flag.Parse()
+
+ if *licensePath != "" {
+ if *schemaPath != "" || *outPath != "" {
+ log.Fatal("genspec: -license rewrites no schema, so it
takes neither -schema nor -out")
+ }
+ if err := genlicense.EnsureHeader(*licensePath); err != nil {
+ log.Fatalf("genspec: adding the license header to %s:
%v", *licensePath, err)
+ }
+ return
+ }
+ if *schemaPath == "" {
+ log.Fatal("genspec: -schema is required")
+ }
+ if *outPath == "" {
+ log.Fatal("genspec: -out is required")
+ }
+
+ doc, err := readSchema(*schemaPath)
+ if err != nil {
+ log.Fatalf("genspec: reading %s: %v", *schemaPath, err)
+ }
+ if err := shapeForAuthoring(doc, authoringShapes); err != nil {
+ log.Fatalf("genspec: shaping %s for authoring: %v",
*schemaPath, err)
+ }
+ if err := normalize(doc, specTitles); err != nil {
+ log.Fatalf("genspec: normalizing %s: %v", *schemaPath, err)
+ }
+ out, err := json.MarshalIndent(doc, "", " ")
+ if err != nil {
+ log.Fatalf("genspec: encoding the normalized schema: %v", err)
+ }
+ if err := os.MkdirAll(filepath.Dir(*outPath), 0o755); err != nil {
+ log.Fatalf("genspec: creating the directory of %s: %v",
*outPath, err)
+ }
+ if err := os.WriteFile(*outPath, append(out, '\n'), 0o644); err != nil {
+ log.Fatalf("genspec: writing %s: %v", *outPath, err)
+ }
+}
+
+func readSchema(path string) (map[string]any, error) {
+ raw, err := os.ReadFile(path)
+ if err != nil {
+ return nil, err
+ }
+ var doc map[string]any
+ dec := json.NewDecoder(bytes.NewReader(raw))
+ // A schema default such as retry_delay's 300.0 would otherwise decode
to
+ // float64 and re-encode as 300, and go-jsonschema reads a default's
JSON type.
+ dec.UseNumber()
+ if err := dec.Decode(&doc); err != nil {
+ return nil, err
+ }
+ return doc, nil
+}
diff --git a/go-sdk/airflow/spec.go b/go-sdk/internal/genspec/main_test.go
similarity index 57%
copy from go-sdk/airflow/spec.go
copy to go-sdk/internal/genspec/main_test.go
index 20621ded56f..5f77bd5c7c6 100644
--- a/go-sdk/airflow/spec.go
+++ b/go-sdk/internal/genspec/main_test.go
@@ -15,14 +15,23 @@
// specific language governing permissions and limitations
// under the License.
-package airflow
+package main
-// DagSpec holds the attributes of a Dag other than its dag_id. [Dag] takes
one.
-type DagSpec struct{}
+import (
+ "encoding/json"
+ "path/filepath"
+ "testing"
-// TaskSpec holds the attributes of a task. [DagRef.Task] takes at most one
per task.
-type TaskSpec struct {
- // TaskID is the task_id of the task. When TaskID is empty, the task_id
is the name of the
- // Go function that the task runs.
- TaskID string
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestReadSchemaKeepsAFloatDefaultAFloat(t *testing.T) {
+ doc, err := readSchema(filepath.FromSlash(coreSchemaPath))
+ require.NoError(t, err)
+
+ operator :=
doc["definitions"].(map[string]any)["operator"].(map[string]any)
+ retryDelay :=
operator["properties"].(map[string]any)["retry_delay"].(map[string]any)
+ assert.Equal(t, json.Number("300.0"), retryDelay["default"],
+ "go-jsonschema reads the JSON type of a default, so 300.0 must
not become 300")
}
diff --git a/go-sdk/internal/genspec/normalize.go
b/go-sdk/internal/genspec/normalize.go
new file mode 100644
index 00000000000..307f12ba47c
--- /dev/null
+++ b/go-sdk/internal/genspec/normalize.go
@@ -0,0 +1,176 @@
+// 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.
+
+package main
+
+import (
+ "fmt"
+ "slices"
+)
+
+var specTitles = map[string]string{
+ "dag": "DagSpec",
+ "operator": "TaskSpec",
+}
+
+// normalize rewrites doc in place so that go-jsonschema can read it, and
injects
+// the title of each definition in titles. It reports the first construct it
cannot
+// rewrite, naming the path to it, so that a schema change on the Python side
that
+// needs a new rule here fails the generate step with somewhere to look.
+func normalize(doc map[string]any, titles map[string]string) error {
+ if err := dropDependentRequired(doc); err != nil {
+ return err
+ }
+ if err := resolveNullableTypes(doc); err != nil {
+ return err
+ }
+ return injectTitles(doc, titles)
+}
+
+// dropDependentRequired removes every dependencies entry written in draft-07's
+// array form, the one later drafts renamed dependentRequired. It is the single
+// construct in the schema that go-jsonschema v0.23.1 cannot parse at all: it
+// fails the whole file with "cannot unmarshal array into Go value of type
+// schemas.ObjectAsType". The entry states that one property requires others
+// alongside it, which constrains an instance and not the generated type. An
entry
+// whose value is a schema is left alone: go-jsonschema reads that form.
+func dropDependentRequired(doc map[string]any) error {
+ return walkObjects(doc, func(_ string, node map[string]any) error {
+ deps, ok := node["dependencies"].(map[string]any)
+ if !ok {
+ return nil
+ }
+ for _, name := range sortedKeys(deps) {
+ if _, isList := deps[name].([]any); isList {
+ delete(deps, name)
+ }
+ }
+ if len(deps) == 0 {
+ delete(node, "dependencies")
+ }
+ return nil
+ })
+}
+
+// resolveNullableTypes rewrites every nullable type pair such as
+// ["boolean", "null"] to the one type it allows, wherever it appears. Whether
the
+// field is a pointer is not decided here: setPointers writes that on every
+// property, so the type pair only has to leave a type go-jsonschema names
well.
+//
+// go-jsonschema reads a type list, but it gives the field a typedef of its own
+// (type DagRerunWithLatestVersion *bool, type DagTagsElem *string) instead of
the
+// plain *bool or *string a single nullable type produces, and that typedef
would
+// be an exported name in the airflow package standing for nothing an author
names.
+func resolveNullableTypes(doc map[string]any) error {
+ return walkObjects(doc, resolveNodeType)
+}
+
+// resolveNodeType rewrites the type list on node itself, so that a pair
nested in
+// items or additionalProperties resolves by the same rule as one on a
property.
+func resolveNodeType(path string, node map[string]any) error {
+ types, ok := node["type"].([]any)
+ if !ok {
+ return nil
+ }
+ kept := slices.DeleteFunc(slices.Clone(types), func(t any) bool {
+ return t == "null"
+ })
+ switch len(kept) {
+ case 1:
+ node["type"] = kept[0]
+ return nil
+ case 0:
+ return fmt.Errorf("%s has type %v, which allows nothing to
generate from", path, types)
+ default:
+ return fmt.Errorf(
+ "%s has type %v: go-jsonschema degrades a type union to
interface{}, "+
+ "which is not a field an author can set",
+ path, types,
+ )
+ }
+}
+
+// injectTitles gives each definition in titles the title that
+// --struct-name-from-title reads. It reports a definition that has gone
missing
+// or already carries a title of its own, either of which means titles is
stale.
+//
+// Neither definition carries a title, so the flag has nothing to read and
+// go-jsonschema falls back to capitalizing the definition keys dag and
operator.
+// Dag is already the name of the constructor and Operator is not the SDK's
+// vocabulary, which is why the titles are injected rather than left to the
tool.
+func injectTitles(doc map[string]any, titles map[string]string) error {
+ definitions, ok := doc["definitions"].(map[string]any)
+ if !ok {
+ return errNoDefinitions
+ }
+ for _, name := range sortedKeys(titles) {
+ definition, ok := definitions[name].(map[string]any)
+ if !ok {
+ return fmt.Errorf(
+ "definitions/%s is missing, and %s generates
from it", name, titles[name],
+ )
+ }
+ switch title := definition["title"]; title {
+ case nil, titles[name]:
+ definition["title"] = titles[name]
+ default:
+ return fmt.Errorf(
+ "definitions/%s already has the title %q, so
generating %s from it "+
+ "would rename the type",
+ name, title, titles[name],
+ )
+ }
+ }
+ return nil
+}
+
+// walkObjects calls visit on doc and on every object below it, in key order so
+// that the error normalize reports for a schema with several offending
+// constructs is the same on every run.
+func walkObjects(doc map[string]any, visit func(path string, node
map[string]any) error) error {
+ return walkFrom("", doc, visit)
+}
+
+func walkFrom(path string, node any, visit func(path string, node
map[string]any) error) error {
+ switch node := node.(type) {
+ case map[string]any:
+ if err := visit(path, node); err != nil {
+ return err
+ }
+ for _, key := range sortedKeys(node) {
+ if err := walkFrom(path+"/"+key, node[key], visit); err
!= nil {
+ return err
+ }
+ }
+ case []any:
+ for i, item := range node {
+ if err := walkFrom(fmt.Sprintf("%s/%d", path, i), item,
visit); err != nil {
+ return err
+ }
+ }
+ }
+ return nil
+}
+
+func sortedKeys[V any](m map[string]V) []string {
+ keys := make([]string, 0, len(m))
+ for key := range m {
+ keys = append(keys, key)
+ }
+ slices.Sort(keys)
+ return keys
+}
diff --git a/go-sdk/internal/genspec/normalize_test.go
b/go-sdk/internal/genspec/normalize_test.go
new file mode 100644
index 00000000000..b10bcb8fa18
--- /dev/null
+++ b/go-sdk/internal/genspec/normalize_test.go
@@ -0,0 +1,256 @@
+// 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.
+
+package main
+
+import (
+ "encoding/json"
+ "path/filepath"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// coreSchemaPath is the schema genspec normalizes, reached from this package.
+const coreSchemaPath =
"../../../airflow-core/src/airflow/serialization/schema.json"
+
+func schemaFrom(t *testing.T, body string) map[string]any {
+ t.Helper()
+
+ var doc map[string]any
+ require.NoError(t, json.Unmarshal([]byte(body), &doc))
+ return doc
+}
+
+func readCoreSchema(t *testing.T) map[string]any {
+ t.Helper()
+
+ doc, err := readSchema(filepath.FromSlash(coreSchemaPath))
+ require.NoError(t, err)
+ return doc
+}
+
+func TestNormalizeResolvesNullableType(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {
+ "type": "object",
+ "properties": {"rerun_with_latest_version":
{"type": ["boolean", "null"]}}
+ },
+ "operator": {"type": "object"}
+ }
+ }`)
+
+ require.NoError(t, normalize(doc, specTitles))
+
+ dag := doc["definitions"].(map[string]any)["dag"].(map[string]any)
+ properties := dag["properties"].(map[string]any)
+ assert.Equal(
+ t,
+ "boolean",
+
properties["rerun_with_latest_version"].(map[string]any)["type"],
+ )
+}
+
+func TestNormalizeKeepsASingleTypeAsItIs(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {"type": "object", "properties": {"dag_id":
{"type": "string"}}},
+ "operator": {"type": "object"}
+ }
+ }`)
+
+ require.NoError(t, normalize(doc, specTitles))
+
+ dag := doc["definitions"].(map[string]any)["dag"].(map[string]any)
+ properties := dag["properties"].(map[string]any)
+ assert.Equal(t, "string", properties["dag_id"].(map[string]any)["type"])
+}
+
+func TestNormalizeDropsDependentRequired(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {"type": "object"},
+ "operator": {
+ "type": "object",
+ "dependencies": {
+ "expand_input": ["partial_kwargs",
"_is_mapped"],
+ "partial_kwargs": ["expand_input",
"_is_mapped"]
+ }
+ }
+ }
+ }`)
+
+ require.NoError(t, normalize(doc, specTitles))
+
+ operator :=
doc["definitions"].(map[string]any)["operator"].(map[string]any)
+ assert.NotContains(t, operator, "dependencies")
+}
+
+func TestNormalizeKeepsSchemaFormDependencies(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {"type": "object"},
+ "operator": {
+ "type": "object",
+ "dependencies": {
+ "expand_input": ["partial_kwargs"],
+ "pool": {"properties": {"pool_slots":
{"type": "number"}}}
+ }
+ }
+ }
+ }`)
+
+ require.NoError(t, normalize(doc, specTitles))
+
+ operator :=
doc["definitions"].(map[string]any)["operator"].(map[string]any)
+ dependencies := operator["dependencies"].(map[string]any)
+ assert.NotContains(t, dependencies, "expand_input")
+ assert.Contains(t, dependencies, "pool",
+ "go-jsonschema reads a dependency whose value is a schema")
+}
+
+func TestNormalizeInjectsTheSpecTitles(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {"dag": {"type": "object"}, "operator": {"type":
"object"}}
+ }`)
+
+ require.NoError(t, normalize(doc, specTitles))
+
+ definitions := doc["definitions"].(map[string]any)
+ assert.Equal(t, "DagSpec", definitions["dag"].(map[string]any)["title"])
+ assert.Equal(t, "TaskSpec",
definitions["operator"].(map[string]any)["title"])
+}
+
+func TestNormalizeRejects(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ schema string
+ wantErr string
+ }{
+ {
+ name: "a type union it cannot resolve",
+ schema: `{
+ "definitions": {
+ "dag": {
+ "type": "object",
+ "properties": {"either":
{"type": ["boolean", "string"]}}
+ },
+ "operator": {"type": "object"}
+ }
+ }`,
+ wantErr: "/definitions/dag/properties/either",
+ },
+ {
+ name: "a type that allows only null",
+ schema: `{
+ "definitions": {
+ "dag": {"type": "object", "properties":
{"nothing": {"type": ["null"]}}},
+ "operator": {"type": "object"}
+ }
+ }`,
+ wantErr: "allows nothing to generate from",
+ },
+ {
+ name: "a missing definition",
+ schema: `{"definitions": {"dag": {"type": "object"}}}`,
+ wantErr: "definitions/operator is missing",
+ },
+ {
+ name: "a definition that already has a title",
+ schema: `{
+ "definitions": {
+ "dag": {"type": "object", "title":
"SerializedDag"},
+ "operator": {"type": "object"}
+ }
+ }`,
+ wantErr: `already has the title "SerializedDag"`,
+ },
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ err := normalize(schemaFrom(t, tc.schema), specTitles)
+
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), tc.wantErr)
+ })
+ }
+}
+
+func TestNormalizeAcceptsTheTitleItWouldInject(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {"type": "object", "title": "DagSpec"},
+ "operator": {"type": "object"}
+ }
+ }`)
+
+ assert.NoError(t, normalize(doc, specTitles))
+}
+
+func TestNormalizeResolvesANullableTypeNestedInItems(t *testing.T) {
+ doc := schemaFrom(t, `{
+ "definitions": {
+ "dag": {
+ "type": "object",
+ "properties": {"tags": {"type": "array",
"items": {"type": ["string", "null"]}}}
+ },
+ "operator": {"type": "object"}
+ }
+ }`)
+
+ require.NoError(t, normalize(doc, specTitles))
+
+ dag := doc["definitions"].(map[string]any)["dag"].(map[string]any)
+ tags := dag["properties"].(map[string]any)["tags"].(map[string]any)
+ assert.Equal(t, "string", tags["items"].(map[string]any)["type"],
+ "a pair nested in items resolves by the same rule as one on a
property")
+}
+
+func TestNormalizeReportsTheSameConstructOnEveryRun(t *testing.T) {
+ body := `{
+ "definitions": {
+ "dag": {
+ "type": "object",
+ "properties": {
+ "a": {"type": ["boolean", "string"]},
+ "b": {"type": ["number", "string"]}
+ }
+ },
+ "operator": {"type": "object"}
+ }
+ }`
+
+ first := normalize(schemaFrom(t, body), specTitles)
+ require.Error(t, first)
+ for range 20 {
+ assert.EqualError(t, normalize(schemaFrom(t, body),
specTitles), first.Error())
+ }
+}
+
+// TestNormalizeReadsTheCoreSchema is the tripwire for a schema change on the
+// Python side that genspec has no rule for yet.
+func TestNormalizeReadsTheCoreSchema(t *testing.T) {
+ doc, err := readSchema(filepath.FromSlash(coreSchemaPath))
+ require.NoError(t, err)
+
+ require.NoError(t, normalize(doc, specTitles))
+
+ definitions := doc["definitions"].(map[string]any)
+ assert.Equal(t, "DagSpec", definitions["dag"].(map[string]any)["title"])
+ assert.Equal(t, "TaskSpec",
definitions["operator"].(map[string]any)["title"])
+}
diff --git a/go-sdk/pkg/execution/genmodels/gen/main.go
b/go-sdk/pkg/execution/genmodels/gen/main.go
index d2ae96adcaf..7a6f3bbc6e0 100644
--- a/go-sdk/pkg/execution/genmodels/gen/main.go
+++ b/go-sdk/pkg/execution/genmodels/gen/main.go
@@ -63,6 +63,8 @@ import (
"strconv"
"strings"
"text/template"
+
+ "github.com/apache/airflow/go-sdk/internal/genlicense"
)
var deadTypeName = regexp.MustCompile(`_[0-9]+$`)
@@ -106,7 +108,7 @@ func main() {
if err := pointerizeFields(*modelsPath, pointerized); err != nil {
log.Fatalf("gen: pointerizing %s: %v", *modelsPath, err)
}
- if err := ensureLicenseHeader(*modelsPath); err != nil {
+ if err := genlicense.EnsureHeader(*modelsPath); err != nil {
log.Fatalf("gen: adding license header to %s: %v", *modelsPath,
err)
}
if err := writeDiscriminators(doc, *outPath, *pkg, structByKey); err !=
nil {
@@ -339,51 +341,6 @@ func pointerizeFields(path string, set
map[string]map[string]bool) error {
return os.WriteFile(path, buf.Bytes(), 0o644)
}
-// asfLicenseHeader is the Apache source header as Go line comments.
go-jsonschema
-// emits none, so the gen tool adds it (the template-generated files carry it
-// inline) to keep models.gen.go self-contained and reproducible from go
generate.
-const asfLicenseHeader = `// 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.`
-
-// ensureLicenseHeader inserts the Apache header into models.gen.go right after
-// go-jsonschema's "Code generated" line when absent, matching the layout of
the
-// other generated Go files. It is idempotent so re-running gen is a no-op.
-func ensureLicenseHeader(path string) error {
- src, err := os.ReadFile(path)
- if err != nil {
- return err
- }
- if bytes.Contains(src, []byte("Licensed to the Apache Software
Foundation")) {
- return nil
- }
- lines := strings.Split(string(src), "\n")
- insertAt := 0
- for i, l := range lines {
- if strings.HasPrefix(l, "// Code generated") {
- insertAt = i + 1
- break
- }
- }
- out := append([]string{}, lines[:insertAt]...)
- out = append(out, strings.Split(asfLicenseHeader, "\n")...)
- out = append(out, lines[insertAt:]...)
- return os.WriteFile(path, []byte(strings.Join(out, "\n")), 0o644)
-}
-
// identUseCounts counts identifier occurrences across the file in one pass. A
// dead typedef's name occurs once (its declaration); used names occur more.
func identUseCounts(file *ast.File) map[string]int {
diff --git a/scripts/ci/prek/check_go_sdk_spec_drift.py
b/scripts/ci/prek/check_go_sdk_spec_drift.py
new file mode 100755
index 00000000000..9e5dfd0a310
--- /dev/null
+++ b/scripts/ci/prek/check_go_sdk_spec_drift.py
@@ -0,0 +1,140 @@
+#!/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.
+"""
+Keep the Go SDK's Dag and task spec structs in step with the serialization
schema.
+
+``go-sdk/airflow/spec.gen.go`` declares ``airflow.DagSpec`` and
+``airflow.TaskSpec``, the two structs a Dag author fills in. Both are generated
+from ``airflow-core/src/airflow/serialization/schema.json``, which Python owns:
+``go-sdk/internal/genspec`` rewrites that schema into the authoring shape and
+go-jsonschema writes the structs from it.
+
+The generated file is committed, so nothing regenerates it when the schema
moves
+on the Python side. Without this check a property added, renamed or retyped
there
+would leave the Go structs silently behind, and a Dag authored in Go would keep
+serializing the old shape.
+
+The check regenerates the file and asks Git whether it changed. A drifted file
is
+left regenerated in the working tree, so the fix is to commit it.
+
+Run from the repo root:
+
+ uv run --project scripts python scripts/ci/prek/check_go_sdk_spec_drift.py
+
+Exits 0 if the committed structs match the schema, 1 otherwise.
+"""
+
+from __future__ import annotations
+
+import os
+import pathlib
+import shutil
+import subprocess
+import sys
+
+REPO_ROOT = pathlib.Path(__file__).resolve().parents[3]
+GO_SDK_MODULE = pathlib.Path("go-sdk")
+GENERATED_SPECS = GO_SDK_MODULE / "airflow" / "spec.gen.go"
+
+
+def regenerate_specs(module_dir: pathlib.Path, go_binary: str = "go") ->
tuple[int, str]:
+ """Run the spec generators. Returns ``(returncode, combined_output)``."""
+ completed = subprocess.run(
+ [go_binary, "generate", "./airflow/..."],
+ cwd=module_dir,
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+ return completed.returncode, (completed.stdout + completed.stderr).strip()
+
+
+def read_drift(repo_root: pathlib.Path, git_binary: str = "git") -> tuple[int,
str]:
+ """Ask Git what regeneration changed. Returns ``(returncode, diff)``."""
+ completed = subprocess.run(
+ [git_binary, "diff", "--", str(GENERATED_SPECS)],
+ cwd=repo_root,
+ capture_output=True,
+ text=True,
+ check=False,
+ )
+ return completed.returncode, completed.stdout.strip()
+
+
+def format_report(
+ generate_returncode: int, generate_output: str, diff_returncode: int,
diff: str
+) -> tuple[int, str]:
+ """Turn a regeneration result and the diff that followed it into
``(exit_code, report)``."""
+ if generate_returncode != 0:
+ return 1, "\n".join(
+ [
+ f"ERROR: regenerating {GENERATED_SPECS} failed.",
+ "",
+ "genspec 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 need the Go",
+ "toolchain and the network to fetch go-jsonschema. `go
generate` reported:",
+ "",
+ generate_output or "(no output)",
+ ]
+ )
+ # An unreadable diff is not an absent one: reporting success here would
let the
+ # check pass on every drift.
+ if diff_returncode != 0:
+ return 1, f"ERROR: `git diff` failed, so whether {GENERATED_SPECS}
drifted is unknown."
+ if not diff:
+ return 0, f"OK: {GENERATED_SPECS} matches the serialization schema."
+ return 1, "\n".join(
+ [
+ f"ERROR: {GENERATED_SPECS} is out of date.",
+ "",
+ "It is generated from
airflow-core/src/airflow/serialization/schema.json by",
+ "go-sdk/internal/genspec, and one of the two has moved since the
file was committed.",
+ "The regenerated file is in your working tree.",
+ "",
+ "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:",
+ "",
+ f" git add {GENERATED_SPECS}",
+ "",
+ "Regeneration changed:",
+ "",
+ diff,
+ ]
+ )
+
+
+def main() -> int:
+ module_dir = REPO_ROOT / GO_SDK_MODULE
+ if not (REPO_ROOT / GENERATED_SPECS).is_file():
+ print(f"ERROR: {GENERATED_SPECS} not found — has the generated file
moved?")
+ return 1
+ if shutil.which("go") is None:
+ if os.environ.get("CI"):
+ print("ERROR: `go` is not on PATH but this is a CI run — the
toolchain is required here.")
+ return 1
+ print(f"SKIPPED: `go` is not on PATH, cannot verify that
{GENERATED_SPECS} is current.")
+ return 0
+ generate_returncode, generate_output = regenerate_specs(module_dir)
+ diff_returncode, diff = read_drift(REPO_ROOT)
+ exit_code, report = format_report(generate_returncode, generate_output,
diff_returncode, diff)
+ print(report)
+ return exit_code
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/scripts/tests/ci/prek/test_check_go_sdk_spec_drift.py
b/scripts/tests/ci/prek/test_check_go_sdk_spec_drift.py
new file mode 100644
index 00000000000..fffbd80c399
--- /dev/null
+++ b/scripts/tests/ci/prek/test_check_go_sdk_spec_drift.py
@@ -0,0 +1,133 @@
+#!/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.
+from __future__ import annotations
+
+import subprocess
+from unittest import mock
+
+import check_go_sdk_spec_drift as checker
+import pytest
+
+SPEC_DRIFT_DIFF = """\
+diff --git a/go-sdk/airflow/spec.gen.go b/go-sdk/airflow/spec.gen.go
+--- a/go-sdk/airflow/spec.gen.go
++++ b/go-sdk/airflow/spec.gen.go
+@@ -40,6 +40,9 @@ type DagSpec struct {
++ // Deadline corresponds to the JSON schema field "deadline".
++ Deadline string `json:"deadline,omitempty,omitzero"`
+"""
+
+
+def test_current_specs_pass():
+ exit_code, report = checker.format_report(0, "", 0, "")
+
+ assert exit_code == 0
+ assert "matches the serialization schema" in report
+
+
+def
test_drifted_specs_fail_with_the_diff_and_where_to_decide_about_a_property():
+ exit_code, report = checker.format_report(0, "", 0, SPEC_DRIFT_DIFF)
+
+ assert exit_code == 1
+ assert "is out of date" in report
+ assert "go-sdk/internal/genspec/authoring.go" in report
+ assert "git add go-sdk/airflow/spec.gen.go" in report
+ assert 'Deadline string `json:"deadline,omitempty,omitzero"`' in report
+
+
+def test_failed_generation_reports_the_generator_output_instead_of_a_diff():
+ exit_code, report = checker.format_report(
+ 1,
+ "genspec: shaping schema.json for authoring:
definitions/dag/properties/fileloc is excluded",
+ 0,
+ "",
+ )
+
+ assert exit_code == 1
+ assert "failed" in report
+ assert "definitions/dag/properties/fileloc is excluded" in report
+
+
+def test_failed_generation_without_output_still_reports():
+ exit_code, report = checker.format_report(1, "", 0, "")
+
+ assert exit_code == 1
+ assert "(no output)" in report
+
+
+def test_unreadable_diff_fails_instead_of_passing_as_no_drift():
+ exit_code, report = checker.format_report(0, "", 128, "")
+
+ assert exit_code == 1
+ assert "is unknown" in report
+
+
[email protected]("check_go_sdk_spec_drift.subprocess.run", autospec=True)
+def test_regeneration_runs_the_generators_in_the_go_sdk_module(mock_run,
tmp_path):
+ mock_run.return_value = subprocess.CompletedProcess(args=[], returncode=0,
stdout="", stderr="")
+
+ checker.regenerate_specs(tmp_path)
+
+ assert mock_run.call_args.args[0] == ["go", "generate", "./airflow/..."]
+ assert mock_run.call_args.kwargs["cwd"] == tmp_path
+
+
[email protected]("check_go_sdk_spec_drift.subprocess.run", autospec=True)
+def test_regeneration_combines_stdout_and_stderr(mock_run, tmp_path):
+ mock_run.return_value = subprocess.CompletedProcess(
+ args=[], returncode=1, stdout="genspec: shaping failed\n",
stderr="exit status 1\n"
+ )
+
+ returncode, output = checker.regenerate_specs(tmp_path)
+
+ assert returncode == 1
+ assert "genspec: shaping failed" in output
+ assert "exit status 1" in output
+
+
[email protected]("check_go_sdk_spec_drift.subprocess.run", autospec=True)
+def test_read_drift_asks_git_only_about_the_generated_file(mock_run, tmp_path):
+ mock_run.return_value = subprocess.CompletedProcess(args=[], returncode=0,
stdout="", stderr="")
+
+ checker.read_drift(tmp_path)
+
+ assert mock_run.call_args.args[0] == ["git", "diff", "--",
"go-sdk/airflow/spec.gen.go"]
+ assert mock_run.call_args.kwargs["cwd"] == tmp_path
+
+
[email protected](
+ ("ci_env", "expected_exit", "expected_text"),
+ [
+ pytest.param({"CI": "true"}, 1, "this is a CI run",
id="ci-fails-loudly"),
+ pytest.param({}, 0, "SKIPPED", id="local-skips"),
+ ],
+)
[email protected]("check_go_sdk_spec_drift.shutil.which", autospec=True,
return_value=None)
+def test_missing_go_toolchain(mock_which, ci_env, expected_exit,
expected_text, monkeypatch, capsys):
+ monkeypatch.delenv("CI", raising=False)
+ for key, value in ci_env.items():
+ monkeypatch.setenv(key, value)
+
+ assert checker.main() == expected_exit
+ assert expected_text in capsys.readouterr().out
+
+
[email protected]("check_go_sdk_spec_drift.pathlib.Path.is_file", autospec=True,
return_value=False)
+def test_missing_generated_file_fails(mock_is_file, capsys):
+ assert checker.main() == 1
+ assert "not found" in capsys.readouterr().out
