Get Started with the Dagger Python SDK
Introduction
This tutorial teaches you the basics of using Dagger in Python. You will learn how to:
- Install the Python SDK
- Create a Python CI tool to test an application
- Improve the Python CI tool to test the application against multiple Python versions
Requirements
This tutorial assumes that:
- You have a basic understanding of the Python programming language. If not, read the Python tutorial.
- You have a Python development environment with Python 3.10 or later. If not, install Python.
- You have the Dagger CLI installed on the host system. If not, install the Dagger CLI.
- You have Docker installed and running on the host system. If not, install Docker.
- You have a Python application with tests defined and in a virtual environment.
This tutorial creates a CI tool to test your Python application against multiple Python versions. If you don't have a Python application already, clone an existing Python project with a well-defined test suite before proceeding. A good example is the FastAPI library, which you can clone as below:
git clone --branch 0.101.0 https://github.com/tiangolo/fastapi
The code samples in this tutorial are based on the above FastAPI project and tag. If using a different project, adjust the code samples accordingly.
Step 1: Install the Dagger Python SDK
The Dagger Python SDK requires Python 3.10 or later.
Create a virtual environment for your project (if you don't already have one). For example:
python3 -m venv .venv && source .venv/bin/activate
Install the Dagger Python SDK in your project's virtual environment using pip
:
pip install dagger-io
Step 2: Create a Dagger client in Python
Create a new file named test.py
and add the following code to it.
"""Execute a command."""
import sys
import anyio
import dagger
async def test():
async with dagger.Connection(dagger.Config(log_output=sys.stderr)) as client:
python = (
client.container()
# pull container
.from_("python:3.11-slim-buster")
# get Python version
.with_exec(["python", "-V"])
)
# execute
version = await python.stdout()
print(f"Hello from Dagger and {version}")
anyio.run(test)
This Python stub imports the Dagger SDK and defines an asynchronous function named test()
. This test()
function performs the following operations:
- It creates a Dagger client with
dagger.Connection()
. This client provides an interface for executing commands against the Dagger engine. The optionaldagger.Config(log_output=sys.stderr)
configuration displays the output from the Dagger engine. - It uses the client's
container().from_()
method to initialize a new container from a base image. In this example, the base image is thepython:3.10-slim-buster
image. This method returns aContainer
representing an OCI-compatible container image. - It uses the
Container.with_exec()
method to define the command to be executed in the container - in this case, the commandpython -V
, which returns the Python version string. Thewith_exec()
method returns a revisedContainer
with the results of command execution. - It retrieves the output stream of the last executed command with the
Container.stdout()
method and prints its contents.
Run the Python CI tool by executing the command below from the project directory:
dagger run python test.py
The dagger run
command executes the specified command in a Dagger session and displays live progress. The tool outputs a string similar to the one below.
Hello from Dagger and Python 3.11.1
Step 3: Test against a single Python version
Now that the basic structure of the CI tool is defined and functional, the next step is to flesh out its test()
function to actually test the Python application.
Replace the test.py
file from the previous step with the version below (highlighted lines indicate changes):
"""Run tests for a single Python version."""
import sys
import anyio
import dagger
async def test():
async with dagger.Connection(dagger.Config(log_output=sys.stderr)) as client:
# get reference to the local project
src = client.host().directory(".")
python = (
client.container().from_("python:3.10-slim-buster")
# mount cloned repository into image
.with_directory("/src", src)
# set current working directory for next commands
.with_workdir("/src")
# install test dependencies
.with_exec(["pip", "install", "-r", "requirements.txt"])
# run tests
.with_exec(["pytest", "tests"])
)
# execute
await python.sync()
print("Tests succeeded!")
anyio.run(test)
The revised test()
function now does the following:
- It creates a Dagger client with
dagger.Connection()
as before. - It uses the client's
host().directory(".")
method to obtain a reference to the current directory on the host. This reference is stored in thesrc
variable. - It uses the client's
container().from_()
method to initialize a new container from a base image. This base image is the Python version to be tested against - thepython:3.10-slim-buster
image. This method returns a newContainer
class with the results. - It uses the
Container.with_directory()
method to mount the host directory into the container at the/src
mount point. - It uses the
Container.with_workdir()
method to set the working directory in the container. - It chains
Container.with_exec()
methods to install test dependencies and run tests in the container. - It uses the
Container.sync()
method to execute the command.
The from_()
, with_directory()
, with_workdir()
and with_exec()
methods all return a Container
, making it easy to chain method calls together and create a pipeline that is easy and intuitive to understand.
Run the Python CI tool by executing the command below:
dagger run python test.py
The tool tests the application, logging its operations to the console as it works. If all tests pass, it displays the final output below:
Tests succeeded!
Step 4: Test against multiple Python versions
Now that the Python CI tool can test the application against a single Python version, the next step is to extend it for multiple Python versions.
Replace the test.py
file from the previous step with the version below (highlighted lines indicate changes):
"""Run tests for multiple Python versions."""
import sys
import anyio
import dagger
async def test():
versions = ["3.7", "3.8", "3.9", "3.10", "3.11"]
async with dagger.Connection(dagger.Config(log_output=sys.stderr)) as client:
# get reference to the local project
src = client.host().directory(".")
for version in versions:
python = (
client.container()
.from_(f"python:{version}-slim-buster")
# mount cloned repository into image
.with_directory("/src", src)
# set current working directory for next commands
.with_workdir("/src")
# install test dependencies
.with_exec(["pip", "install", "-r", "requirements.txt"])
# run tests
.with_exec(["pytest", "tests"])
)
print(f"Starting tests for Python {version}")
# execute
await python.sync()
print(f"Tests for Python {version} succeeded!")
print("All tasks have finished")
anyio.run(test)
This revision of the CI tool does much the same as before, except that it now supports multiple Python versions.
- It defines the test matrix, consisting of Python versions
3.7
to3.11
. - It iterates over this matrix, downloading a Python container image for each specified version and testing the source application in that version.
Run the CI tool by executing the command below:
dagger run python test.py
The tool tests the application against each version in sequence and displays the following final output:
Starting tests for Python 3.7
Tests for Python 3.7 succeeded!
Starting tests for Python 3.8
Tests for Python 3.8 succeeded!
Starting tests for Python 3.9
Tests for Python 3.9 succeeded!
Starting tests for Python 3.10
Tests for Python 3.10 succeeded!
Starting tests for Python 3.11
Tests for Python 3.11 succeeded!
All tasks have finished
One further improvement is to speed things up by having the tests run concurrently. Here's a revised test.py
which demonstrates how to do this (highlighted lines indicate changes):
"""Run tests for multiple Python versions concurrently."""
import sys
import anyio
import dagger
async def test():
versions = ["3.7", "3.8", "3.9", "3.10", "3.11"]
async with dagger.Connection(dagger.Config(log_output=sys.stderr)) as client:
# get reference to the local project
src = client.host().directory(".")
async def test_version(version: str):
python = (
client.container().from_(f"python:{version}-slim-buster")
# mount cloned repository into image
.with_directory("/src", src)
# set current working directory for next commands
.with_workdir("/src")
# install test dependencies
.with_exec(["pip", "install", "-r", "requirements.txt"])
# run tests
.with_exec(["pytest", "tests"])
)
print(f"Starting tests for Python {version}")
# execute
await python.sync()
print(f"Tests for Python {version} succeeded!")
# when this block exits, all tasks will be awaited (i.e., executed)
async with anyio.create_task_group() as tg:
for version in versions:
tg.start_soon(test_version, version)
print("All tasks have finished")
anyio.run(test)
Run the tool again by executing the command below:
dagger run python test.py
Now, the tool performs tests concurrently, with a noticeable difference in the total time required.
Conclusion
This tutorial introduced you to the Dagger Python SDK. It explained how to install the SDK and use it with a Python package. It also provided a working example of a Python CI tool powered by the SDK, demonstrating how to test an application against multiple Python versions in parallel.
Use the SDK Reference to learn more about the Dagger Python SDK.