"""This will trigger another flow."""
from metaflow.decorators import step
from metaflow.flowspec import FlowSpec
from metaflow.parameters import Parameter


class TriggerFlow(FlowSpec):
    @step
    def start(self):
        print("triggered step")
        self.output = "output from TriggerFlow"
        self.next(self.join)

    @step
    def join(self):
        self.next(self.end)

    @step
    def end(self):
        print("end")


if __name__ == "__main__":
    TriggerFlow()

"""Triggered by TriggerFlow."""
from metaflow import trigger_on_finish, current
from metaflow.decorators import step
from metaflow.flowspec import FlowSpec
from metaflow.parameters import Parameter


@trigger_on_finish(flow="TriggerFlow")
class SkipFlow(FlowSpec):
    @step
    def start(self):
        print(current.trigger.run.data.output)
        self.next(self.middle)

    @step
    def middle(self):
        print("middle step")
        self.next(self.join)

    @step
    def join(self):
        self.next(self.end)

    @step
    def end(self):
        print("end")


if __name__ == "__main__":
    SkipFlow()