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 a6380161db6 Docs: Add sdk.TIRunContext for Go-SDK (#68319)
a6380161db6 is described below
commit a6380161db6009603bbf5e439d26e31414822aea
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Thu Jun 11 19:15:55 2026 +0800
Docs: Add sdk.TIRunContext for Go-SDK (#68319)
Document the sdk.TIRunContext authoring style in the Go SDK guide and
README: declare it as the task's context parameter and read the task
instance identifiers and Dag run scheduling timestamps directly.
---
.../authoring-and-scheduling/language-sdks/go.rst | 39 +++++++++++++++++++---
go-sdk/README.md | 31 +++++++++++++++--
2 files changed, 62 insertions(+), 8 deletions(-)
diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst
b/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst
index d4ff4874049..722bd7549e5 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst
@@ -102,14 +102,13 @@ task declares only the parameters it needs.
.. code-block:: go
import (
- "context"
"log/slog"
"runtime"
"github.com/apache/airflow/go-sdk/sdk"
)
- func extract(ctx context.Context, client sdk.Client, log *slog.Logger)
(any, error) {
+ 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
@@ -119,7 +118,7 @@ task declares only the parameters it needs.
return map[string]any{"go_version": runtime.Version()}, nil
}
- func transform(ctx context.Context, client sdk.VariableClient, log
*slog.Logger) error {
+ func transform(ctx sdk.TIRunContext, client sdk.VariableClient, log
*slog.Logger) error {
val, err := client.GetVariable(ctx, "my_variable")
if err != nil {
return err
@@ -217,8 +216,9 @@ parameters your task actually needs:
* - Parameter type
- Injected value
- * - ``context.Context``
- - Cancellation/deadline context for the task. Respect it for long-running
work.
+ * - ``sdk.TIRunContext``
+ - The task's execution context: the cancellation/deadline signal plus the
task instance identifiers and
+ Dag run timestamps. Respect it for long-running work. See
:ref:`go-sdk/runtime-context`.
* - ``*slog.Logger``
- A logger whose output is routed back to the Airflow task log.
* - ``sdk.Client`` (or a narrower interface)
@@ -252,6 +252,35 @@ Go types.
Not-found lookups return sentinel errors - ``VariableNotFound``,
``ConnectionNotFound``, ``XComNotFound`` -
so you can branch on a missing value with ``errors.Is`` rather than parsing an
error string.
+.. _go-sdk/runtime-context:
+
+Reading the task runtime context
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Declare an ``sdk.TIRunContext`` parameter on a task to read the identifiers
and scheduling timestamps of the
+running task instance and its Dag run -- the Go equivalent of the execution
context the Python and Java SDKs
+expose. It is an interface that embeds ``context.Context``, so the same
``ctx`` drives cancellation and
+client calls. The runtime binds it by type, just like the other injected
parameters:
+
+.. code-block:: go
+
+ func extract(ctx sdk.TIRunContext, log *slog.Logger) (any, error) {
+ ti := ctx.TaskInstance()
+ log.Info("running",
+ "dag_id", ti.DagID,
+ "run_id", ti.RunID,
+ "task_id", ti.TaskID,
+ "try_number", ti.TryNumber,
+ "logical_date", ctx.DagRun().LogicalDate,
+ )
+ return nil, nil
+ }
+
+``ctx.TaskInstance()`` returns ``DagID``, ``RunID``, ``TaskID``, ``MapIndex``
(nil for an unmapped task),
+and ``TryNumber``; ``ctx.DagRun()`` returns ``DagID``, ``RunID``, and the
``*time.Time`` fields
+``LogicalDate``, ``DataIntervalStart``, and ``DataIntervalEnd`` (nil when the
run has no such value, e.g. a
+manual trigger).
+
.. _go-sdk/types:
XCom type mapping
diff --git a/go-sdk/README.md b/go-sdk/README.md
index 5ceb1afb4cc..bc1a02fbe40 100644
--- a/go-sdk/README.md
+++ b/go-sdk/README.md
@@ -101,18 +101,18 @@ func main() {
```
A task is an ordinary Go function. The runtime inspects its signature and
injects arguments by type:
-`context.Context`, `*slog.Logger`, and an `sdk.Client` (or a narrower
interface such as
+`sdk.TIRunContext`, `*slog.Logger`, and an `sdk.Client` (or a narrower
interface such as
`sdk.VariableClient`). An optional `(any, error)` return becomes the task's
XCom; an `error` return marks
the task failed.
```go
-func extract(ctx context.Context, client sdk.Client, log *slog.Logger) (any,
error) {
+func extract(ctx sdk.TIRunContext, client sdk.Client, log *slog.Logger) (any,
error) {
conn, err := client.GetConnection(ctx, "test_http")
// ... do work, honour ctx cancellation ...
return map[string]any{"go_version": runtime.Version()}, nil
}
-func transform(ctx context.Context, client sdk.VariableClient, log
*slog.Logger) error {
+func transform(ctx sdk.TIRunContext, client sdk.VariableClient, log
*slog.Logger) error {
val, err := client.GetVariable(ctx, "my_variable")
if err != nil {
return err
@@ -126,6 +126,31 @@ Asking for the narrowest interface a task needs (e.g.
`sdk.VariableClient` inste
unit testing easier and documents which Airflow features the task touches.
`RegisterDags` is the single
source of truth for which `dag_id`s and `task_id`s a bundle can run.
+### Reading the task runtime context
+
+Declare an `sdk.TIRunContext` parameter on a task to read the identifiers and
scheduling timestamps of the
+running task instance and its Dag run -- the Go equivalent of the execution
context the Python and Java SDKs
+expose. It is an interface that embeds `context.Context`, so the same `ctx`
drives cancellation and client
+calls. The runtime binds it by type, just like the other injected parameters:
+
+```go
+func extract(ctx sdk.TIRunContext, log *slog.Logger) (any, error) {
+ ti := ctx.TaskInstance()
+ log.Info("running",
+ "dag_id", ti.DagID,
+ "run_id", ti.RunID,
+ "task_id", ti.TaskID,
+ "try_number", ti.TryNumber,
+ "logical_date", ctx.DagRun().LogicalDate,
+ )
+ return nil, nil
+}
+```
+
+`ctx.TaskInstance()` returns `DagID`, `RunID`, `TaskID`, `MapIndex` (nil for
an unmapped task), and
+`TryNumber`; `ctx.DagRun()` returns `DagID`, `RunID`, and the `*time.Time`
fields `LogicalDate`,
+`DataIntervalStart`, and `DataIntervalEnd` (nil when the run has no such
value, e.g. a manual trigger).
+
## Deployment modes
A bundle can run in two ways. The same bundle binary works in both; you pick
one per deployment:
