async def METHOD_NAME(self, project_id, tmpdir):
    """
    A simple diamond flow
    """
    @prefect.task
    def numbers():
        return [1, 2, 3]
    @prefect.task
    def add(x):
        return x + 1
    @prefect.task
    def get_sum(x):
        return sum(x)
    with prefect.Flow(
        "simple reduce",
        run_config=prefect.run_configs.LocalRun(),
        storage=Local(directory=tmpdir),
    ) as METHOD_NAME:
        METHOD_NAME.numbers = numbers()
        METHOD_NAME.add = add.map(METHOD_NAME.numbers)
        METHOD_NAME.sum = get_sum(METHOD_NAME.add)
    add_state_handler(METHOD_NAME)
    with set_temporary_config(key="dev", value=True):
        METHOD_NAME.server_id = await api.flows.create_flow(
            project_id=project_id, serialized_flow=METHOD_NAME.serialize(build=True)
        )
    return METHOD_NAME