authoring-go-sdk-tasks

name

authoring-go-sdk-tasks

description

Writes Airflow task logic in Go using the Airflow Go SDK. Use when the user wants to implement Airflow tasks in Go, asks about `BundleProvider`/`RegisterDags`, the `bundlev1` Registry/Dag interfaces, registering Go tasks (`AddTask`/`AddTaskWithName`), dependency injection by parameter type (`context.Context`, `sdk.TIRunContext`, `*slog.Logger`, `sdk.Client`), or reading connections/variables/XComs from Go. This skill covers the Go-specific native API; the shared Python-stub pattern and conceptual model live in authoring-language-sdk-tasks. For building/packing/shipping the bundle see deploying-go-sdk-bundles; for coordinator config see configuring-airflow-language-sdks.

Authoring Go SDK Tasks

The Airflow Go SDK implements the language-SDK model for Go: your DAG stays in Python, and each task is a compiled Go function registered inside a bundle (a single native executable). This skill covers the Go-specific native API. The shared model (the Python @task.stub pattern, ID matching, the XCom-as-JSON contract) lives in authoring-language-sdk-tasks; read that first if you are new to language SDKs.

Experimental. The Go SDK is under active development and not production-ready. Module path github.com/apache/airflow/go-sdk (Go 1.24+). APIs may change.

Related skills: authoring-language-sdk-tasks (shared Python stub + concepts), deploying-go-sdk-bundles (build, pack, and ship the bundle), configuring-airflow-language-sdks (route the queue to the Go coordinator).

---

Recap: the Python side

A Go task is paired with a Python stub that carries no logic; it declares the task, its queue, and the dependency graph. IDs must match the Go registration exactly, and queue= routes the task to the Go runtime. Full rules are in authoring-language-sdk-tasks; the minimal shape:

from airflow.sdk import dag, task


@task.stub(queue="golang")
def extract(): ...


@task.stub(queue="golang")
def transform(): ...


@dag()
def simple_dag():
    extract() >> transform()


simple_dag()

The queue value ("golang" here) is an arbitrary label that must match the queue routed to the Go coordinator (queue_to_coordinator). See configuring-airflow-language-sdks.

---

The bundle entry point

A bundle implements bundlev1.BundleProvider: report its version and register your DAGs and tasks. main is one line; bundlev1server.Serve wires the bundle to the Airflow runtime for you.

package main

import (
	"log"

	v1 "github.com/apache/airflow/go-sdk/bundle/bundlev1"
	"github.com/apache/airflow/go-sdk/bundle/bundlev1/bundlev1server"
)

type myBundle struct{}

var _ v1.BundleProvider = (*myBundle)(nil)

func (m *myBundle) GetBundleVersion() v1.BundleInfo {
	return v1.BundleInfo{Name: bundleName, Version: &bundleVersion}
}

func (m *myBundle) RegisterDags(dagbag v1.Registry) error {
	simpleDag := dagbag.AddDag("simple_dag")      // dag_id must match the Python @dag name
	simpleDag.AddTask(extract)                    // task_id is the function name; must match the stub
	simpleDag.AddTaskWithName("transform", transform) // or set the task_id explicitly
	return nil
}

func main() {
	if err := bundlev1server.Serve(&myBundle{}); err != nil {
		log.Fatal(err)
	}
}

AddTask(fn) derives the task_id from the Go function's name; use AddTaskWithName("<task_id>", fn) when that name can't match the Python stub (an unexported, renamed, or reused function). RegisterDags is the single source of truth for task identity: the bundle's manifest (used by the packer and by the coordinator) is generated by running it, never hand-written.

---

Task functions: dependency injection by parameter type

A task is an ordinary Go function. The runtime inspects its signature and injects arguments by type; declare only what you need.

Parameter type	Injected value
context.Context	Task context for cancellation. Always available.
sdk.TIRunContext	Richer context (embeds context.Context) exposing TaskInstance() and DagRun(). See Runtime context.
*slog.Logger	Logger wired to the Airflow task log.
sdk.Client	Full Airflow model access: Variables, Connections, XComs.
sdk.VariableClient / sdk.ConnectionClient / sdk.XComClient	A narrower slice of sdk.Client. Prefer the narrowest you need; it documents intent and is trivial to fake in tests.

The optional return signature is (result, error): a non-nil result is pushed as the task's return_value XCom; a non-nil error fails the task (which triggers the stub's retry policy). Returning only error, or nothing, is also valid.

func extract(ctx sdk.TIRunContext, client sdk.Client, log *slog.Logger) (any, error) {
	conn, err := client.GetConnection(ctx, "test_http")
	if err != nil {
		return nil, err
	}
	log.Info("connected", "host", conn.Host)
	return map[string]any{"go_version": runtime.Version()}, nil
}

func transform(ctx sdk.TIRunContext, client sdk.VariableClient) error {
	val, err := client.GetVariable(ctx, "my_variable")
	if err != nil {
		return err // VariableNotFound (a sentinel error) if absent
	}
	_ = val
	return nil
}

---

The sdk.Client surface

Call	Returns	Notes
GetVariable(ctx, key)	(string, error)	VariableNotFound if absent.
UnmarshalJSONVariable(ctx, key, &ptr)	error	Decode a JSON variable into a struct/pointer.
GetConnection(ctx, connID)	(Connection, error)	ConnectionNotFound if absent.
GetXCom(ctx, dagID, runID, taskID, mapIndex, key, value)	(any, error)	XComNotFound only if the key is absent; a stored null returns (nil, nil).
PushXCom(ctx, ti, key, value)	error	Rarely needed; a returned value is pushed for you.

Connection exposes ID, Type, Host, Port (int), Login *string, Password *string (nil when unset, distinct from empty), Path (schema), Extra map[string]any, plus GetURI(). Not-found cases return the sentinels sdk.VariableNotFound, sdk.ConnectionNotFound, sdk.XComNotFound.

To read an upstream task's result, call GetXCom explicitly, taking the dag_id/run_id/task_id you need from the runtime context (below).

---

Runtime context

Declare an sdk.TIRunContext parameter to read metadata about the task instance and its DAG run. It is an interface that embeds context.Context, so it is usable anywhere a context.Context is expected.

func extract(ctx sdk.TIRunContext, log *slog.Logger) error {
	ti, dagRun := ctx.TaskInstance(), ctx.DagRun()
	log.Info("running",
		"task_id", ti.TaskID,
		"run_id", dagRun.RunID,
		"logical_date", dagRun.LogicalDate)
	return nil
}

• TaskInstance(): DagID, RunID, TaskID, MapIndex *int (nil when unmapped), TryNumber.
• DagRun(): DagID, RunID, and the *time.Time timestamps LogicalDate, DataIntervalStart, DataIntervalEnd (nil when not sent).

The accessors are populated from the task's startup details before the body runs. Because TIRunContext embeds context.Context, pass it straight to client calls and cancellation checks (ctx.Done()); declare it as your context parameter by default. In tests, build the argument with sdk.NewTIRunContext(ctx, ti, dagRun) (it panics on a nil ctx).

---

Go-specific pitfalls

• IDs must match the Python stub (dag_id from AddDag, task_id from the registered function name), and the stub's queue= must route to the Go coordinator, or the task is never delivered.
• RegisterDags is authoritative. Do not hand-write the manifest; the packer generates it by running RegisterDags.
• Ask for the narrowest client interface you need (sdk.VariableClient over sdk.Client) for clearer intent and easier fakes.
• A non-nil error return fails the task and applies the stub's retries; a recovered panic is also a failure.
• See authoring-language-sdk-tasks for the language-agnostic pitfalls (one process per task instance, set queue and retries on the stub).

---

Related Skills

• authoring-language-sdk-tasks: Shared Python-stub pattern and concepts (read first).
• deploying-go-sdk-bundles: Build and pack the bundle with go tool airflow-go-pack, then deploy it for the coordinator.
• configuring-airflow-language-sdks: Route the queue to the Go coordinator (ExecutableCoordinator).
• authoring-dags: General Airflow DAG authoring.