Add @task.eks_pod TaskFlow decorator for EksPodOperator

[Leondon9]

Adds a @task.eks_pod TaskFlow decorator that runs a Python function in a pod on an Amazon EKS cluster. It wraps EksPodOperator, so the cluster kubeconfig and short-lived token come from the AWS connection (aws_conn_id) and the worker needs no aws CLI or kubeconfig.

Motivation

The usual way to run a Python function on EKS in the TaskFlow style is @task.kubernetes against a kubeconfig whose user authenticates with an aws eks get-token exec plugin. However that kubeconfig is supplied — a Kubernetes connection (kube_config), a file on the worker (kube_config_path / config_file), or the worker's default ~/.kube/config — the token is minted by an aws subprocess when the config is loaded. That has two real costs:

• the worker (and triggerer) image must ship the aws CLI, purely for auth; and
• AWS credentials have to be provisioned in that worker's environment (env vars, instance profile, IRSA, ...), separately from the aws_conn_id already used by every other AWS operator, so EKS auth ends up configured in two places.

(If Airflow runs inside the target EKS cluster, in-cluster auth avoids the exec plugin; this addresses the common case where it does not — a different or external cluster.)

@task.eks_pod wraps EksPodOperator, which builds the kubeconfig and a short-lived token from the AWS connection over boto3/STS. EKS then authenticates the same way as the rest of the Amazon provider, and the worker needs no aws CLI and no separately provisioned kubeconfig or credentials.

Before (@task.kubernetes; the worker authenticates via an aws eks get-token kubeconfig + the aws CLI):

@task.kubernetes(
    image="python:3.12-slim",
    kubernetes_conn_id="eks_kube_conn",  # one way to supply the exec-plugin kubeconfig
    namespace="default",
    in_cluster=False,
)
def process():
    return {"rows": 42}

After (@task.eks_pod, authenticating through the AWS connection):

@task.eks_pod(
    cluster_name="my-eks-cluster",
    image="python:3.12-slim",
    namespace="default",
    aws_conn_id="aws_default",
    region="us-east-1",
)
def process():
    return {"rows": 42}

Note

The script-injection logic is kept self-contained here rather than shared with the @task.kubernetes implementation in the cncf provider, to avoid a cross-provider private import. Factoring out a shared helper can be a follow-up.

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines