Control Flow: Conditional Routing with `.route()`

While .next() defines a single path and .branch() defines concurrent paths, the .route() method defines conditional execution paths. It allows the pipeline to select one step out of several options based on the result of a provided decision function.

A Route is a special Feeder that takes the output of the preceding step and passes it to a selector function. This function returns an integer index, which Chorey uses to choose one step to execute from a list of defined choices.

Route Components

The .route() method requires three main components:

*choices: Step: A variable number of possible next steps. These are the pipeline branches the route can choose from. They may output different types, and the overall output type of the route will be a fully typed union of all possible outputs.
selector: Callable[[Input], int]: A function that takes the output of the preceding step and returns a zero-based index corresponding to the chosen step in choices.
decision_label: str | None: An optional label used in the Mermaid visualization to describe the decision point.

Conditional Routing Example

from chorey import step
from dataclasses import dataclass

@dataclass
class Document:
    id: int
    content: str

@dataclass
class Summary:
    document_id: int
    summary_text: str

@dataclass
class FullAnalysis:
    document_id: int
    word_count: int
    sentiment: str

async def fetch_document(doc_id: int) -> Document:
    # Simulate fetching a document from a database or API
    return Document(id=doc_id, content="This is a sample document content.")

async def summarize_document(doc: Document) -> Summary:
    # Simulate summarizing the document
    return Summary(document_id=doc.id, summary_text="This is a summary.")

async def analyze_document(doc: Document) -> FullAnalysis:
    # Simulate performing a full analysis of the document
    return FullAnalysis(document_id=doc.id, word_count=len(doc.content.split()), sentiment="Positive")

async def store_processing_result(result: Summary | FullAnalysis) -> None:
    # Simulate storing the result
    print(f"Storing result: {result}")
    print(f"Type of result: {type(result)}")

def selector(doc: Document) -> int:
    # Route based on document length
    return 0 if len(doc.content) < 50 else 1

# Define the pipeline with conditional routing
pipeline = (
    step(fetch_document)
    .route(
        summarize_document,
        analyze_document,
        selector=selector,
        decision_label="Document Length Check"
    )
    .next(store_processing_result)
)

Type Safety: The Union Output Type

A critical advantage of Chorey's Route mechanism is that it correctly handles steps with dissimilar output types. When a Route is defined with different possible outputs, the resulting object's final output type becomes a Union of all those potential types. This behavior preserves strict type safety for the rest of the pipeline, but it introduces an important requirement for the subsequent step.

In the example above, the Route can output either a Summary or a FullAnalysis. Therefore, the next step, store_processing_result, must accept an argument of type Summary | FullAnalysis. This ensures that the pipeline remains type-safe and that any mismatches are caught at development time. If the subsequent step expects a different type (or not all possible types), a type checker will raise an error.

Selector Function

The selector function is a simple sync or async callable that takes the output of the previous step and returns an integer index. This index corresponds to the position of the desired step in the choices list. The function must ensure that the returned index is valid (i.e., within the range of available choices) to avoid runtime errors.

When to Use `.route()`

Use .route() when you need to switch between entirely different processing logic based on data or configuration.

Use Case	Description
Data Triage	Route data to different processing steps based on its characteristics (e.g., size, type, source).
Feature Flags	Enable or disable certain processing paths based on configuration flags or environment settings. More about handling context throughout a pipeline here

Control Flow: Conditional Routing with .route()