Hi Prefect! Quick question about flow parameters: if I wanted to know what parameters a flow “accepts”, I’m aware I can do

flow.parameters() -> Set[Parameter]

. But it seems like the implementation is “shallow,” in that it just checks for top-level tasks that are parameters (as opposed to recursively pulling parameters from tasks):

def parameters(self) -> Set[Parameter]:
    """
    Returns any parameters of the flow.

    Returns:
        - set: a set of any Parameters in this flow
    """
    return {p for p in self.tasks if isinstance(p, Parameter)}

Is there a reasonable way to extend the Flow class to search for all parameters declared within the

Flow(..) as flow

context manager?

Hi @Danny Vilela -- I think it is deceptive that the parameters check is shallow there. When a task is added (called in a flow context), any tasks detected in its arguments are also added to

Flow.tasks

and that's how parameters end up there. Can you explain a bit more about the issue you're running into?

Ah, yeah. So I declare parameters within my flow, but I pass them into configs (e.g., a

PipelineConfig

dataclass). This/these config(s) can be shared across multiple tasks, and I do that to allow each task to uniformly access the same configuration settings as any other task. However, if I declare parameters within my flow and pass them to those config(s), they’re not bound to the flow’s tasks and thus not picked up from

flow.parameters()

. I’m fully open to the idea that this isn’t a best practice, and that it may even be an anti-pattern when working with prefect. I guess I’m wondering if this has come up before, and if it’s been solved. Or maybe I’m not properly applying the idea of parameterizing my flow?

I think the config should be a task?

config = create_config(param_a, ...)

where

create_config(...) -> PipelineConfig

?

Ahh yeah, that sounds reasonable! I’ll give that a shot. Thanks @Zanie!