Hi @Marvin, if I was creating a class that has some defined flows, what is the best way of initialising calling flows so that I can dynamically set DaskTaskRunner?

To dynamically set a

DaskTaskRunner

for flows within a class in Prefect 3.x, you can follow these steps: 1. Define Your Class and Flows: Create a class and define your flows using the

@flow

decorator. You can set the

task_runner

parameter dynamically. 2. Initialize the DaskTaskRunner: When you instantiate your class, you can choose to initialize the

DaskTaskRunner

and pass it to the flow. Here's a basic example to illustrate:

python
from prefect import flow
from prefect_dask.task_runners import DaskTaskRunner

class MyFlowClass:
    def __init__(self, use_dask: bool = False):
        self.task_runner = DaskTaskRunner() if use_dask else None

    @flow
    def my_flow(self):
        # Your flow logic here
        pass

    def run_flow(self):
        # Run the flow with the specified task runner
        self.my_flow(task_runner=self.task_runner)

# Usage
my_flow_instance = MyFlowClass(use_dask=True)
my_flow_instance.run_flow()

### Key Points: - Dynamic Task Runner: By setting

use_dask

to

True

, you initialize the

DaskTaskRunner

and pass it to the flow when calling

run_flow()

. - Flexible Design: This design allows you to decide at runtime whether to use Dask for task execution. ### Additional Resources: - Prefect Flows Documentation - Prefect Dask Integration This approach should align well with your preference for imperative programming. If you have further questions or need more specific implementations, feel free to ask.