from metaflow import FlowSpec, Parameter, step

class Mixin:    
    # defining this as a class attribute lets us access it without creating an instance
    method_list = [
        "method1",
        "method2",
    ]
        
    def method1(self):
        print(self.arg1 + 1)
    
    def method2(self):
        print(self.arg1 + 2)
        
class DynamicFlowBase:
    @classmethod
    def wrap_method(cls, method_str, next_method_str):
        method = getattr(cls, method_str)
        new_name = "wrapped_" + method_str
        
        @step
        def wrapped_method(self):
            method(self)
            next_step = getattr(self, next_method_str)            
            self.next(next_step)
        
        wrapped_method.__name__ = new_name
        
        setattr(
            cls, new_name, wrapped_method
        )
            
    @classmethod
    def construct_flow(cls, method_list):        
        first_step = "wrapped_" + method_list[0]
        
        # create start step
        @step
        def start(self):
            print("Running overwritten start step...")
            next_step = getattr(self, first_step)
            self.next(next_step)
        setattr(cls, "start", start)
            
        # wrap each method in method_list as a metaflow step
        for i, method_str in enumerate(method_list):
            if i < len(method_list)-1:
                next_method_str = "wrapped_" + method_list[i+1]
            else:
                next_method_str = "end"                
            cls.wrap_method(method_str, next_method_str)
       
        # create end step
        @step
        def end(self):
            print("Flow is complete!")
        setattr(cls, "end", end)
        
        return cls()

        
class TestDynamicFlow(FlowSpec, DynamicFlowBase, Mixin):
    arg1 = Parameter("arg1", default=1)
    
    # placeholder steps to make Metaflow happy
    @step
    def start(self):
        self.next(self.wrapped_method1)
    @step
    def wrapped_method1(self):
        self.next(self.wrapped_method2)
    @step
    def wrapped_method2(self):
        self.next(self.end)
    @step
    def end(self):
        pass
    

if __name__ == "__main__":
    TestDynamicFlow.construct_flow(TestDynamicFlow.method_list)

TestDynamicFlow

Based on static analysis of the code, step start was expected to transition to step(s) end. However, when the code was executed, self.next() was called with wrapped_method1. Make sure there is only one unconditional self.next() call in the end of your step.