[helm] Support a name-keyed map for dagster-user-deployments.deployments

Allow `dagster-user-deployments.deployments` to be specified as a map keyed
by deployment name, in addition to the existing array form. Helm merges maps
by key but replaces arrays wholesale, so the map form lets layered values
files (e.g. a base file plus per-environment overrides, as used with an
ArgoCD ApplicationSet) override a single field of a single deployment instead
of restating the entire list.

A new `dagsterUserDeployments.deploymentsList` template helper normalizes the
value (list or map) into a canonical list; when a map is used the key becomes
the deployment name. Every existing `range` over deployments (three subchart
templates and the parent chart's workspace configmap, daemon init containers,
and webserver init containers) is routed through the helper, so array-form
output is unchanged.

The pydantic models are updated so `deployments` becomes
`list[UserDeployment] | dict[str, UserDeployment]`. `name` is optional on the
base model (the map key supplies it), but it stays required for the list form:
a `_RequireName` annotation overlays `required: ["name"]` onto the array branch
of the generated JSON schema via `allOf`, reusing the single `UserDeployment`
$def, so a nameless list entry still fails schema validation / `helm lint`. Both
values.schema.json files are regenerated via `dagster-helm schema apply`.

Adds a commented map example to both values.yaml files and a docs section
(noting map keys render in sorted order), plus pytest coverage for the map
form (rendering, naming, array/map equivalence, key-wins-over-name, and the
missing-name guard).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: liadsh-twine

docs/docs/deployment/oss/deployment-options/kubernetes/customizing-your-deployment.md

