Skip to content

topmark.pipeline.runner

Run the TopMark V2 header processing pipeline for a single file.

This module defines the HeaderProcessor protocol interface, a registry system for associating file extensions with processor implementations, and helper functions for processor lookup and registration. It enables extensible, comment-style-based header processing for different file types.

run 

run(ctx, steps, *, prune_views=True, keep_diff_view=False)

Execute the pipeline sequentially.

Parameters:

Name Type Description Default
ctx ProcessingContext

Mutable processing context.

 required
steps Sequence[Step[ProcessingContext]]

Ordered sequence of pipeline steps. Each step takes and returns a context.

 required
prune_views bool

Release consumed view payloads between steps once no remaining step declares them as consumed (default: True).

 True
keep_diff_view bool

Whether to preserve the diff view during between-step pruning (required when the pipeline generates a unified diff in PatcherStep).

 False

Returns:

Type Description
ProcessingContext

The final processing context after all steps have run.
Source code in src/topmark/pipeline/runner.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def run(
    ctx: ProcessingContext,
    steps: Sequence[Step[ProcessingContext]],
    *,
    prune_views: bool = True,
    keep_diff_view: bool = False,
) -> ProcessingContext:
    """Execute the pipeline sequentially.

    Args:
        ctx: Mutable processing context.
        steps: Ordered sequence of pipeline steps. Each step takes and returns a context.
        prune_views: Release consumed view payloads between steps once no remaining
            step declares them as consumed (default: `True`).
        keep_diff_view: Whether to preserve the diff view during between-step pruning
            (required when the pipeline generates a unified diff in
            [`PatcherStep`][topmark.pipeline.steps.patcher.PatcherStep]).

    Returns:
        The final processing context after all steps have run.
    """
    step_count: int = len(steps)
    for index, step in enumerate(steps):
        ctx = step(ctx)
        if prune_views is True:
            remaining_view_consumers: set[ViewSlot] = set()
            for remaining_step in steps[index + 1 : step_count]:
                remaining_view_consumers.update(remaining_step.consumes_views)
            ctx.views.release_consumed(
                remaining_view_consumers=remaining_view_consumers,
                keep_diff_view=keep_diff_view,
            )

    return ctx