Use Custom Callbacks in a Dagger Pipeline

Introduction

All Dagger SDKs support adding callbacks to the pipeline invocation chain. Using a callback enables greater code reusability and modularity, and also avoids "breaking the chain" when constructing a Dagger pipeline.

This guide explains the basics of creating and using custom callbacks in a Dagger pipeline. You will learn how to:

• Create a custom callback
• Chain the callback into a Dagger pipeline

Requirements

This tutorial assumes that:

• You have a Go, Python or Node.js development environment. If not, install Go, Python or Node.js.
• You have a Dagger SDK installed for one of the above languages. If not, follow the installation instructions for the Dagger Go, Python or Node.js SDK.
• You have Docker installed and running on the host system. If not, install Docker.

Example

All Dagger SDKs support adding a callback via the With() API method. The callback must return a function that receives a Container from the chain, and returns a Container back to it.

Assume that you have the following Dagger pipeline:

Go

package main

import (
	"context"
	"fmt"
	"os"

	"dagger.io/dagger"
)

func main() {
	ctx := context.Background()

	// initialize Dagger client
	client, err := dagger.Connect(ctx, dagger.WithLogOutput(os.Stderr))
	if err != nil {
		panic(err)
	}
	defer client.Close()

	ctr := client.
		Container().
		From("alpine")

	// breaks the chain!
	ctr = AddMounts(ctr, client)

	out, err := ctr.
		WithExec([]string{"ls"}).
		Stdout(ctx)
	if err != nil {
		panic(err)
	}

	fmt.Println(out)
}

func AddMounts(ctr *dagger.Container, client *dagger.Client) *dagger.Container {
	return ctr.
		WithMountedDirectory("/foo", client.Host().Directory("/tmp/foo")).
		WithMountedDirectory("/bar", client.Host().Directory("/tmp/bar"))
}

Here, the AddMounts() function accepts a container, mounts two directories, and returns it to the main() function. Within the main() function, the call to AddMounts() breaks the Dagger pipeline construction chain.

This pipeline can be rewritten to use a callback and the With() API method, as below:

package main

import (
	"context"
	"fmt"
	"os"

	"dagger.io/dagger"
)

func main() {
	ctx := context.Background()

	// initialize Dagger client
	client, err := dagger.Connect(ctx, dagger.WithLogOutput(os.Stderr))
	if err != nil {
		panic(err)
	}
	defer client.Close()

	out, err := client.
		Container().
		From("alpine").
		With(Mounts(client)).
		WithExec([]string{"ls"}).
		Stdout(ctx)
	if err != nil {
		panic(err)
	}

	fmt.Println(out)
}

func Mounts(client *dagger.Client) dagger.WithContainerFunc {
	return func(ctr *dagger.Container) *dagger.Container {
		return ctr.
			WithMountedDirectory("/foo", client.Host().Directory("/tmp/foo")).
			WithMountedDirectory("/bar", client.Host().Directory("/tmp/bar"))
	}
}

Here, the Mounts() callback function returns a function that receives a Container from the chain, and returns a Container back to it. It can then be attached to the Dagger pipeline in the normal manner, as an argument to With().

Node.js

import { connect, Client, Container } from "@dagger.io/dagger"

connect(
  async (client: Client) => {
    let ctr = client.container().from("alpine")

    // breaks the chain!
    ctr = addMounts(ctr, client)

    const out = await ctr.withExec(["ls"]).stdout()
    console.log(out)
  },
  { LogOutput: process.stderr },
)

function addMounts(ctr: Container, client: Client): Container {
  return ctr
    .withMountedDirectory("/foo", client.host().directory("/tmp/foo"))
    .withMountedDirectory("/bar", client.host().directory("/tmp/bar"))
}

Here, the addMounts() function accepts a container, mounts two directories, and returns the container. Within the main program, the call to addMounts() breaks the Dagger pipeline construction chain.

This pipeline can be rewritten to use a callback and the with() API method, as below:

import { connect, Client, Container } from "@dagger.io/dagger"

connect(
  async (client: Client) => {
    const out = await client
      .container()
      .from("alpine")
      .with(mounts(client))
      .withExec(["ls"])
      .stdout()
    console.log(out)
  },
  { LogOutput: process.stderr },
)

function mounts(client: Client) {
  return (ctr: Container): Container =>
    ctr
      .withMountedDirectory("/foo", client.host().directory("/tmp/foo"))
      .withMountedDirectory("/bar", client.host().directory("/tmp/bar"))
}

