Quickstart

Goal: a tiny step where you transform rows and framework writes them. Three files are enough — manifest, thin hooks, processor. Full mental model: Core concepts.

Ship-with-framework reference: examples/hello-step/ in the Framework repo (pytest examples/hello-step). Validate any manifest with drtoller validate path/to/step.drtml --compile. Bootstrap: drtoller init my-pipeline — see AI quickstart.

Minimal package

acme_steps/uppercase/
  uppercase.drtml
  hooks.py
  processor.py

Manifest

The important ideas in the YAML below:

• run.type: partition — feed → processor loop;
• processor.batch — your function;
• params.runtime.* — batching / metrics knobs (framework reads these, not domain.*).

drtml: 4
step: uppercase
run:
  type: partition
  processor:
    batch: acme_steps.uppercase.processor:process_batch
  derive:
    document_registry:
      from: documents.document_id
      operation: rows
orchestration:
  artifacts_root: /tmp/drtoller-data
params:
  runtime:
    document_batch_size: 32
    metrics_enabled: true
    resume_enabled: false
    max_docs: 0
    progress_log_every_n_docs: 100
    finalize_flush_every_n_docs: 0
    corpus_batch_max_chars: 0
    pipeline:
      read_q_max: 4
      write_q_max: 4
      lane_q_max: 4
  mapping:
    upstream_runs: source-run-id
storage:
  default: artifacts
  stores:
    artifacts: {backend: local, root: /tmp/drtoller-data}
inputs:
  documents:
    schema: {document_id: string, text: string}
    run: mapping.upstream_runs
    feed: {batch: 256}
outputs:
  uppercase_documents:
    schema: {run_id: string, document_id: string, text: string}
    from: documents
  document_registry:
    kind: registry
    key: document_id
    schema: {run_id: string, document_id: string}

Processor

from typing import Any, Mapping
from drtoller.framework.processing.process_result import ProcessResult
from drtoller.framework.storage.feed import ProcessorBatch


def process_batch(
    raws: list[Any], *, ctx, params: Mapping[str, Any],
    processor_ctx: Mapping[str, Any], partition_id: str,
) -> list[ProcessResult]:
    del params, processor_ctx, partition_id
    results = []
    for batch in raws:
        if not isinstance(batch, ProcessorBatch):
            raise TypeError("expected ProcessorBatch")
        rows = [
            {
                "run_id": ctx.run_id,
                "document_id": str(row["document_id"]),
                "text": str(row.get("text") or "").upper(),
            }
            for row in batch.primary_rows
        ]
        results.append(ProcessResult(ok=True, rows={"documents": rows}, shard_id=batch.shard_id))
    return results

documents is a logical output key. DRTML maps it to physical dataset uppercase_documents via from:.

Hook

from drtoller.framework.drtml.resolve import manifest_path_adjacent
from drtoller.framework.processing.partition_loop import run_partition_loop


def run(ctx, *, manifest_path=None):
    path = manifest_path or manifest_path_adjacent(__file__, step_id="uppercase")
    return run_partition_loop(ctx, manifest_path=path)

What the framework does

1. loads and validates the v4 manifest;
2. opens storage from storage.stores;
3. feeds primary input rows according to inputs.*.feed;
4. calls the processor and writes logical outputs;
5. derives registry rows when run.derive is present;
6. emits metrics/UI/orchestration metadata from the same contract.