Sam Greenwood
12/04/2025, 5:05 PMMarvin
12/04/2025, 5:05 PMMarvin
12/04/2025, 5:12 PMfrom prefect import flow
@flow
def my_flow():
...
# Deploy from a git repo (pull step auto-generated) and set container working dir
flow.from_source(
source="<https://github.com/org/repo.git>",
entrypoint="flows/my_flow.py:my_flow",
).deploy(
name="docker-wd",
work_pool_name="my-docker-pool",
image="my-registry/my-image:latest",
job_variables={
"container_create_kwargs": {
"working_dir": "/app" # container’s cwd at start
}
},
)
Notes:
- This uses docker-py’s ContainerCollection.create kwargs; Prefect passes them through, so working_dir works here.
- Alternative: bake WORKDIR into your Dockerfile.
Kubernetes worker
- Set the pod container workingDir in the job manifest:
from prefect import flow
@flow
def my_flow():
...
flow.from_source(
source="<https://github.com/org/repo.git>",
entrypoint="flows/my_flow.py:my_flow",
).deploy(
name="k8s-wd",
work_pool_name="my-k8s-pool",
job_variables={
"namespace": "default",
"job_manifest": {
"apiVersion": "batch/v1",
"kind": "Job",
"metadata": {"labels": "{{ labels }}"},
"spec": {
"template": {
"spec": {
"containers": [{
"name": "prefect-job",
"image": "{{ image }}",
"workingDir": "/app" # container’s cwd at start
}],
"restartPolicy": "Never",
}
}
}
},
},
)
If you truly want a pull step to change directories
- You can only do this via prefect.yaml, not via Flow.deploy():
# prefect.yaml
pull:
- prefect.deployments.steps.set_working_directory:
directory: /app
- From Python, pull steps are inferred from storage:
- Git -> git_clone
- Remote storage -> pull_from_remote_storage
- Local path -> set_working_directory
- If you use flow.from_source(source="/app", ...), Prefect will generate a set_working_directory pull step to that path, but the path must exist in the runtime environment (e.g., inside the container).
References
- Docker worker docs: Prefect Docker integration
- Kubernetes worker docs: Prefect Kubernetes integration
- docker-py container create (supports working_dir): docker-py ContainerCollection.create
- Deploy flows: Deploying flows in Prefect