Dagger : connexion, cycle async et gestion des erreurs

En Python, les pipelines Dagger utilisent l'async/await pour gérer les opérations asynchrones. La particularité de Dagger est sa lazy evaluation : le code que vous écrivez ne s'exécute pas immédiatement, il construit un graphe d'opérations qui sera exécuté quand vous le demanderez explicitement.

Prérequis : Avoir installé Dagger et créé un module. Si vous n'êtes pas familier avec async/await en Python, consultez la documentation officielle asyncio.

Le cycle de vie d'un pipeline Dagger

Quand vous appelez des méthodes comme dag.container().from_(), vous ne lancez pas de conteneurs, vous décrivez ce que vous voulez faire. L'exécution réelle se produit seulement quand vous appelez une méthode "terminale" comme stdout(), sync() ou export().

Lazy evaluation : pourquoi c'est important

La lazy evaluation offre plusieurs avantages :

Optimisation automatique : Dagger peut réorganiser et paralléliser les opérations
Cache intelligent : les étapes identiques sont exécutées une seule fois
Composition : vous pouvez construire des pipelines complexes puis les exécuter

Quand l'exécution se déclenche-t-elle ?

L'exécution se déclenche quand vous appelez une méthode qui demande un résultat :

Méthode	Ce qu'elle fait	Retour
`stdout()`	Récupère la sortie standard	`str`
`stderr()`	Récupère la sortie d'erreur	`str`
`sync()`	Force l'exécution	l'objet lui-même
`export()`	Exporte un fichier/répertoire	chemin
`contents()`	Lit le contenu d'un fichier	`str`

Dagger	GitLab CI équivalent
`stdout()`	Lire les logs d'un job
`export()`	`artifacts: paths:`
`sync()`	Fin d'un job (le runner attend)
`contents()`	Télécharger un artefact

En GitLab, vous récupérez la sortie après le job. En Dagger, vous la récupérez dans votre code, vous pouvez parser, valider, décider.

Fonctions synchrones vs asynchrones

Le SDK Python Dagger vous permet d'écrire deux types de fonctions.

