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 4e23910701b Go SDK: check Go naming conventions with staticcheck
(#73874)
4e23910701b is described below
commit 4e23910701b4ac4524c68ef6a73c2ec3f7ffcfd9
Author: PoAn Yang <[email protected]>
AuthorDate: Tue Sep 29 18:23:00 2026 +0900
Go SDK: check Go naming conventions with staticcheck (#73874)
Signed-off-by: PoAn Yang <[email protected]>
---
go-sdk/.pre-commit-config.yaml | 7 +++++++
.../adr/0007-mixed-lang-task-handler-interface.md | 6 +++---
go-sdk/airflow/bundle.go | 20 ++++++++++----------
go-sdk/airflow/task_handler.go | 10 +++++-----
go-sdk/airflow/task_handler_test.go | 6 +++---
.../bundle/concurrentxcom/concurrentxcom_test.go | 2 +-
go-sdk/internal/bundle/task.go | 2 +-
go-sdk/pkg/binding/binding_test.go | 3 ++-
go-sdk/pkg/execution/client.go | 22 +++++++++++-----------
go-sdk/pkg/execution/client_test.go | 6 +++---
go-sdk/pkg/execution/comms.go | 13 +++++++++----
go-sdk/pkg/execution/comms_test.go | 10 +++++-----
go-sdk/pkg/execution/integration_test.go | 14 +++++++-------
go-sdk/pkg/execution/messages.go | 6 +++---
go-sdk/pkg/execution/messages_test.go | 4 ++--
go-sdk/sdk/sdk.go | 4 ++--
16 files changed, 74 insertions(+), 61 deletions(-)
diff --git a/go-sdk/.pre-commit-config.yaml b/go-sdk/.pre-commit-config.yaml
index 14381b4ffbd..523b64d8dd0 100644
--- a/go-sdk/.pre-commit-config.yaml
+++ b/go-sdk/.pre-commit-config.yaml
@@ -66,3 +66,10 @@ repos:
additional_dependencies: [github.com/segmentio/golines@latest,
mvdan.cc/[email protected]]
types: [go]
language: golang
+ - id: staticcheck
+ name: Check naming in Go code
+ entry: staticcheck -checks ST1003 ./...
+ additional_dependencies: [honnef.co/go/tools/cmd/[email protected]]
+ types: [go]
+ pass_filenames: false
+ language: golang
diff --git a/go-sdk/adr/0007-mixed-lang-task-handler-interface.md
b/go-sdk/adr/0007-mixed-lang-task-handler-interface.md
index f3739149248..aa159e3357f 100644
--- a/go-sdk/adr/0007-mixed-lang-task-handler-interface.md
+++ b/go-sdk/adr/0007-mixed-lang-task-handler-interface.md
@@ -31,7 +31,7 @@ Proposed.
`main` reads build, register, serve, with `bundle.Serve()` as its last
statement.
It replaces `BundleProvider` and `Registry`, the callback and the write
half of the same bundle.
2. **`bundle.Register(items ...airflow.Registerable)`** is the single
registration verb, taking native Dags and task handlers.
-3. **A Go bundle registers task handlers, not Dags**:
`airflow.TaskHandler(dagId, taskId, fn)`, the Go body for a task Python
declares with `@task.stub`.
+3. **A Go bundle registers task handlers, not Dags**:
`airflow.TaskHandler(dagID, taskID, fn)`, the Go body for a task Python
declares with `@task.stub`.
4. **Both dag_id and task_id are written out on TaskHandler definition**,
because Python owns them; nothing is derived from the Go function name.
5. **Every handler must take an `airflow.Context` first**: a struct embedding
`context.Context`, exposing `Logger()`, `Client()`, `TaskInstance()`, and
`DagRun()`.
What Airflow supplies a task arrives as a method on that value rather than
as a parameter of its own.
@@ -134,7 +134,7 @@ func (b *BundleRef) Serve() error
// once there is one.
type Registerable interface{ registerable() }
-func TaskHandler(dagId, taskId string, fn any) Registerable
+func TaskHandler(dagID, taskID string, fn any) Registerable
// Context is what every handler takes first. It is a context, so it passes
straight to the
// logger and the client rather than being stored inside either of them.
@@ -161,7 +161,7 @@ func FromContext(ctx context.Context) (Context, bool)
## Alternatives
-- **`airflow.TaskHandler(dagId, fn, airflow.WithTaskId(...))`**, defaulting
the task_id to the Go
+- **`airflow.TaskHandler(dagID, fn, airflow.WithTaskId(...))`**, defaulting
the task_id to the Go
function name. Rejected: see the ids in Context above.
- **Package-level accessors over a plain `context.Context`**
(`airflow.Logger(ctx)`, `airflow.Client(ctx)`), leaving the handler's first
parameter as `context.Context`.
Rejected: `airflow.Logger(ctx)` reads oddly next to `actx.Logger()`, asking
the package for something the context already holds.
diff --git a/go-sdk/airflow/bundle.go b/go-sdk/airflow/bundle.go
index 417964b42d4..697b6b0ae7d 100644
--- a/go-sdk/airflow/bundle.go
+++ b/go-sdk/airflow/bundle.go
@@ -87,7 +87,7 @@ func (b *BundleRef) Register(items ...Registerable) {
for _, item := range items {
switch item := item.(type) {
case *taskHandler:
- b.taskHandlers.add(item.dagId, item.taskId, item.task)
+ b.taskHandlers.add(item.dagID, item.taskID, item.task)
case *DagRef:
if item == nil {
panic("airflow.BundleRef.Register: cannot
register a nil *airflow.DagRef")
@@ -114,32 +114,32 @@ var (
_ bundle.EnumerableBundle = (*taskHandlerMap)(nil)
)
-func (m *taskHandlerMap) add(dagId, taskId string, task bundle.Task) {
+func (m *taskHandlerMap) add(dagID, taskID string, task bundle.Task) {
m.mu.Lock()
defer m.mu.Unlock()
if m.handlers == nil {
m.handlers = make(map[string]map[string]bundle.Task)
}
- dagHandlers, exists := m.handlers[dagId]
+ dagHandlers, exists := m.handlers[dagID]
if !exists {
dagHandlers = make(map[string]bundle.Task)
- m.handlers[dagId] = dagHandlers
+ m.handlers[dagID] = dagHandlers
}
- if _, exists := dagHandlers[taskId]; exists {
+ if _, exists := dagHandlers[taskID]; exists {
panic(fmt.Sprintf(
- "airflow.BundleRef.Register: task %q of Dag %q is
already registered", taskId, dagId,
+ "airflow.BundleRef.Register: task %q of Dag %q is
already registered", taskID, dagID,
))
}
- dagHandlers[taskId] = task
- m.order = append(m.order, bundle.TaskHandlerInfo{DagID: dagId, TaskID:
taskId})
+ dagHandlers[taskID] = task
+ m.order = append(m.order, bundle.TaskHandlerInfo{DagID: dagID, TaskID:
taskID})
}
-func (m *taskHandlerMap) LookupTask(dagId, taskId string) (bundle.Task, bool) {
+func (m *taskHandlerMap) LookupTask(dagID, taskID string) (bundle.Task, bool) {
m.mu.RLock()
defer m.mu.RUnlock()
- task, exists := m.handlers[dagId][taskId]
+ task, exists := m.handlers[dagID][taskID]
return task, exists
}
diff --git a/go-sdk/airflow/task_handler.go b/go-sdk/airflow/task_handler.go
index c732308a1e6..7af494cc39a 100644
--- a/go-sdk/airflow/task_handler.go
+++ b/go-sdk/airflow/task_handler.go
@@ -25,7 +25,7 @@ import (
)
type taskHandler struct {
- dagId, taskId string
+ dagID, taskID string
task bundle.Task
}
@@ -34,7 +34,7 @@ func (*taskHandler) registerable() {}
// TaskHandler makes fn the Go body of a task that a Python Dag declares with
@task.stub.
// Pass what it returns to [BundleRef.Register].
//
-// dagId is the dag_id of that Python Dag, and taskId is the task_id of the
stub task.
+// dagID is the dag_id of that Python Dag, and taskID is the task_id of the
stub task.
//
// fn takes a [Context] first, as the package documentation describes. Every
parameter after
// the Context is data, filled from the arguments of the Python stub's
TaskFlow call.
@@ -45,12 +45,12 @@ func (*taskHandler) registerable() {}
// not a function, does not take a Context first, or does not return an error.
// main calls TaskHandler before Serve, so a handler that fails the check
stops the executable as
// soon as it starts instead of when the task first runs.
-func TaskHandler(dagId, taskId string, fn any) Registerable {
+func TaskHandler(dagID, taskID string, fn any) Registerable {
task, err := newTaskFunction(fn)
if err != nil {
- panic(fmt.Sprintf("airflow.TaskHandler(%q, %q): %v", dagId,
taskId, err))
+ panic(fmt.Sprintf("airflow.TaskHandler(%q, %q): %v", dagID,
taskID, err))
}
- return &taskHandler{dagId: dagId, taskId: taskId, task: task}
+ return &taskHandler{dagID: dagID, taskID: taskID, task: task}
}
// newTaskFunction checks fn before bundle.NewTaskFunction does, so that the
error names the Go
diff --git a/go-sdk/airflow/task_handler_test.go
b/go-sdk/airflow/task_handler_test.go
index 4f0787c1982..8dfa7707d63 100644
--- a/go-sdk/airflow/task_handler_test.go
+++ b/go-sdk/airflow/task_handler_test.go
@@ -42,15 +42,15 @@ func literalArg(name, jsonType string, value any)
map[string]any {
}
func startupDetails(
- dagId, taskId string,
+ dagID, taskID string,
args ...genmodels.TaskArgBinding,
) *genmodels.StartupDetails {
mapIndex := -1
details := &genmodels.StartupDetails{
TI: genmodels.TaskInstance{
ID: "550e8400-e29b-41d4-a716-446655440000",
- DagID: dagId,
- TaskID: taskId,
+ DagID: dagID,
+ TaskID: taskID,
RunID: "run1",
TryNumber: 2,
MapIndex: &mapIndex,
diff --git a/go-sdk/example/bundle/concurrentxcom/concurrentxcom_test.go
b/go-sdk/example/bundle/concurrentxcom/concurrentxcom_test.go
index 4421fb48b61..d3a8a7deea8 100644
--- a/go-sdk/example/bundle/concurrentxcom/concurrentxcom_test.go
+++ b/go-sdk/example/bundle/concurrentxcom/concurrentxcom_test.go
@@ -54,7 +54,7 @@ func (m *mockXComClient) PushXCom(
func (m *mockXComClient) GetXCom(
ctx context.Context,
- dagId, runId, taskId string,
+ dagID, runID, taskID string,
mapIndex *int,
key string,
value any,
diff --git a/go-sdk/internal/bundle/task.go b/go-sdk/internal/bundle/task.go
index a2399fd9200..fe806da2264 100644
--- a/go-sdk/internal/bundle/task.go
+++ b/go-sdk/internal/bundle/task.go
@@ -40,7 +40,7 @@ type Task interface {
// Bundle looks up a registered task by dag_id and task_id. The coordinator
// runtime uses Bundle to find the task the supervisor asked for.
type Bundle interface {
- LookupTask(dagId, taskId string) (Task, bool)
+ LookupTask(dagID, taskID string) (Task, bool)
}
// TaskHandlerInfo identifies a registered task handler by its dag_id and
task_id.
diff --git a/go-sdk/pkg/binding/binding_test.go
b/go-sdk/pkg/binding/binding_test.go
index 704ad9a90d9..2243c339148 100644
--- a/go-sdk/pkg/binding/binding_test.go
+++ b/go-sdk/pkg/binding/binding_test.go
@@ -770,7 +770,8 @@ func (s *BindingSuite) TestAnalyzeStructValidation() {
Bad chan int `arg:"bad"`
}
type foldedDuplicateArgNames struct {
- RegionCode string
+ RegionCode string
+ //lint:ignore ST1003 the test needs a field that differs from
RegionCode only by an underscore
Region_code string
}
diff --git a/go-sdk/pkg/execution/client.go b/go-sdk/pkg/execution/client.go
index b79c6843ab9..118ffd893da 100644
--- a/go-sdk/pkg/execution/client.go
+++ b/go-sdk/pkg/execution/client.go
@@ -38,15 +38,15 @@ const (
errCodeXComNotFound = "XCOM_NOT_FOUND"
)
-// translateApiError converts a supervisor *ApiError whose Err field matches
+// translateAPIError converts a supervisor *APIError whose Err field matches
// code into a sentinel-wrapped error. Any other error - including a
-// *ApiError with a different code - is returned unchanged so callers can keep
+// *APIError with a different code - is returned unchanged so callers can keep
// distinguishing transport / server errors from "thing not found".
-func translateApiError(err error, code string, sentinel error, key string)
error {
+func translateAPIError(err error, code string, sentinel error, key string)
error {
if err == nil {
return nil
}
- var apiErr *ApiError
+ var apiErr *APIError
if errors.As(err, &apiErr) && apiErr.Err == code {
return fmt.Errorf("%w: %q", sentinel, key)
}
@@ -79,7 +79,7 @@ func (c *CoordinatorClient) GetVariable(ctx context.Context,
key string) (string
genmodels.GetVariable{Key: key},
)
if err != nil {
- return "", translateApiError(err, errCodeVariableNotFound,
sdk.VariableNotFound, key)
+ return "", translateAPIError(err, errCodeVariableNotFound,
sdk.VariableNotFound, key)
}
var result genmodels.VariableResult
@@ -154,7 +154,7 @@ func (c *CoordinatorClient) GetConnection(
genmodels.GetConnection{ConnID: connID},
)
if err != nil {
- return sdk.Connection{}, translateApiError(
+ return sdk.Connection{}, translateAPIError(
err, errCodeConnectionNotFound, sdk.ConnectionNotFound,
connID,
)
}
@@ -197,16 +197,16 @@ func (c *CoordinatorClient) GetConnection(
// GetXCom requests an XCom value from the supervisor.
func (c *CoordinatorClient) GetXCom(
ctx context.Context,
- dagId, runId, taskId string,
+ dagID, runID, taskID string,
mapIndex *int,
key string,
_ any,
) (any, error) {
msg := genmodels.GetXCom{
Key: key,
- DagID: dagId,
- TaskID: taskId,
- RunID: runId,
+ DagID: dagID,
+ TaskID: taskID,
+ RunID: runID,
}
// Assign the pointer, not the dereferenced int: map_index is a nullable
// interface{} field and msgpack's omitempty treats an interface{}
holding
@@ -217,7 +217,7 @@ func (c *CoordinatorClient) GetXCom(
resp, err := c.comm.Communicate(ctx, msg)
if err != nil {
- return nil, translateApiError(err, errCodeXComNotFound,
sdk.XComNotFound, key)
+ return nil, translateAPIError(err, errCodeXComNotFound,
sdk.XComNotFound, key)
}
var result genmodels.XComResult
diff --git a/go-sdk/pkg/execution/client_test.go
b/go-sdk/pkg/execution/client_test.go
index cd72641c40a..0c0a09d7cdb 100644
--- a/go-sdk/pkg/execution/client_test.go
+++ b/go-sdk/pkg/execution/client_test.go
@@ -137,7 +137,7 @@ func TestCoordinatorClientErrorTranslation(t *testing.T) {
}
}
-// TestCoordinatorClientErrorPassThrough verifies that unrelated *ApiError
+// TestCoordinatorClientErrorPassThrough verifies that unrelated *APIError
// values (e.g. a generic API_SERVER_ERROR) are returned unchanged.
func TestCoordinatorClientErrorPassThrough(t *testing.T) {
responsePayload := encodeResponseFrame(t, 0, nil, map[string]any{
@@ -156,7 +156,7 @@ func TestCoordinatorClientErrorPassThrough(t *testing.T) {
require.Error(t, err)
assert.False(t, errors.Is(err, sdk.VariableNotFound),
"generic supervisor errors must not be translated to
VariableNotFound")
- var apiErr *ApiError
+ var apiErr *APIError
require.True(t, errors.As(err, &apiErr))
assert.Equal(t, "API_SERVER_ERROR", apiErr.Err)
}
@@ -267,7 +267,7 @@ func TestCoordinatorClientVariableWriteErrors(t *testing.T)
{
comm := NewCoordinatorComm(&responseBuf, io.Discard,
logger)
client := NewCoordinatorClient(comm)
- var apiErr *ApiError
+ var apiErr *APIError
require.ErrorAs(t, tc.call(client), &apiErr)
assert.Equal(t, "API_SERVER_ERROR", apiErr.Err)
})
diff --git a/go-sdk/pkg/execution/comms.go b/go-sdk/pkg/execution/comms.go
index b6fe9771474..fa31e951755 100644
--- a/go-sdk/pkg/execution/comms.go
+++ b/go-sdk/pkg/execution/comms.go
@@ -121,7 +121,7 @@ func (c *CoordinatorComm) SendRequest(id int64, body any)
error {
//
// If the response carries an error (either as the third element of a 3-tuple
// frame or as a body whose "type" is "ErrorResponse") it is returned as an
-// *ApiError. If the dispatcher's read loop has terminated, the underlying read
+// *APIError. If the dispatcher's read loop has terminated, the underlying read
// error is returned wrapped in ErrDispatcherClosed.
//
// On success it returns the response body's raw msgpack bytes; the caller
@@ -248,15 +248,20 @@ func (c *CoordinatorComm) readLoop() {
}
}
-// ApiError represents an error returned by the supervisor over the comm
socket.
-type ApiError struct {
+// APIError represents an error returned by the supervisor over the comm
socket.
+type APIError struct {
Err string
Detail any
}
-func (e *ApiError) Error() string {
+func (e *APIError) Error() string {
if e.Detail != nil {
return fmt.Sprintf("[%s] %v", e.Err, e.Detail)
}
return e.Err
}
+
+// Deprecated: Use APIError.
+//
+//lint:ignore ST1003 kept so code that type-asserts *execution.ApiError still
compiles
+type ApiError = APIError
diff --git a/go-sdk/pkg/execution/comms_test.go
b/go-sdk/pkg/execution/comms_test.go
index e3a5b452d29..b57e393793b 100644
--- a/go-sdk/pkg/execution/comms_test.go
+++ b/go-sdk/pkg/execution/comms_test.go
@@ -130,7 +130,7 @@ func TestCoordinatorCommCommunicateError(t *testing.T) {
)
require.Error(t, err)
- apiErr, ok := err.(*ApiError)
+ apiErr, ok := err.(*APIError)
require.True(t, ok)
assert.Equal(t, "not_found", apiErr.Err)
}
@@ -158,16 +158,16 @@ func TestCoordinatorCommCommunicateBodyError(t
*testing.T) {
)
require.Error(t, err)
- apiErr, ok := err.(*ApiError)
+ apiErr, ok := err.(*APIError)
require.True(t, ok)
assert.Equal(t, "server_error", apiErr.Err)
}
-func TestApiErrorFormat(t *testing.T) {
- err := &ApiError{Err: "not_found", Detail: "Variable 'x' not found"}
+func TestAPIErrorFormat(t *testing.T) {
+ err := &APIError{Err: "not_found", Detail: "Variable 'x' not found"}
assert.Equal(t, "[not_found] Variable 'x' not found", err.Error())
- err2 := &ApiError{Err: "server_error"}
+ err2 := &APIError{Err: "server_error"}
assert.Equal(t, "server_error", err2.Error())
}
diff --git a/go-sdk/pkg/execution/integration_test.go
b/go-sdk/pkg/execution/integration_test.go
index 251d82d4ab0..c53bed89695 100644
--- a/go-sdk/pkg/execution/integration_test.go
+++ b/go-sdk/pkg/execution/integration_test.go
@@ -83,21 +83,21 @@ type testBundle map[string]testDag
type testDag map[string]bundle.Task
-func (b testBundle) AddDag(dagId string) testDag {
- b[dagId] = testDag{}
- return b[dagId]
+func (b testBundle) AddDag(dagID string) testDag {
+ b[dagID] = testDag{}
+ return b[dagID]
}
-func (d testDag) AddTaskWithName(taskId string, fn any) {
+func (d testDag) AddTaskWithName(taskID string, fn any) {
task, err := bundle.NewTaskFunction(fn)
if err != nil {
panic(err)
}
- d[taskId] = task
+ d[taskID] = task
}
-func (b testBundle) LookupTask(dagId, taskId string) (bundle.Task, bool) {
- task, ok := b[dagId][taskId]
+func (b testBundle) LookupTask(dagID, taskID string) (bundle.Task, bool) {
+ task, ok := b[dagID][taskID]
return task, ok
}
diff --git a/go-sdk/pkg/execution/messages.go b/go-sdk/pkg/execution/messages.go
index bb81d60c0a4..1dd59620125 100644
--- a/go-sdk/pkg/execution/messages.go
+++ b/go-sdk/pkg/execution/messages.go
@@ -123,7 +123,7 @@ func decodeBody(raw msgpack.RawMessage, dst any) error {
// apiErrorFromFrame returns the supervisor error carried by a frame, or nil if
// it is not an error reply. An error arrives either as the third element of a
// 3-tuple frame (frame.Err) or as a 2-tuple body whose "type" is
"ErrorResponse".
-func apiErrorFromFrame(f IncomingFrame) *ApiError {
+func apiErrorFromFrame(f IncomingFrame) *APIError {
var raw msgpack.RawMessage
switch {
case !isNilRaw(f.Err):
@@ -146,7 +146,7 @@ func apiErrorFromFrame(f IncomingFrame) *ApiError {
if errCode == "" {
errCode = string(genmodels.ErrorTypeGENERICERROR)
}
- return &ApiError{
+ return &APIError{
Err: errCode,
Detail: fmt.Sprintf("undecodable error frame: %v", err),
}
@@ -155,7 +155,7 @@ func apiErrorFromFrame(f IncomingFrame) *ApiError {
if resp.Detail != nil {
detail = map[string]any(*resp.Detail)
}
- return &ApiError{Err: string(resp.Error), Detail: detail}
+ return &APIError{Err: string(resp.Error), Detail: detail}
}
// ifaceString returns the string carried by a nullable schema field decoded as
diff --git a/go-sdk/pkg/execution/messages_test.go
b/go-sdk/pkg/execution/messages_test.go
index bf2a51249d5..fa81df4ea66 100644
--- a/go-sdk/pkg/execution/messages_test.go
+++ b/go-sdk/pkg/execution/messages_test.go
@@ -492,7 +492,7 @@ func TestPeekBodyType(t *testing.T) {
assert.Equal(t, "", peekBodyType(msgpack.RawMessage{0xc0})) // msgpack
nil
}
-func TestApiErrorFromFrame(t *testing.T) {
+func TestAPIErrorFromFrame(t *testing.T) {
t.Run("error element of 3-tuple", func(t *testing.T) {
f := IncomingFrame{
Body: marshalBody(t, map[string]any{"type":
"ConnectionResult"}),
@@ -536,7 +536,7 @@ func TestApiErrorFromFrame(t *testing.T) {
t.Run("off-contract detail still recovers the error code", func(t
*testing.T) {
// detail is a string instead of the schema's object|null; the
typed
- // error code must survive so translateApiError maps it
correctly.
+ // error code must survive so translateAPIError maps it
correctly.
f := IncomingFrame{
Err: marshalBody(
t,
diff --git a/go-sdk/sdk/sdk.go b/go-sdk/sdk/sdk.go
index f047216108e..fdfa677268c 100644
--- a/go-sdk/sdk/sdk.go
+++ b/go-sdk/sdk/sdk.go
@@ -103,14 +103,14 @@ type ConnectionClient interface {
// another task's XCom, or to push under a custom key.
type XComClient interface {
// GetXCom returns the value stored under key by the task identified by
- // dagId/runId/taskId. For a mapped task instance pass its mapIndex,
+ // dagID/runID/taskID. For a mapped task instance pass its mapIndex,
// otherwise pass nil. If no value exists the error wraps XComNotFound.
//
// value is reserved for future typed decoding and is currently
ignored; the
// stored value is returned as the first result instead.
GetXCom(
ctx context.Context,
- dagId, runId, taskId string,
+ dagID, runID, taskID string,
mapIndex *int,
key string,
value any,
