Is there a way for me to define a Flow in such a way as to be generic and then I can subclass it for each child flow? I am looking to reduce code repetition for some flows I have that follow a general pattern, but have small differences, or added steps/tasks. I don't think the imperative API is what I am looking for.

I have seen some users do this. Here is an example:

class OurFlow(PrefectFlow):
    def __init__(self, name, **kwargs):
        super().__init__(name, storage=Docker(registry_url='<http://gcr.io/pict-app|gcr.io/pict-app>',
                                              image_name=name,
                                              base_image=base_image(),
                                              path=path_in_docker(inspect.stack()[1].filename),
                                              stored_as_script=True))
        self.run_config = UniversalRun(labels=[label()])
        self.executor=DaskExecutor(address='<tcp://dask-scheduler:8786>', debug=True)

ok, so if I take the parameters I pass to the

with Flow()

and pass them to

super().__init()

I can then add Tasks via the imperative API inside my init?

I don’t think you can set Tasks like that with this init, unless you just call Task.run() which calls the Python function underneath the task. At that point, you lose the monitoring since it isn’t a Task. I think it might be easier if you create a function that generates a flow and return it, and then go imperative style all the way?

hmmm... that may not be as helpful as I would want, but it sounds promsing. Thank you