CI/CD

Dagger Python SDK: Writing Testable CI Pipelines with Python

HelpMeTest

HelpMeTest

— 5 min read
Dagger Python SDK: Writing Testable CI Pipelines with Python

The Dagger Python SDK lets you write CI pipelines as ordinary Python functions. No YAML, no DSL — just Python. You can import Dagger functions like any library, test them with pytest, and run them locally with the same command that runs in CI.

This guide focuses on practical patterns: running pytest suites, connecting services, caching aggressively, and structuring your pipeline for testability.

Installation

pip install dagger-io

# Initialize a new Dagger module in your project
<span class="hljs-built_in">cd myproject
dagger init --sdk python --name my-pipeline

This creates:

myproject/
  dagger/
    src/
      main.py        # Your pipeline functions
    pyproject.toml
  dagger.json        # Module config

Your First Testing Function

# dagger/src/main.py
import dagger
from dagger import dag, function, object_type

@object_type
class MyPipeline:
    
    @function
    async def test(
        self,
        source: dagger.Directory,
        python_version: str = "3.12",
    ) -> str:
        """Run pytest. Returns test output."""
        return await (
            dag.container()
            .from_(f"python:{python_version}-slim")
            .with_directory("/src", source, exclude=[
                ".venv",
                "__pycache__",
                ".pytest_cache",
                "*.egg-info",
            ])
            .with_workdir("/src")
            .with_exec(["pip", "install", "--quiet", "--no-cache-dir", "-e", ".[dev]"])
            .with_exec([
                "pytest",
                "--tb=short",
                "--no-header",
                "-q",
                "tests/",
            ])
            .stdout()
        )

Run it:

dagger call test --<span class="hljs-built_in">source .

<span class="hljs-comment"># With specific Python version
dagger call <span class="hljs-built_in">test --<span class="hljs-built_in">source . --python-version 3.11

Caching Dependencies Properly

The most important optimization: cache pip downloads between runs.

@function
async def test(self, source: dagger.Directory) -> str:
    pip_cache = dag.cache_volume("pip-downloads")
    venv_cache = dag.cache_volume("venv-py312")
    
    return await (
        dag.container()
        .from_("python:3.12-slim")
        # Cache the pip download cache
        .with_mounted_cache("/root/.cache/pip", pip_cache)
        # Cache the virtual environment
        .with_mounted_cache("/venv", venv_cache)
        .with_env_variable("VIRTUAL_ENV", "/venv")
        .with_env_variable("PATH", "/venv/bin:$PATH")
        .with_directory("/src", source)
        .with_workdir("/src")
        .with_exec(["python", "-m", "venv", "/venv"])
        .with_exec(["pip", "install", "--quiet", "-e", ".[dev]"])
        .with_exec(["pytest", "-q", "tests/"])
        .stdout()
    )

With Dagger Cloud, these caches are shared across CI runners. Your pip install runs once per pyproject.toml change.

Integration Tests with Services

Dagger's .as_service() starts a container and makes it network-accessible to other containers:

@function
async def integration_test(
    self,
    source: dagger.Directory,
    postgres_password: dagger.Secret,
) -> str:
    """Run integration tests with a real PostgreSQL database."""
    
    # Start PostgreSQL as a sidecar service
    postgres = (
        dag.container()
        .from_("postgres:16-alpine")
        .with_env_variable("POSTGRES_USER", "testuser")
        .with_secret_variable("POSTGRES_PASSWORD", postgres_password)
        .with_env_variable("POSTGRES_DB", "testdb")
        .with_exposed_port(5432)
        .as_service()
    )
    
    return await (
        dag.container()
        .from_("python:3.12-slim")
        # Bind the postgres service — accessible at "postgres:5432"
        .with_service_binding("postgres", postgres)
        .with_env_variable(
            "DATABASE_URL",
            "postgresql://testuser@postgres:5432/testdb"
        )
        .with_secret_variable("POSTGRES_PASSWORD", postgres_password)
        .with_directory("/src", source)
        .with_workdir("/src")
        .with_exec(["pip", "install", "-e", ".[dev]"])
        .with_exec(["pytest", "tests/integration/", "-v", "--tb=long"])
        .stdout()
    )

