Dagger is a CI/CD focused container scripting engine.
Using the Python/Go/TypeScript SDKs, Dagger can:
- Define container images (like
Dockerfiles) - Run containers (like
docker runanddocker-compose)
Plus everything Python/Go/TypeScript can do.
Dagger init- Convert
SpokaneTech_PyDockerfile in to a Dagger function.- Base Container
- Dev
- Prod
- Demo
functionsterminalservice/up
- Ruff/Bandit Checks
- When to use async/await
- Third party module
Initalize a new Dagger module.
dagger init --name=spokane-tech --sdk=pythonSpokaneTech_Py/dagger
├── pyproject.toml
├── sdk
│ └── ...
└── src
└── main.py
The contents of dagger/sdk is generated & only needed for auto-complete.
- Re-generate using
dagger develop - Use via
pip install -e dagger/sdk(in a virtual environment)
Demo #1
import dagger
from dagger import dag, function, object_type
PYTHON_VERSION = "3.11-slim-bullseye"
GUNICORN_CMD = ["gunicorn", ...]
@object_type
class SpokaneTech:
src: dagger.Directory
req: dagger.File
@function
def base_container(self) -> dagger.Container:
return dag.container()
@function
def dev(self) -> dagger.Container:
return self.base_container()
@function
def prod(self) -> dagger.Container:
return self.base_container()PYTHON_VERSION = "3.11-slim-bullseye"
@object_type
class SpokaneTech:
src: dagger.Directory
req: dagger.File
@function
def base_container(self) -> dagger.Container:
return (
dag.container()
.from_(f"python:{PYTHON_VERSION}")
.with_env_variable("PYTHONDONTWRITEBYTECODE", "1")
.with_env_variable("PYTHONUNBUFFERED", "1")
.with_exposed_port(8000)
.with_file("/tmp/requirements.txt", self.req)
.with_exec(["pip", "install", "--upgrade", "pip"])
.with_exec(["pip", "install", "-r", "/tmp/requirements.txt"])
.with_exec(["rm", "-rf", "/root/.cache"])
.with_directory("/code/src", self.src)
.with_workdir("/code")
)GUNICORN_CMD = ["gunicorn", ...]
@object_type
class SpokaneTech:
src: dagger.Directory
req: dagger.File
@function
def dev(self, run: bool = False) -> dagger.Container:
ctr = (
self.base_container()
.with_env_variable("SPOKANE_TECH_DEV", "true")
.with_exec(["python", "src/manage.py", "migrate"])
)
if run:
ctr = ctr.with_exec(["python", "src/manage.py", "runserver", "0.0.0.0:8000"])
return ctr
@function
def prod(self) -> dagger.Container:
return self.base_container().with_exec(GUNICORN_CMD)Demo #2
Helper function run_linter:
- Want to capture stderr/stdout
- Still want the overall return code to be non-zero on failure (for CI)
- No
@functiondecorator - not part of CLI API
@object_type
class SpokaneTech:
src: dagger.Directory
req: dagger.File
async def run_linter(self, cmd: str, ctr: dagger.Container) -> str:
lint = ctr.with_exec(["bash", "-c", f"{cmd} &> results; echo -n $? > exitcode"])
async with TaskGroup() as tg:
code = tg.create_task(lint.file("exitcode").contents())
result = tg.create_task(lint.file("results").contents())
exit_code = int(code.result())
result = result.result()
if exit_code != 0:
print(result)
sys.exit(exit_code)
return result@object_type
class SpokaneTech:
src: dagger.Directory
req: dagger.File
@function
async def lint(self) -> str:
ctr = self.base_container().with_exec(["pip", "install", "ruff"])
return await self.run_linter("ruff check", ctr)
@function
async def format(self) -> str:
ctr = self.base_container().with_exec(["pip", "install", "ruff"])
return await self.run_linter("ruff format --check", ctr)
@function
async def bandit(self, pyproject: dagger.File) -> str:
ctr = self.base_container().with_exec(["pip", "install", "bandit"]).with_file("pyproject.toml", pyproject)
return await self.run_linter("bandit -c pyproject.toml -r src", ctr)
@function
async def test(self, pyproject: dagger.File) -> str:
ctr = self.base_container().with_file("pyproject.toml", pyproject).with_workdir("src/")
return await self.run_linter("pytest", ctr)Async & Await now?
- Regular function chains build a DAG - in the form of a GraphQL query.
- Async/Await calls evaluate the DAG - by executing the query.
Demo #3
- Dagger modules can use other Dagger modules
- Dagger modules are language agnostic
- Third Party modules can be browsed at daggerverse.dev
Example:
Use https://daggerverse.dev/mod/github.com/gerhard/daggerverse/notify to
send a Discord message when a linter fails.
Install the module
- Only at the project level, no global installs
dagger install github.com/gerhard/daggerverse/notify@v0.2.0
Adds a field to dagger.json & automatically (re-)generates the SDK auto-complete code.
{
"name": "spokane-tech",
"sdk": "python",
+ "dependencies": [
+ {
+ "name": "notify",
+ "source": "github.com/gerhard/daggerverse/notify@4ad0f0317fd57c41001b48c5a9a3a49d39e43210"
+ }
+ ],
"source": "dagger",
"engineVersion": "v0.10.2"
}Installed modules are attached to the dag object.
@object_type
class SpokaneTech:
src: dagger.Directory
req: dagger.File
+ webhook: dagger.Secret | None = None
async def run_linter(self, cmd: str, ctr: dagger.Container) -> str:
lint = ctr.with_exec(["bash", "-c", f"{cmd} &> results; echo -n $? > exitcode"])
async with TaskGroup() as tg:
code = tg.create_task(lint.file("exitcode").contents())
result = tg.create_task(lint.file("results").contents())
exit_code = int(code.result())
result = result.result()
if exit_code != 0:
+ if self.webhook is not None:
+ await dag.notify().discord(self.webhook, f"Check `{cmd}` failed. \n ```{result}```")
print(result)
sys.exit(exit_code)
return result
Presented using https://github.com/maaslalani/slides