https://prefect.io logo
w

wiretrack

12/24/2020, 8:19 PM
any ideas, hints, suggestions, links to articles that could be of any help?
d

Dylan

12/24/2020, 8:20 PM
Also, we ❤️threads here, helps keep the main channel readable
w

wiretrack

12/24/2020, 8:22 PM
@Dylan sorry bout that, my bad
d

Dylan

12/24/2020, 8:22 PM
No worries! 😄
Let me know if you have any further questions marvin
w

wiretrack

12/24/2020, 8:23 PM
@Dylan thanks, I did find the article, but still couldn’t apply it to my use case. i do have the StartFlowRun and later the
with Flow('parent-flow') as flow
setting the upstreams
the comment says: “assumes you have registered the following flows”. isn’t it what I’m doing when I say
customer_flow = Flow('customer_flow')
?
d

Dylan

12/24/2020, 8:25 PM
Ahh, it is not
You need to create the flow and tasks and then call ‘flow.register()’
One moment
w

wiretrack

12/24/2020, 8:28 PM
hm.. but even if i move the registration below everything, still get the same error (I do have the tasks refined above this code block) maybe I’m not setting 
dependencies
 in the context of the imperative api maybe they are required?
the imperative api example:
Copy code
from prefect import Task, Flow

# The tasks
class RunMeFirst(Task):
    def run(self):
        print("I'm running first!")

class PlusOneTask(Task):
    def run(self, x):
        return x + 1

# Instanciates the Flow
flow = Flow('My Imperative Flow')

# The task instance (I'm using the decorator format), this is where I get the error
plus_one = PlusOneTask()

# I don't have that.
flow.set_dependencies(
    task=plus_one,
    upstream_tasks=[RunMeFirst()],
    keyword_tasks=dict(x=10))

flow.visualize()
d

Dylan

12/24/2020, 8:41 PM
I’m not sure if those case statements are being set correctly with the imperative api
Since they are themselves tasks
I think the context manager is only available within a flow context block
like this
Copy code
with Flow("my flow") as flow:
  result = my_task()
  with case(result, True):
    #do something
w

wiretrack

12/24/2020, 8:47 PM
i think i got that from the docs but let me double check. the error appear in my second line (check_status task), do you think that the case statements must be interfering?
d

Dylan

12/24/2020, 8:48 PM
I’m not as familiar with the imperative api
w

wiretrack

12/24/2020, 8:49 PM
fair enough, let me try to adapt to the functional
d

Dylan

12/24/2020, 8:59 PM
You’ll need to register the parent as well
w

wiretrack

12/24/2020, 8:59 PM
yes yes yes, I realized I was making a stupid mistake, lol
cool, seems to work with the functional approach. thanks a lot mate!
d

Dylan

12/24/2020, 9:01 PM
anytime! And Happy Holidays 😄
w

wiretrack

12/24/2020, 9:01 PM
i’ll keep digging around, currently building a POC for an integration we have with about 10-15 different ERP systems that have to run every 1-5 minutes with different query for each system and each client. is been a nightmare to manage it trhough procedures and linked servers, no visibility on the health of the system whatsoever. airflow didn’t do the trick, NiFi is a bit outdated and not very nice to work when you have tons of flows, so Prefect seems perfect for the job, happy to have found it!
d

Dylan

12/24/2020, 9:02 PM
Glad to hear it!! And welcome to the community prefect duck
w

wiretrack

12/24/2020, 9:02 PM
thanks a lot! happy hollydays and thank you very much for your help, best!