ttasks

A small Python task ledger, executor, and DAG workflow library.

ttasks models work as Task objects, executes them through a configurable TaskExecutor, persists them through an optional Store, and runs dependency graphs with TaskGraph.

Requirements

• Python 3.12+
• uv for the development workflow used by this repository
• GitHub Copilot authentication for TaskType.PROMPT and TaskType.AGENT tasks

Quick start

fromttasksimportTask, TaskExecutorexecutor=TaskExecutor()
task=Task.bash("echo hello", title="Say hello")
result=executor.execute(task)
asserttask.is_doneassertresult.output=="hello\n"asserttask.resultisresult

Subscribe to executor events to observe lifecycle changes, progress, and streamed subprocess output:

fromttasksimportTask, TaskEvent, TaskExecutordefprint_event(event: TaskEvent) ->None:
print(f"{event.type.value}: {event.task.title} -> {event.status.value}")
executor=TaskExecutor()
unsubscribe=executor.events.subscribe(print_event)
try:
executor.execute(Task.bash("echo hello", title="Say hello"))
finally:
unsubscribe()

Use a store when task and graph state should be inspectable or durable:

frompathlibimportPathfromttasksimportSQLiteStore, Task, TaskExecutor, TaskGraphstore=SQLiteStore(Path("ttasks.db")) # use InMemoryStore() for testsexecutor=TaskExecutor(store=store)
build=Task.bash("echo build", title="Build")
test=Task.bash("echo test", title="Test")
graph=TaskGraph(title="stored pipeline")
graph.add(build)
graph.add(test, after=[build])
graph.run(executor)
assertstore.tasks[build.id].status==build.statusassertstore.graphs[graph.id].okisTrue

For a fuller runnable example, see main.py.

What it provides

• Task and TaskResult domain objects for tracking work and outcomes.
• TaskExecutor for running Bash, PowerShell, Copilot prompt, Copilot agent, or custom handler tasks.
• Event streams for lifecycle, progress, and subprocess output updates.
• TaskGraph for dependency-ordered DAG workflows with finally tasks.
• In-memory and SQLite stores for task and graph persistence.
• Async single-task submission, graceful shutdown, cancellation, timeouts, and opt-in single-task retries.

Documentation

User docs are published with the generated API reference on GitHub Pages:

• Quickstart
• Tutorials
• Patterns
• API reference

Build the docs locally:

uv run mkdocs build --strict --site-dir site
uv run pdoc ttasks --output-directory site/api

Development

Run the project checks:

uv run ruff check .
uv run ty check
uv run pytest

Run the full suite including live tests:

uv run pytest -o addopts='' --cov=ttasks --cov-report=term-missing --cov-fail-under=100