[Python] How to use Airflow params to template operator fields that receive complex objects

I am building an Airflow DAG with a KubernetesPodOperator that I would like to parametrise heavily.

I would like to to parametrise cmds, image, and volume_mounts mount path. The first two work as I intend from the get go, however, I find parametrising mount path very difficult to deal with. I have double checked that all of the fields are included in template_fields in this operator.

Here is the code:

@dag(
dag_id=PIPELINE_NAME,
schedule=None,
dag_display_name="Run arbitrary command with persistent data space",
max_active_runs=1,
)
def run_arbitary_command_pipeline(
command: str = "",
image: str = "python:3.13-slim",
data_storage_size: str = "10Gi",
shared_data_mount_path="/mnt/data/",
):
pipeline_name_normalized = PIPELINE_NAME.replace("_", "-").lower()
pvc_name = f"{pipeline_name_normalized}-data-space-pvc"

run_command = KubernetesPodOperator(
task_id="run_arbitrary_command",
cmds=["sh", "-c", command],
image=image,
task_display_name="Run arbitrary command",
volumes=[k8s.V1Volume(name=pvc_name, persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(claim_name=pvc_name))],
volume_mounts=[k8s.V1VolumeMount(name=pvc_name, mount_path=shared_data_mount_path)],
)


Airflow automagically deals with command and image – I can see both of them correctly expanded in the "Rendered Templates" section in UI. However, volume_mounts is expanded to:

"[{'mount_path': <airflow.sdk.definitions.param.DagParam object at 0xfad05e250590>,\n 'mount_propagation': None,\n 'name': 'run-arbitrary-command-pipeline-data-space-pvc',\n 'read_only': None,\n 'recursive_read_only': None,\n 'sub_path': None,\n 'sub_path_expr': None}]"


And the pipeline fails because:

RepresenterError: ('cannot represent an object', <airflow.sdk.definitions.param.DagParam object at 0xfad05cb8ede0>)


Which is not surprising given how the rendered template looks like.

If I try to use templating in a more explicit way with:

volume_mounts=[k8s.V1VolumeMount(name=pvc_name, mount_path="{{ params.shared_data_mount_path }}")],


then the template is not expanded at all:

"[{'mount_path': '{{ params.shared_data_mount_path }}',\n 'mount_propagation': None,\n 'name': 'run-arbitrary-command-pipeline-data-space-pvc',\n 'read_only': None,\n 'recursive_read_only': None,\n 'sub_path': None,\n 'sub_path_expr': None}]"


and the pipeline doesn't fail, but the parameter effectively doesn't work as the template string is used verbatim as mount path.

Is there any way to fix this and mount the PVC under the path provided from params? I feel that this problem results both from Airflow limitations and my poor understanding of its intricacies. Would appreciate any good sources to understand how and when the templating works beyond what the docs explain.

Continue reading...