Provide DaskRunner configuration(s) via block

[imagejan]

Can we provide configuration like this one:

prefect-workflows/eicm_flows/compute_all_matrices.py

Lines 46 to 73 in 69c4c41

	task_runner=DaskTaskRunner(
	cluster_class="dask_jobqueue.SLURMCluster",
	cluster_kwargs={
	"account": "dlthings",
	"queue": "cpu_long",
	"cores": 2,
	"processes": 1,
	"memory": "4 GB",
	"walltime": "1:00:00",
	"job_extra_directives": [
	"--ntasks=1",
	"--output=/tungstenfs/scratch/gmicro_share/_prefect/slurm/output/%j.out",
	],
	"worker_extra_args": [
	"--lifetime",
	"60m",
	"--lifetime-stagger",
	"10m",
	],
	"job_script_prologue": [
	"conda run -p /tungstenfs/scratch/gmicro_share/_prefect/miniconda3/envs/airtable python /tungstenfs/scratch/gmicro_share/_prefect/airtable/log-slurm-job.py --config /tungstenfs/scratch/gmicro/_prefect/airtable/slurm-job-log.ini"
	],
	},
	adapt_kwargs={
	"minimum": 1,
	"maximum": 3,
	},
	)

via a block, so we avoid

• some boilerplate code for recurring task runner definitions, and
• storing infrastructure-specific information (e.g. network paths) in the source code?

[tibuch]

Parts of it probably yes, but last I tried I could not reliably retrieve the data from blocks outside of the flow.

[imagejan]

This worked for me (with prefect==2.7.7):

from prefect.blocks.system import JSON

test_kwargs = JSON.load('test-kwargs').value


@flow(
    **test_kwargs
)
def test_flow():
    get_run_logger().info("Running test flow...")

... where the test-kwargs block provided a name and description attribute for the flow annotation. (I first tested a custom block class, before noticing that the JSON block provided by prefect serves our purpose just right.) Tested on the faim-dev workspace.