https://prefect.io logo
Title
a

Anjaneyulu K

11/08/2022, 10:29 PM
Hi #prefect-getting-started, Got Runtime error as task runner is already started in below scenario
@flow(name="Status check")
def run():
    a= "hello"
    if(condition):
        success
    else:
        run()

if __name__ == "__main__":
   run()
Can anyone help
a

Anna Geller

11/08/2022, 11:58 PM
can you post full code and the exception?
a

Anjaneyulu K

11/08/2022, 11:59 PM
@flow(name="Status check")
def run():
    a= task1 output

    if(task1 output condition):
        running task2
    else:
       wait(10)
        run()
Runtime error : task runner is already started
very simple and straight forwarded
a

Anna Geller

11/09/2022, 12:02 AM
this is not a full reproducible example
a

Anjaneyulu K

11/09/2022, 12:22 AM
from prefect import task, flow
import time
import random


from prefect.deployments import Deployment


@task
def getNumber(x):
    return x
@task
def hello():
    print("Hello world")

@flow(name="Status check")
def run(x):
    a= getNumber(x)

    if a == 2:
        hello()
    else:
        time.sleep(5)
        run(x)

if __name__ == "__main__":
   run(random.randint(0,9))
   deployment = Deployment.build_from_flow(
       flow=run,
       name="ex-dep2",
       version=1,
       work_queue_name="demo-exp1"      

   )
   deployment.apply()
This will reproduce the error.
a

Anna Geller

11/09/2022, 2:42 AM
oh I see, you are running the flow run from itself, this is not allowed, why would you want to do that?