Fonction synchrone (pas d'await)

Une fonction synchrone retourne directement une valeur ou un objet Dagger sans attendre d'exécution :

@function
def hello(self, name: str = "World") -> str:
    """Retourne un message — pas d'interaction avec Dagger."""
    return f"Bonjour, {name} !"

dagger call hello --name="DevOps"
# Bonjour, DevOps !

Vous pouvez aussi retourner un Container sans l'exécuter :

@function
def build_container(self) -> dagger.Container:
    """Retourne un conteneur configuré (lazy - pas encore exécuté)."""
    return (
        dag.container()
        .from_("python:3.12-slim")
        .with_exec(["python", "--version"])
    )

Ce code ne lance aucun conteneur. Il ne fait que décrire le pipeline.

Fonction asynchrone (avec await)

Une fonction asynchrone attend le résultat d'une opération qui force l'exécution :

@function
async def get_python_version(self) -> str:
    """Récupère la version de Python — force l'exécution."""
    result = await (
        dag.container()
        .from_("python:3.12-slim")
        .with_exec(["python", "--version"])
        .stdout()  # <-- Force l'exécution
    )
    return result.strip()

dagger call get-python-version
# Python 3.12.12

Récupérer stdout et stderr

Dagger sépare proprement la sortie standard et la sortie d'erreur.

stdout() : la sortie standard

@function
async def get_output(self) -> str:
    """Récupère la sortie standard d'une commande."""
    return await (
        dag.container()
        .from_("alpine:latest")
        .with_exec(["echo", "Hello from stdout"])
        .stdout()
    )

stderr() : la sortie d'erreur

@function
async def get_error(self) -> str:
    """Récupère la sortie d'erreur d'une commande."""
    return await (
        dag.container()
        .from_("alpine:latest")
        .with_exec(["sh", "-c", "echo 'Ceci va sur stderr' >&2"])
        .stderr()
    )

Les deux à la fois

Pour récupérer les deux sorties, stockez le conteneur dans une variable :

@function
async def show_outputs(self) -> str:
    """Récupère stdout ET stderr séparément."""
    container = (
        dag.container()
        .from_("alpine:latest")
        .with_exec(["sh", "-c", "echo stdout && echo stderr >&2"])
    )

    stdout_content = await container.stdout()
    stderr_content = await container.stderr()

    return f"STDOUT: {stdout_content.strip()} | STDERR: {stderr_content.strip()}"

dagger call show-outputs
# STDOUT: stdout | STDERR: stderr

sync() : forcer l'exécution sans récupérer la sortie

Parfois, vous voulez vous assurer qu'une étape est exécutée sans récupérer sa sortie. C'est le rôle de sync() :

@function
async def prepare_environment(self) -> dagger.Container:
    """Prépare l'environnement et s'assure qu'il est construit."""
    container = (
        dag.container()
        .from_("python:3.12-slim")
        .with_exec(["pip", "install", "pytest"])
    )

    # sync() force l'exécution et retourne le même Container
    await container.sync()

    return container

sync() est l'équivalent de needs: en GitLab CI, il garantit qu'une étape est terminée avant de continuer. La différence : en GitLab, needs est déclaratif (vous dites "attends ce job"). En Dagger, sync() est impératif (vous dites "exécute maintenant et attends").

# GitLab : déclaratif
test:
  needs: [build]  # attend que build soit fini

# Dagger : impératif
await build_container.sync()  # exécute et attend
test_container = build_container.with_exec(["pytest"])

Gestion des erreurs

Quand une commande échoue dans un conteneur, Dagger lève une exception. Vous pouvez la capturer avec try/except.

Capturer les erreurs d'exécution

import dagger
from dagger import dag, function, object_type

@object_type
class MonModule:
    @function
    async def demo_error_handling(self) -> str:
        """Capture les erreurs d'exécution."""
        try:
            await (
                dag.container()
                .from_("alpine:latest")
                .with_exec(["cat", "/fichier-inexistant"])
                .stdout()
            )
            return "Succès"
        except dagger.ExecError as e:
            return f"Erreur capturée : {e}"

dagger call demo-error-handling
# Erreur capturée : exit code: 1 [traceparent:...]

Types d'erreurs Dagger

Exception	Cause
`dagger.ExecError`	Une commande `with_exec()` a échoué (code de sortie ≠ 0)
`dagger.QueryError`	Erreur lors de l'appel à l'API Dagger
`Exception`	Autres erreurs Python

Dagger	GitLab CI
`try/except dagger.ExecError`	`allow_failure: true`
Retry dans le code Python	`retry: 2`
Condition dans le `except`	`rules: when: on_failure`

Avantage Dagger : vous pouvez inspecter l'erreur, décider dynamiquement de réessayer, logger, envoyer une alerte, tout ça dans le même code. En GitLab, ces logiques sont dispersées entre YAML et scripts.

Pattern recommandé

@function
async def safe_operation(self) -> str:
    """Pattern de gestion d'erreurs robuste."""
    try:
        result = await (
            dag.container()
            .from_("alpine:latest")
            .with_exec(["commande", "risquee"])
            .stdout()
        )
        return f"Succès : {result}"
    except dagger.ExecError as e:
        # Erreur de commande (code de sortie ≠ 0)
        return f"La commande a échoué : {e}"
    except dagger.QueryError as e:
        # Erreur d'API Dagger
        return f"Erreur Dagger : {e}"
    except Exception as e:
        # Autres erreurs
        return f"Erreur inattendue : {type(e).__name__}: {e}"

Debug interactif

Dagger offre plusieurs outils pour debugger vos pipelines.

Terminal interactif

Vous pouvez ouvrir un terminal dans n'importe quel conteneur :

dagger call build-container terminal

Cela ouvre un shell interactif dans le conteneur, où vous pouvez explorer l'état du système de fichiers, exécuter des commandes manuellement, etc.

Mode --interactive

Si une commande échoue, vous pouvez relancer avec --interactive pour obtenir un terminal au point d'échec :

dagger call --interactive ma-fonction

Verbosité des logs

Ajoutez des flags -v pour plus de détails :

dagger call -v ma-fonction     # Garde les spans visibles
dagger call -vv ma-fonction    # Révèle les spans internes
dagger call -vvv ma-fonction   # Révèle les spans < 100ms
dagger call --debug ma-fonction  # Informations de debug complètes

Dagger	GitLab CI
`dagger call --interactive`	SSH dans le runner (si configuré)
`terminal`	`artifacts: paths: ["/app"]` puis inspection
`-v`, `-vv`, `-vvv`	`CI_DEBUG_TRACE=true`

Avantage Dagger : le debug est instantané et local. Pas besoin de pusher, attendre le pipeline, télécharger les artifacts. Vous ouvrez un terminal dans le conteneur exact où l'erreur s'est produite.

Exemple complet commenté

Cet exemple rassemble tous les concepts vus dans ce guide. Prenez le temps de le lire, chaque ligne applique une notion expliquée précédemment.

Le code

import dagger
from dagger import dag, function, object_type


@object_type
class MonPipeline:
    """Pipeline avec gestion async et erreurs."""

    @function
    async def build_and_test(self, source: dagger.Directory) -> str:
        """Build et teste un projet Python."""
        try:
            # ① LAZY : on décrit le pipeline, rien ne s'exécute encore
            container = (
                dag.container()
                .from_("python:3.12-slim")
                .with_directory("/app", source)
                .with_workdir("/app")
                .with_exec(["pip", "install", "-e", "."])
            )

            # ② SYNC : on force l'exécution pour valider le build
            # Si pip échoue ici, on saute directement au except
            await container.sync()

            # ③ STDOUT : on récupère la sortie des tests
            # C'est aussi un trigger, donc await obligatoire
            test_output = await (
                container
                .with_exec(["pytest", "-v"])
                .stdout()
            )

            return f"Tests réussis :\n{test_output}"

        except dagger.ExecError as e:
            # ④ ERREUR : une commande a échoué (pip install ou pytest)
            return f"Échec : {e}"

Décryptage ligne par ligne

Ligne	Concept	Ce qui se passe
`dag.container().from_()...`	Lazy	On construit un graphe, aucun conteneur ne démarre
`await container.sync()`	Trigger	L'Engine exécute tout jusqu'ici (pull image + pip install)
`await ... .stdout()`	Async + Trigger	On attend le résultat de pytest
`except dagger.ExecError`	Gestion erreurs	On capture les codes de sortie ≠ 0

Lancer l'exemple

Pour tester ce pipeline sur un projet Python local :

# Depuis un dossier contenant pyproject.toml et des tests
dagger call build-and-test --source=.

Sortie en cas de succès :

Tests réussis :
============================= test session starts ==============================
collected 3 items
tests/test_main.py ...                                                   [100%]
============================== 3 passed in 0.42s ===============================

Sortie en cas d'échec :

Échec : exit code: 1 [traceparent:00-abc123...]

Debugger pas à pas

Scénario 1 : pip install échoue

# Lancer en mode interactif pour inspecter l'état
dagger call --interactive build-and-test --source=.

Quand l'erreur se produit, un terminal s'ouvre dans le conteneur. Vous pouvez :

# Vérifier les fichiers présents
ls -la /app

# Tester pip manuellement
pip install -e . 2>&1

# Inspecter le pyproject.toml
cat /app/pyproject.toml

Scénario 2 : pytest échoue

# Augmenter la verbosité pour voir les détails
dagger call -vv build-and-test --source=.

Ou ouvrir un terminal sur le conteneur après le build :

# D'abord, créer une fonction qui retourne le conteneur
dagger call build-container terminal

Puis dans le terminal :

# Lancer pytest manuellement pour voir les erreurs
cd /app && pytest -v --tb=long

Scénario 3 : comprendre ce qui est caché

# Voir TOUT ce que Dagger fait en interne
dagger call --debug build-and-test --source=.

Pourquoi sync() avant stdout() ?

Dans l'exemple, on appelle sync() après pip install, puis stdout() après pytest. Voici pourquoi :

Sans sync()	Avec sync()
Si pip échoue, l'erreur est mélangée avec pytest	L'erreur pip est isolée, on sait exactement où ça casse
Impossible de savoir quelle étape a planté	Debug ciblé : build OK → tests KO
Moins d'overhead (1 seul trigger)	Légèrement plus lent (2 triggers)

Règle : utilisez sync() entre les étapes logiques pour un debug précis.

Variante : récupérer stderr en cas d'échec

Pour avoir plus de détails sur l'erreur :

@function
async def build_and_test_verbose(self, source: dagger.Directory) -> str:
    """Version avec diagnostic détaillé."""
    container = (
        dag.container()
        .from_("python:3.12-slim")
        .with_directory("/app", source)
        .with_workdir("/app")
        .with_exec(["pip", "install", "-e", "."])
    )

    try:
        await container.sync()
    except dagger.ExecError:
        # Récupérer stderr pour comprendre l'échec pip
        err = await container.stderr()
        return f"pip install a échoué :\n{err}"

    test_container = container.with_exec(["pytest", "-v"])

    try:
        output = await test_container.stdout()
        return f"Tests OK :\n{output}"
    except dagger.ExecError:
        err = await test_container.stderr()
        return f"pytest a échoué :\n{err}"

À retenir

Lazy evaluation : dag.container()... ne s'exécute pas immédiatement
Triggers : stdout(), stderr(), sync(), export(), contents() forcent l'exécution
Async : toute fonction qui appelle un trigger doit être async def avec await
Erreurs : utilisez try/except dagger.ExecError pour capturer les échecs
Debug : dagger call --interactive ou terminal pour explorer les conteneurs
Verbosité : -v, -vv, -vvv, --debug pour plus d'informations

Prochaines étapes

Module 4 · Container API Maîtriser from_(), with_exec(), variables d'environnement

Module 2 · Installation Revenir aux bases de l'installation Dagger

Ressources complémentaires

Documentation officielle : docs.dagger.io
Observability : docs.dagger.io/features/observability
Python async/await : Documentation asyncio