You can add Redis, RabbitMQ, or any service the same way:

redis = (
    dag.container()
    .from_("redis:7-alpine")
    .with_exposed_port(6379)
    .as_service()
)

container = container.with_service_binding("redis", redis)

Parallel Test Execution

Run lint, unit tests, and type checking simultaneously:

import asyncio

@function
async def ci(self, source: dagger.Directory) -> str:
    """Full CI suite: lint + typecheck + unit + integration (parallel where possible)."""
    
    # Lint and typecheck in parallel
    lint_task = self._lint(source)
    type_task = self._typecheck(source)
    unit_task = self._unit_test(source)
    
    results = await asyncio.gather(
        lint_task,
        type_task,
        unit_task,
        return_exceptions=True,
    )
    
    failed = []
    for i, (name, result) in enumerate(zip(["lint", "typecheck", "unit"], results)):
        if isinstance(result, Exception):
            failed.append(f"{name}: {result}")
    
    if failed:
        raise Exception("CI failed:\n" + "\n".join(failed))
    
    # Integration tests run after unit tests pass
    await self._integration_test(source)
    
    return "All checks passed"

async def _lint(self, source: dagger.Directory) -> str:
    return await (
        dag.container()
        .from_("python:3.12-slim")
        .with_exec(["pip", "install", "ruff"])
        .with_directory("/src", source)
        .with_workdir("/src")
        .with_exec(["ruff", "check", "--output-format=github", "."])
        .stdout()
    )

async def _typecheck(self, source: dagger.Directory) -> str:
    return await (
        dag.container()
        .from_("python:3.12-slim")
        .with_exec(["pip", "install", "mypy"])
        .with_directory("/src", source)
        .with_workdir("/src")
        .with_exec(["pip", "install", "-e", "."])
        .with_exec(["mypy", "src/", "--ignore-missing-imports"])
        .stdout()
    )

Testing Your Dagger Functions with pytest

Since Dagger functions are just Python async functions, you can test them directly:

# tests/test_pipeline.py
import pytest
import dagger
from main import MyPipeline

@pytest.mark.anyio
async def test_lint_passes_on_valid_code():
    pipeline = MyPipeline()
    
    async with dagger.Connection() as client:
        source = client.host().directory(".", exclude=["tests/", ".venv"])
        result = await pipeline._lint(source)
        assert "error" not in result.lower()

@pytest.mark.anyio
async def test_unit_tests_pass():
    pipeline = MyPipeline()
    
    async with dagger.Connection() as client:
        source = client.host().directory(".")
        result = await pipeline.test(source)
        assert "failed" not in result
        assert "passed" in result

This is the key advantage over YAML: you can write tests for your CI pipeline itself.

Matrix Testing

Test multiple Python versions without YAML gymnastics:

@function
async def matrix_test(
    self,
    source: dagger.Directory,
    versions: list[str] | None = None,
) -> str:
    """Run tests against multiple Python versions."""
    if versions is None:
        versions = ["3.10", "3.11", "3.12"]
    
    tasks = [
        dag.container()
        .from_(f"python:{v}-slim")
        .with_directory("/src", source)
        .with_workdir("/src")
        .with_exec(["pip", "install", "-e", ".[dev]"])
        .with_exec(["pytest", "-q", "--tb=line"])
        .stdout()
        for v in versions
    ]
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    output = []
    for version, result in zip(versions, results):
        if isinstance(result, Exception):
            output.append(f"Python {version}: FAILED — {result}")
        else:
            output.append(f"Python {version}: PASSED")
    
    return "\n".join(output)

Secrets Handling

Never hardcode credentials. Use Dagger secrets:

