How can I dynamically pass parameters to a Prefect flow running as a Kubernetes Job (Pod) or Docker container?
I’m currently using the
flow.from_source()
method, which downloads the flow code from an S3 bucket, builds the image dynamically, and then runs the flow.
However, I’d like to avoid building the Docker image at runtime. Is there a way to use a
prebuilt Docker image (pulled from a registry) and still pass parameters dynamically — preferably through a
Pythonic method or REST API — without relying on deployment templates or environment variables?
I understand that
job_variables
can be set dynamically and overridden during a flow run, but I’m looking for an alternative that allows passing parameters (like
flow_data
) more flexibly — ideally at runtime — in a way that automatically maps values from
flow_data
into
job_variables
and passes them as arguments to the flow entrypoint, while using a pre-built Docker image.
flow_from_source = await flow.from_source(
source=s3_bucket_block,
entrypoint="flows/bill_flow.py:bill_assessment_flow"
)
flow_dependencies = get_flow_dependencies()
deployment = await flow_from_source.deploy(
name=PREFECT_DEPLOYMENT_NAME,
tags=["billing"],
work_pool_name="kubernetes-pool",
schedule=None,
push=False, # Skip pushing image
job_variables={
"finished_job_ttl": 100,
# "image": "mat/prefect-k8s-worker:15", # Uncomment to use a custom prebuilt image
"namespace": "prefect",
"env": {
"PREFECT_API_URL": "<http://prefect-server:4200/api>",
"EXTRA_PIP_PACKAGES": flow_dependencies,
"PYTHONPATH": "/opt/prefect/"
}
}
)
app.state.deployment_id = deployment
flow_run = await client.create_flow_run_from_deployment(
deployment_id=request.app.state.deployment_id,
tags=run_tags,
parameters={
"flow_data": {
"source_provider": source_provider,
"target_provider": target_provider,
"company_id": company_id,
"company_name": company_name,
"assessment_task_id": assessment_task_id
}
}
)
<http://logger.info|logger.info>(f"Created flow run with ID: {flow_run.id}")
