Highway DSL v2.0.0

from highway_dsl import WorkflowBuilder

workflow = (WorkflowBuilder("my_workflow", version="2.0.0")
    .task("task_1", "module.function", args=["arg1"])
    .task("task_2", "module.function2", dependencies=["task_1"])
    .build()
)

# Serialize to YAML
yaml_output = workflow.to_yaml()

# Serialize to JSON
json_output = workflow.to_json()

from highway_dsl import TaskOperator, RetryPolicy, TimeoutPolicy
from datetime import timedelta

task = TaskOperator(
    task_id="process_data",
    function="data.processor.transform",
    args=["{{input_file}}"],
    kwargs={"mode": "batch"},
    result_key="processed_data",
    retry_policy=RetryPolicy(
        max_retries=3,
        delay=timedelta(seconds=30),
        backoff_factor=2.0
    ),
    timeout_policy=TimeoutPolicy(
        timeout=timedelta(minutes=10),
        kill_on_timeout=True
    )
)

builder.condition(
    "data_quality_check",
    condition="{{quality_score}} > 0.95",
    if_true=lambda b: b.task("advanced_process", "proc.advanced"),
    if_false=lambda b: b.task("basic_process", "proc.basic")
)

builder.parallel(
    "process_all_data",
    branches={
        "images": lambda b: b.task("process_img", "img.process"),
        "videos": lambda b: b.task("process_vid", "vid.process"),
        "audio": lambda b: b.task("process_aud", "aud.process")
    }
)

builder.foreach(
    "process_items",
    items="{{data.records}}",
    loop_body=lambda b: (
        b.task("validate", "validate.record", args=["{{item}}"])
         .task("transform", "transform.record", args=["{{item}}"])
         .task("store", "store.record", args=["{{item}}"])
    )
)

builder.while_loop(
    "retry_until_success",
    condition="{{result.status}} == 'failed'",
    loop_body=lambda b: (
        b.task("attempt", "processor.attempt")
         .task("check", "processor.check_result", result_key="result")
    )
)

# Branch 1: Emit event after processing
builder.task("process", "processor.run")
       .emit_event("emit_ready", event_name="processing_complete")

# Branch 2: Wait for event before continuing
builder.wait_for_event(
    "wait_ready",
    event_name="processing_complete",
    timeout=timedelta(hours=1)
)
       .task("downstream", "processor.downstream")

builder.switch(
    "route_by_priority",
    switch_on="{{task.priority}}",
    cases={
        "critical": "handle_critical",
        "high": "handle_high",
        "medium": "handle_medium",
        "low": "handle_low"
    },
    default="handle_default"
)

from highway_dsl import JoinMode

# Wait for all tasks to complete (regardless of success/failure)
builder.join(
    "wait_all",
    join_on=["task_1", "task_2", "task_3"],
    join_mode=JoinMode.ALL_OF
)

# Wait for all tasks to succeed (fail if any fails)
builder.join(
    "wait_all_success",
    join_on=["task_1", "task_2", "task_3"],
    join_mode=JoinMode.ALL_SUCCESS
)

# Wait for any one task to complete
builder.join(
    "wait_any",
    join_on=["task_1", "task_2", "task_3"],
    join_mode=JoinMode.ANY_OF
)

# Wait for first successful completion
builder.join(
    "wait_first_success",
    join_on=["task_1", "task_2", "task_3"],
    join_mode=JoinMode.ONE_SUCCESS
)

from highway_dsl import WorkflowBuilder
from datetime import timedelta

workflow = (
    WorkflowBuilder("data_pipeline", version="2.0.0")

    # Extract data from multiple sources in parallel
    .parallel(
        "extract_data",
        branches={
            "database": lambda b: b.task("extract_db", "extract.from_db"),
            "api": lambda b: b.task("extract_api", "extract.from_api"),
            "files": lambda b: b.task("extract_files", "extract.from_files")
        }
    )

    # Wait for all extractions to complete successfully
    .join(
        "wait_extractions",
        join_on=["extract_db", "extract_api", "extract_files"],
        join_mode="ALL_SUCCESS"
    )

    # Merge and transform
    .task("merge", "transform.merge_sources", result_key="merged_data")
    .task("validate", "transform.validate", args=["{{merged_data}}"])

    # Branch based on data quality
    .condition(
        "quality_check",
        condition="{{validate.score}} >= 0.95",
        if_true=lambda b: b.task("load_prod", "load.to_production"),
        if_false=lambda b: b.task("load_staging", "load.to_staging")
    )

    .build()
)

workflow = (
    WorkflowBuilder("event_driven", version="2.0.0")

    .parallel(
        "process_branches",
        branches={
            "producer": lambda b: (
                b.task("generate", "producer.generate_data")
                 .emit_event("notify", event_name="data_ready")
            ),
            "consumer": lambda b: (
                b.wait_for_event("wait", event_name="data_ready",
                                 timeout=timedelta(minutes=5))
                 .task("process", "consumer.process_data")
            )
        }
    )

    .build()
)

from highway_dsl import Duration

# String format: "P[n]Y[n]M[n]DT[n]H[n]M[n]S"
Duration("PT30S")      # 30 seconds
Duration("PT5M")       # 5 minutes
Duration("PT2H")       # 2 hours
Duration("P1D")        # 1 day
Duration("PT1H30M")    # 1 hour 30 minutes

# Programmatic (timedelta)
from datetime import timedelta
Duration(timedelta(seconds=30))

# Define callback tasks that execute on success/failure
builder.task("risky_task", "processor.risky")
       .on_success("notify_success")
       .on_failure("notify_failure")

# Define the callback handlers
builder.task("notify_success", "notify.send_success")
builder.task("notify_failure", "notify.send_failure")

from datetime import datetime

workflow = (
    WorkflowBuilder("scheduled_workflow")
    .set_schedule("0 */6 * * *")        # Cron: every 6 hours
    .set_start_date(datetime(2025, 1, 1))
    .set_catchup(False)                    # Don't backfill
    .add_tags("production", "critical")

    # ... define tasks ...

    .build()
)