@function
async def test_with_secrets(
    self,
    source: dagger.Directory,
    api_key: dagger.Secret,        # Injected securely
    db_password: dagger.Secret,
) -> str:
    return await (
        dag.container()
        .from_("python:3.12-slim")
        .with_secret_variable("API_KEY", api_key)
        .with_secret_variable("DB_PASSWORD", db_password)
        .with_directory("/src", source)
        .with_workdir("/src")
        .with_exec(["pip", "install", "-e", ".[dev]"])
        .with_exec(["pytest", "tests/", "-k", "integration"])
        .stdout()
    )

Pass from CLI:

dagger call test-with-secrets \
  --source . \
  --api-key <span class="hljs-built_in">env:API_KEY \
  --db-password <span class="hljs-built_in">env:DB_PASSWORD

Or from GitHub Actions secrets:

- run: dagger call test-with-secrets --source . --api-key env:API_KEY
  env:
    API_KEY: ${{ secrets.API_KEY }}

Coverage Reports as Artifacts

Return test artifacts from Dagger functions:

@function
async def test_with_coverage(self, source: dagger.Directory) -> dagger.Directory:
    """Run tests with coverage. Returns a directory with HTML report."""
    return await (
        dag.container()
        .from_("python:3.12-slim")
        .with_directory("/src", source)
        .with_workdir("/src")
        .with_exec(["pip", "install", "-e", ".[dev]", "pytest-cov"])
        .with_exec([
            "pytest",
            "--cov=src",
            "--cov-report=html:coverage-html",
            "--cov-report=xml:coverage.xml",
            "--cov-fail-under=80",
        ])
        .directory("coverage-html")
    )

Use from CLI:

# Export coverage report to ./coverage-output/
dagger call test-with-coverage --<span class="hljs-built_in">source . <span class="hljs-built_in">export --path ./coverage-output

Common Patterns

Fail Fast on Lint Before Running Tests

@function
async def ci(self, source: dagger.Directory) -> str:
    # Lint first — fastest feedback, no point running tests on broken code
    await self._lint(source)
    
    # Then run tests
    return await self.test(source)

Share a Built Base Image

def _base_image(self, source: dagger.Directory) -> dagger.Container:
    """Build once, use for multiple steps."""
    return (
        dag.container()
        .from_("python:3.12-slim")
        .with_directory("/src", source)
        .with_workdir("/src")
        .with_exec(["pip", "install", "-e", ".[dev]"])
    )

@function
async def test(self, source: dagger.Directory) -> str:
    return await self._base_image(source).with_exec(["pytest"]).stdout()

@function
async def lint(self, source: dagger.Directory) -> str:
    return await self._base_image(source).with_exec(["ruff", "check", "."]).stdout()

Dagger caches the base image layer — both functions reuse the pip install result without re-running it.

Using HelpMeTest for E2E in Dagger

For E2E browser tests, HelpMeTest provides a CLI that triggers cloud-hosted Playwright tests:

@function
async def e2e(
    self,
    source: dagger.Directory,
    app_url: str,
    token: dagger.Secret,
) -> str:
    """Trigger HelpMeTest E2E suite and wait for results."""
    return await (
        dag.container()
        .from_("node:22-alpine")
        .with_exec(["npm", "install", "-g", "helpmetest"])
        .with_secret_variable("HELPMETEST_TOKEN", token)
        .with_env_variable("APP_URL", app_url)
        .with_exec([
            "helpmetest", "run",
            "--project", "my-app",
            "--base-url", app_url,
            "--wait",
            "--timeout", "300",
        ])
        .stdout()
    )

This keeps your Dagger container lean while running full browser tests in HelpMeTest's infrastructure.

Summary

The Dagger Python SDK changes how you write CI:

  • Local parity: dagger call test --source . works identically in your terminal and in GitHub Actions
  • Caching: dependency installs cached at the volume level, shared across runners with Dagger Cloud
  • Services: start Postgres/Redis/anything as sidecars with .as_service()
  • Parallelism: asyncio.gather() runs steps concurrently without YAML job dependencies
  • Testability: write pytest tests for your Dagger functions themselves

Start with converting your longest-running CI step. The local execution alone is worth the migration.

Read more