import prefect
from prefect import task, Flow, Parameter

@task
def hello_task(name):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(f"Hello {name}!")
    

with Flow("hello-flow") as flow:
    # An optional parameter "people", with a default list of names
    people = Parameter("people", default=["Arthur", "Ford", "Marvin"])
    # Map `say_hello` across the list of names
    hello_task.map(people)

#Register the flow
flow.register(project_name="FirstProject")

#Prior to running/registration, ensure that an agent is running in a seperate terminal/machine in order to intercept the created flow run.
from prefect import Flow
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
import pendulum

with Flow("parent_flow") as flow:
    child_flow_run_id = create_flow_run(
        flow_name="hello-flow", 
        project_name= "FirstProject", 
        scheduled_start_time = pendulum.now().add(minutes=2), #Replace with 60 minutes.
        parameters={"people" : ['Bianca', 'Roger']}
    )
    child_flowrunview = wait_for_flow_run(
        child_flow_run_id, 
        raise_final_state=True, 
        stream_logs=True
    )

#Register the flow first
#flow.register(project_name="FirstProject")

#Then, run the flow 
#flow.run()

"parent_flow"

"hello-flow"