Execution Pipeline System
Relevant source files
The following files were used as context for generating this wiki page:
The Execution Pipeline System is a declarative framework designed to manage complex, multi-stage tasks within the flext-cli environment. It utilizes a Directed Acyclic Graph (DAG) approach for stage execution, allowing for sophisticated dependency management, state sharing, and resilient execution patterns. The system ensures that stages are executed in the correct order based on their requirements while providing built-in support for retries, conditional skipping, and fail-fast semantics.
At its core, the system leverages graphlib.TopologicalSorter to resolve execution order and uses a shared context to propagate data between stages. This architecture allows developers to define isolated units of logic (stages) and compose them into robust workflows that can handle transient failures through configurable retry mechanisms.
Sources: src/flext_cli/_protocols/pipeline.py:1-10, src/flext_cli/_utilities/pipeline.py:11-20
Architecture and Core Components
The pipeline system is structured around three primary pillars: data models defining the schema, protocols defining the interfaces, and a service layer that provides a Domain Specific Language (DSL) for orchestration.
Pipeline Data Models
The system uses Pydantic-based models to ensure type safety and validation across the execution lifecycle. Key models include PipelineStageSpec for definitions and PipelineStageResult for execution outcomes.
| Model | Description | Key Fields |
|---|---|---|
PipelineStageContext | Carries shared state and settings between stages. | workspace_root, shared, settings |
PipelineStageSpec | Declarative definition of a single execution unit. | stage_id, depends_on, handler, retry |
PipelineStageResult | The output generated by a single stage. | status, output, duration_ms, error |
PipelineResult | Aggregated result of the entire pipeline run. | stages, total_duration_ms, success |
Sources: src/flext_cli/_models/pipeline.py:21-120
Structural Relationships
The following class diagram illustrates the relationship between the core pipeline components and how they interact to form an executable graph.
The diagram shows the lifecycle where a PipelineStageSpec utilizes a PipelineStageContext to produce a PipelineStageResult, which are eventually aggregated into a final PipelineResult.
Sources: src/flext_cli/_models/pipeline.py:21-120, src/flext_cli/_protocols/pipeline.py:15-35
Execution Flow and Dependency Resolution
The pipeline engine performs dependency resolution using topological sorting. Before execution, the system validates the graph to ensure there are no circular dependencies, which would result in a CycleError.
Stage Execution Logic
Each stage is executed within a supervised loop that handles:
- Skip Predicates: Evaluating
skip_ifto determine if a stage should be bypassed. - Execution: Invoking the provided
handler. - Error Handling: Catching
CLI_SAFE_EXCEPTIONSand managing retries. - State Propagation: Writing outputs to the
sharedmapping in the context.
This flow illustrates the execution loop, including the retry logic and the "fail-fast" mechanism that prevents execution of downstream stages after a failure.
Sources: src/flext_cli/_utilities/pipeline.py:23-134, tests/unit/_cases/test_pipeline/testsflextclipipeline_part_02.py:31-45
Public DSL and Orchestration
The FlextCliPipeline service provides a high-level API (DSL) for defining pipelines. It supports both complex DAGs and simple linear chains.
Linear Pipelines
A common pattern is the linear_pipeline, which automatically sets dependencies so that each stage depends on the immediately preceding one in the provided order.
# Example of building a linear pipeline via the DSL
stages = cli.linear_pipeline(
stage_order=("init", "build", "deploy"),
handlers={
"init": init_handler,
"build": build_handler,
"deploy": deploy_handler,
}
)
Sources: src/flext_cli/services/pipeline.py:79-106, tests/unit/_cases/test_pipeline/testsflextclipipeline_part_01.py:64-79
Error Handling and Resilience
The system differentiates between standard failures and exceptions. If a handler raises a CLI_SAFE_EXCEPTION, the engine logs a warning and attempts a retry if configured. If all retries fail, the stage is marked as FAILED.
- Max Retries: Controlled by
c.Cli.PIPELINE_MAX_RETRY. - Fail Fast: When enabled, any stage failure immediately marks all subsequent stages as
SKIPPED.
Sources: src/flext_cli/_utilities/pipeline.py:104-134, tests/unit/_cases/test_pipeline/testsflextclipipeline_part_02.py:75-103
Summary of Execution States
Stages transition through various states during the pipeline lifecycle, defined in t.Cli.PipelineStageStatus.
| Status | Description |
|---|---|
OK | Stage completed successfully. |
FAILED | Stage handler returned a failure or exhausted retries after exceptions. |
SKIPPED | Stage was bypassed due to skip_if predicate or a prior failure in fail_fast mode. |
Sources: src/flext_cli/_models/pipeline.py:72-85, src/flext_cli/_utilities/pipeline.py:66-78
The Execution Pipeline System provides a structured approach to CLI task automation, ensuring that complex workflows remain maintainable, observable through duration tracking, and resilient via standardized error handling and state management.