@@ -229,6 +229,26 @@ global:
 generatePostgresqlPasswordSecret: false
 ```
 
+## Defining user code deployments as a map
+
+The `dagster-user-deployments.deployments` value accepts either a list (the default) or a map keyed by deployment name:
+
+```yaml
+dagster-user-deployments:
+  deployments:
+    k8s-example-user-code-1:
+      image:
+        repository: 'docker.io/dagster/user-code-example'
+        tag: ~
+        pullPolicy: Always
+      dagsterApiGrpcArgs:
+        - '--python-file'
+        - '/example_project/example_repo/repo.py'
+      port: 3030
+```
+
+The map form is useful when you layer values files — for example a base file plus a per-environment override. Helm merges maps by key but replaces lists wholesale, so with the map form an override file can change a single field of a single deployment instead of restating the entire list. When the map form is used, the map key is the deployment name and the per-entry `name` field is optional. Note that Helm renders map keys in alphabetical order, so deployments are always emitted in sorted key order rather than declaration order.
+
 ## Security
 
 Users will likely want to permission a ServiceAccount bound to a properly scoped Role to launch Jobs and create other Kubernetes resources.

helm/dagster/charts/dagster-user-deployments/templates/configmap-env-user.yaml

@@ -1,4 +1,4 @@
-{{ range $deployment := .Values.deployments }}
+{{ range $deployment := (include "dagsterUserDeployments.deploymentsList" .Values.deployments | fromYamlArray) }}
 apiVersion: v1
 kind: ConfigMap
 metadata:

helm/dagster/charts/dagster-user-deployments/templates/deployment-user.yaml

@@ -1,5 +1,5 @@
 {{- $celeryConfigSecretName := .Values.global.celeryConfigSecretName | default .Values.celeryConfigSecretName }}
-{{ range $deployment := .Values.deployments }}
+{{ range $deployment := (include "dagsterUserDeployments.deploymentsList" .Values.deployments | fromYamlArray) }}
 {{- $userDeploymentChecksum := $deployment | toJson | sha256sum }}
 {{- if and $deployment.codeServerArgs (gt (int ($deployment.replicaCount | default 1)) 1) }}
 {{- fail (printf "deployment %q: codeServerArgs cannot be used with replicaCount > 1. A reloadable code server cannot be reloaded consistently across multiple replicas behind a round-robin service." $deployment.name) }}

helm/dagster/charts/dagster-user-deployments/templates/helpers/_helpers.tpl

@@ -6,6 +6,34 @@ Expand the name of the chart.
 {{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" -}}
 {{- end -}}
 
+{{/*
+Normalize the user-code deployments value into a list.
+Accepts either a list of deployment entries or a map keyed by deployment name.
+For the map form, the key becomes each entry's `name` (the key wins over any `name` field).
+Emits YAML that call sites parse with `fromYamlArray`.
+
+NOTE: this helper is intentionally duplicated verbatim in
+  helm/dagster/templates/helpers/_helpers.tpl
+so it is available whether the subchart is rendered under the umbrella chart or
+standalone. Keep both copies in sync when making changes.
+*/}}
+{{- define "dagsterUserDeployments.deploymentsList" -}}
+{{- $deployments := . -}}
+{{- if kindIs "map" $deployments -}}
+{{- $list := list -}}
+{{- range $name, $deployment := $deployments -}}
+{{- $entry := merge (dict "name" $name) (deepCopy $deployment) -}}
+{{- $list = append $list $entry -}}
+{{- end -}}
+{{- $list | toYaml -}}
+{{- else -}}
+{{- range $deployment := (default (list) $deployments) -}}
+{{- if not $deployment.name }}{{ fail "each user deployment must set 'name' (or be provided as a name-keyed map)" }}{{- end }}
+{{- end -}}
+{{- (default (list) $deployments) | toYaml -}}
+{{- end -}}
+{{- end -}}
+
 {{/*
 Create a default fully qualified app name.
 We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec).

helm/dagster/charts/dagster-user-deployments/templates/service-user.yaml

@@ -1,4 +1,4 @@
-{{ range $deployment := .Values.deployments }}
+{{ range $deployment := (include "dagsterUserDeployments.deploymentsList" .Values.deployments | fromYamlArray) }}
 apiVersion: v1
 kind: Service
 metadata:

helm/dagster/charts/dagster-user-deployments/values.schema.json

@@ -28,6 +28,25 @@ includeInstance: false
 # pipeline execution use the same image, we recommend using a unique tag (ie not "latest").
 #
 # All user code will be invoked within the images.
+#
+# "deployments" may be provided either as a list (shown below) or as a map keyed by
+# deployment name. The map form lets layered values files (e.g. a base file plus a
+# per-environment override) deep-merge by key instead of replacing the whole list, so
+# an override can change a single field of a single deployment. When the map form is
+# used, the map key is the deployment name and the per-entry "name" field is optional.
+#
+# Example map form:
+#
+# deployments:
+#   k8s-example-user-code-1:
+#     image:
+#       repository: "docker.io/dagster/user-code-example"
+#       tag: ~
+#       pullPolicy: Always
+#     dagsterApiGrpcArgs:
+#       - "-f"
+#       - "/example_project/example_repo/repo.py"
+#     port: 3030
 ####################################################################################################
 deployments:
   - name: "k8s-example-user-code-1"

helm/dagster/charts/dagster-user-deployments/values.yaml

@@ -1,8 +1,26 @@
-from pydantic import BaseModel, Field, create_model
+from typing import Annotated
+
+from pydantic import BaseModel, Field, GetJsonSchemaHandler, create_model
+from pydantic.json_schema import JsonSchemaValue
+from pydantic_core import CoreSchema
 
 from schema.charts.utils import kubernetes
 
 
+class _RequireName:
+    """Annotation that overlays ``required: ["name"]`` onto a ``UserDeployment``'s JSON
+    schema. Applied only to the list form of ``deployments`` so that a missing ``name`` in
+    an array entry is still caught by schema validation (``helm lint``), while the map form
+    — where the map key supplies the name — leaves ``name`` optional. Reuses the shared
+    ``UserDeployment`` ``$def`` via ``allOf`` rather than duplicating it.
+    """
+
+    def __get_pydantic_json_schema__(
+        self, core_schema: CoreSchema, handler: GetJsonSchemaHandler
+    ) -> JsonSchemaValue:
+        return {"allOf": [handler(core_schema), {"required": ["name"]}]}
+
+
 class UserDeploymentIncludeConfigInLaunchedRuns(BaseModel):
     enabled: bool
 
@@ -13,7 +31,9 @@ class UserDeploymentIncludeConfigInLaunchedRuns(BaseModel):
 
 
 class UserDeployment(BaseModel):
-    name: str
+    # Optional so that deployments can be supplied as a name-keyed map, where the
+    # map key provides the name. For the list form, the Helm templates require a name.
+    name: str | None = None
     image: kubernetes.Image
     dagsterApiGrpcArgs: list[str] | None = None
     codeServerArgs: list[str] | None = None
@@ -45,8 +65,13 @@ class UserDeployment(BaseModel):
     deploymentStrategy: kubernetes.DeploymentStrategy | None = None
 
 
+# `deployments` may be either a list of deployments (each requiring a `name`) or a map
+# keyed by deployment name (where the key supplies the name, so `name` is optional).
+UserDeploymentsValue = list[Annotated[UserDeployment, _RequireName()]] | dict[str, UserDeployment]
+
+
 class UserDeployments(BaseModel):
     enabled: bool
     enableSubchart: bool
     imagePullSecrets: list[kubernetes.SecretRef]
-    deployments: list[UserDeployment]
+    deployments: UserDeploymentsValue

helm/dagster/schema/schema/charts/dagster_user_deployments/subschema/user_deployments.py

@@ -1,7 +1,7 @@
 from pydantic import BaseModel, Field
 
 from schema.charts.dagster.subschema import Global, ServiceAccount
-from schema.charts.dagster_user_deployments.subschema.user_deployments import UserDeployment
+from schema.charts.dagster_user_deployments.subschema.user_deployments import UserDeploymentsValue
 from schema.charts.utils import kubernetes
 
 
@@ -12,7 +12,7 @@ class DagsterUserDeploymentsHelmValues(BaseModel):
     postgresqlSecretName: str
     celeryConfigSecretName: str
     includeInstance: bool
-    deployments: list[UserDeployment]
+    deployments: UserDeploymentsValue
     imagePullSecrets: list[kubernetes.SecretRef]
     serviceAccount: ServiceAccount
     global_: Global = Field(..., alias="global")

helm/dagster/schema/schema/charts/dagster_user_deployments/values.py

@@ -1959,3 +1959,123 @@ def test_include_instance(subchart_template: HelmTemplate, include_instance: boo
         )
     else:
         assert "dagster-instance" not in volume_names
+
+
+@pytest.fixture(name="service_template")
+def service_helm_template() -> HelmTemplate:
+    return HelmTemplate(
+        helm_dir_path="helm/dagster",
+        subchart_paths=["charts/dagster-user-deployments"],
+        output="charts/dagster-user-deployments/templates/service-user.yaml",
+        model=models.V1Service,
+    )
+
+
+def _map_helm_values(deployments: dict[str, UserDeployment]) -> DagsterHelmValues:
+    return DagsterHelmValues.construct(
+        dagsterUserDeployments=UserDeployments.construct(
+            enabled=True,
+            enableSubchart=True,
+            deployments=deployments,
+        )
+    )
+
+
+def test_deployments_map_render(template: HelmTemplate):
+    # deployments provided as a name-keyed map; the key becomes each deployment's name.
+    deployments = {
+        "deployment-one": create_simple_user_deployment("deployment-one"),
+        "deployment-two": create_complex_user_deployment("deployment-two"),
+    }
+    user_deployments = template.render(_map_helm_values(deployments))
+
+    assert len(user_deployments) == len(deployments)
+
+    # Helm ranges maps in sorted-key order.
+    for user_deployment, name in zip(user_deployments, sorted(deployments)):
+        assert user_deployment.metadata.name.endswith(name)
+        assert user_deployment.metadata.labels["deployment"] == name
+        assert user_deployment.spec.template.metadata.labels["deployment"] == name
+
+
+def test_deployments_map_matches_list(template: HelmTemplate):
+    # The same deployment expressed as a single-entry list vs a single-entry map should
+    # render an identical Deployment (including the config checksum annotation).
+    list_values = DagsterHelmValues.construct(
+        dagsterUserDeployments=UserDeployments.construct(
+            enabled=True,
+            enableSubchart=True,
+            deployments=[create_complex_user_deployment("deployment-one")],
+        )
+    )
+    map_values = _map_helm_values(
+        {"deployment-one": create_complex_user_deployment("deployment-one")}
+    )
+
+    [from_list] = template.render(list_values)
+    [from_map] = template.render(map_values)
+
+    assert template.api_client.sanitize_for_serialization(
+        from_list
+    ) == template.api_client.sanitize_for_serialization(from_map)
+
+
+def test_deployments_map_key_wins_over_name(template: HelmTemplate):
+    # When both a map key and an entry-level `name` are present, the key is authoritative.
+    deployment = create_simple_user_deployment("ignored-name")
+    user_deployments = template.render(_map_helm_values({"key-name": deployment}))
+
+    [user_deployment] = user_deployments
+    assert user_deployment.metadata.name.endswith("key-name")
+    assert user_deployment.metadata.labels["deployment"] == "key-name"
+
+
+def test_deployments_map_service(service_template: HelmTemplate):
+    deployments = {
+        "deployment-one": create_simple_user_deployment("deployment-one"),
+        "deployment-two": create_simple_user_deployment("deployment-two"),
+    }
+    services = service_template.render(_map_helm_values(deployments))
+
+    assert len(services) == len(deployments)
+    for service, name in zip(services, sorted(deployments)):
+        assert service.metadata.name == name
+        assert service.spec.selector["deployment"] == name
+
+
+def test_deployments_map_configmap_env(user_deployment_configmap_template: HelmTemplate):
+    deployment = UserDeployment(
+        image=kubernetes.Image(repository="repo/a", tag="tag1", pullPolicy="Always"),
+        dagsterApiGrpcArgs=["-m", "a"],
+        port=3030,
+        env={"FOO": "bar"},
+    )
+    configmaps = user_deployment_configmap_template.render(_map_helm_values({"code-a": deployment}))
+
+    [configmap] = configmaps
+    assert configmap.metadata.name.endswith("-code-a-user-env")
+    assert configmap.data["FOO"] == "bar"
+
+
+def test_deployments_list_missing_name_fails(template: HelmTemplate, capfd):
+    # A list entry without a name is invalid (only the map form can omit it). The schema
+    # requires `name` for the array branch, so this fails at values-validation time; the
+    # template also guards it as a backstop when schema validation is skipped.
+    nameless = UserDeployment.construct(
+        image=kubernetes.Image(repository="repo/x", tag="tag1", pullPolicy="Always"),
+        dagsterApiGrpcArgs=["-m", "x"],
+        port=3030,
+    )
+    with pytest.raises(subprocess.CalledProcessError):
+        template.render(
+            DagsterHelmValues.construct(
+                dagsterUserDeployments=UserDeployments.construct(
+                    enabled=True,
+                    enableSubchart=True,
+                    deployments=[nameless],
+                )
+            )
+        )
+
+    _, err = capfd.readouterr()
+    assert "name is required" in err