Does anyone have any experience with dynamically configuring a state handler for a flow or some subset of a flow's tasks? We have a flow that basically serves as a "flow template" (specifically, for running different

dbt

projects) that multiple teams at my org use. The individual teams will set up an orchestrator flow that has one task (the

StartFlowRun

task), and they'll pass in the various parameters that they need to pass in so that their specific

dbt

project will run. They also will directly attach state handlers that they configure to their

StartFlowRun

task within their orchestrator flow. Of course, in order for the status of the underlying template flow to bubble up to the orchestrator flow, we have to set

wait=True

on that flow. The issue with this setup is that it unnecessarily eats up our flow concurrency, since this setup always leads to at least 2 flows running at a given time. We don't really need the orchestrator flow to wait around; we're only keeping it waiting as a means to capture the template flow's status, which then gets passed along to the custom state handler that the respective team within our org has configured.

Hey @Sean Talia, I think the best thing you can do here is pass in a Parameter to the child flow, and then in the state handler, pull the parameter from the context and then choose what to do based on that. You can’t dynamically attach something because it’s already attached during flow registration.

okay I think I see what you're saying; the child flow's state handler (or any of the state handlers attached to the child flow's tasks) needs to have an additional argument in the function signature that would be used for configuring the handler's notification method/destination?

I mean the

prefect.context.get("parameters")

and this returns a dict of parameters inside the state handler.

oh amazing! I didn't realize that those parameter values were available in the context like that, that's way easier

The context in general is available so you can pull stuff like

task_run_count