from prefect import flow
import asyncio
import coolname
import random

BEGS = ["I'd love a Prefect keyboard.",
        "Yay Marvin!",
        "I wish I had a Prefect rubber duck too!"]\



def generate_subflow(level=2, subflow_count=3):
    if level > 0:
        @flow(retries=10, name=f"subflow-{level}-{coolname.generate_slug(2)}")
        async def subflow():
            if random.random() < 0.05:
                await asyncio.sleep(10)
                raise RuntimeError("Sometimes I fail. But Prefect helps me try again.")
            else:
                print(random.choice(BEGS))

            next_level = [generate_subflow(level=level-1, subflow_count=subflow_count)()
                          for _ in range(subflow_count)]
            await asyncio.gather(*next_level)
        return subflow
    else:
        @flow
        async def goodbye():
            print("Goodbye for now!")
        return goodbye


@flow
async def contest_flow(subflow_count=25, levels=10):
    to_run = [generate_subflow(level=levels, subflow_count=subflow_count)() for _ in range(subflow_count)]
    await asyncio.gather(*to_run)


if __name__ == "__main__":
    results = asyncio.run(contest_flow())