Python

import sys

import anyio

import dagger


async def main():
    config = dagger.Config(log_output=sys.stdout)

    async with dagger.Connection(config) as client:
        ctr = client.container().from_("alpine")

        # breaks the chain!
        ctr = add_mounts(ctr, client)

        out = await ctr.with_exec(["ls"]).stdout()

    print(out)


def add_mounts(ctr: dagger.Container, client: dagger.Client):
    return ctr.with_mounted_directory(
        "/foo", client.host().directory("/tmp/foo")
    ).with_mounted_directory("/bar", client.host().directory("/tmp/bar"))


anyio.run(main)

Here, the add_mounts() function accepts a container, mounts two directories, and returns the container. Within the main program, the call to add_mounts() breaks the Dagger pipeline construction chain.

This pipeline can be rewritten to use a callback and the with() API method, as below:

import sys

import anyio

import dagger


async def main():
    config = dagger.Config(log_output=sys.stdout)

    async with dagger.Connection(config) as client:
        out = await (
            client.container()
            .from_("alpine")
            .with_(mounts(client))
            .with_exec(["ls"])
            .stdout()
        )

    print(out)


def mounts(client: dagger.Client):
    def _mounts(ctr: dagger.Container):
        return ctr.with_mounted_directory(
            "/foo", client.host().directory("/tmp/foo")
        ).with_mounted_directory("/bar", client.host().directory("/tmp/bar"))

    return _mounts


anyio.run(main)

Here's another example, this one demonstrating how to add multiple environment variables to a container using a callback:

Go

package main

import (
	"context"
	"fmt"
	"os"

	"dagger.io/dagger"
)

func main() {
	// create Dagger client
	ctx := context.Background()
	client, err := dagger.Connect(ctx, dagger.WithLogOutput(os.Stderr))
	if err != nil {
		panic(err)
	}
	defer client.Close()

	// setup container and
	// define environment variables
	ctr := client.
		Container().
		From("alpine").
		With(EnvVariables(map[string]string{
			"ENV_VAR_1": "VALUE 1",
			"ENV_VAR_2": "VALUE 2",
			"ENV_VAR_3": "VALUE 3",
		})).
		WithExec([]string{"env"})

	// print environment variables
	out, err := ctr.Stdout(ctx)
	if err != nil {
		panic(err)
	}
	fmt.Println(out)
}

func EnvVariables(envs map[string]string) dagger.WithContainerFunc {
	return func(c *dagger.Container) *dagger.Container {
		for key, value := range envs {
			c = c.WithEnvVariable(key, value)
		}
		return c
	}
}

Node.js

import { connect, Client, Container } from "@dagger.io/dagger"

// create Dagger client
connect(
  async (client: Client) => {
    // setup container and
    // define environment variables
    const ctr = client
      .container()
      .from("alpine")
      .with(
        envVariables({
          ENV_VAR_1: "VALUE 1",
          ENV_VAR_2: "VALUE 2",
          ENV_VAR_3: "VALUE 3",
        }),
      )
      .withExec(["env"])

    // print environment variables
    console.log(await ctr.stdout())
  },
  { LogOutput: process.stderr },
)

function envVariables(envs: Record<string, string>) {
  return (c: Container): Container => {
    Object.entries(envs).forEach(([key, value]) => {
      c = c.withEnvVariable(key, value)
    })
    return c
  }
}

Python

import sys

import anyio

import dagger


async def main():
    # create Dagger client
    async with dagger.Connection(dagger.Config(log_output=sys.stderr)) as client:
        # setup container and
        # define environment variables
        ctr = (
            client.container()
            .from_("alpine")
            .with_(
                env_variables(
                    {
                        "ENV_VAR_1": "VALUE 1",
                        "ENV_VAR_2": "VALUE 2",
                        "ENV_VAR_3": "VALUE 3",
                    }
                )
            )
            .with_exec(["env"])
        )

        # print environment variables
        print(await ctr.stdout())


def env_variables(envs: dict[str, str]):
    def env_variables_inner(ctr: dagger.Container):
        for key, value in envs.items():
            ctr = ctr.with_env_variable(key, value)
        return ctr

    return env_variables_inner


anyio.run(main)

Conclusion

This guide explained how to create and chain custom callback functions in your Dagger pipeline.

Use the API Key Concepts page and the Go, Node.js and Python SDK References to learn more about Dagger.