We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[sdk]  Missing optional parameter to  v2 `use_config_map_as_volume` and `use_secret_as_volume` 
In kfp v1 the config map or secret can be marked as optional when used as a volume source, as shown in the example below.

This capability is missing in the recent added `use_config_map_as_volume` and `use_secret_as_volume` v2 sdk functions. (https://github.com/kubeflow/pipelines/pull/10400, https://github.com/kubeflow/pipelines/pull/10483)

```bash
vol = k8s_client.V1Volume(
        name='dummy',
        config_map=k8s_client.V1ConfigMapVolumeSource(name='name', optional=True),
    )
    component.add_pvolumes({mountPoint: vol})


vol = k8s_client.V1Volume(
        name='dummy',
        secret=k8s_client.V1SecretVolumeSource(secret_name='name', optional=False),
    )
    component.add_pvolumes({mountPoint: vol})

```


What is the use case or pain point?

Making the secret or config map an optional resource allows the KFP component to function without failing even if those resources are not present.



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `kubernetes_platform/python/kfp/kubernetes/secret.py`
Content:
```
1 # Copyright 2023 The Kubeflow Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 from typing import Dict
16 
17 from google.protobuf import json_format
18 from kfp.dsl import PipelineTask
19 from kfp.kubernetes import common
20 from kfp.kubernetes import kubernetes_executor_config_pb2 as pb
21 
22 
23 def use_secret_as_env(
24     task: PipelineTask,
25     secret_name: str,
26     secret_key_to_env: Dict[str, str],
27 ) -> PipelineTask:
28     """Use a Kubernetes Secret as an environment variable as described by the `Kubernetes documentation
29     https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-environment-variables `_.
30 
31     Args:
32         task: Pipeline task.
33         secret_name: Name of the Secret.
34         secret_key_to_env: Dictionary of Secret data key to environment variable name. For example, ``{'password': 'PASSWORD'}`` sets the data of the Secret's password field to the environment variable ``PASSWORD``.
35 
36     Returns:
37         Task object with updated secret configuration.
38     """
39 
40     msg = common.get_existing_kubernetes_config_as_message(task)
41 
42     key_to_env = [
43         pb.SecretAsEnv.SecretKeyToEnvMap(
44             secret_key=secret_key,
45             env_var=env_var,
46         ) for secret_key, env_var in secret_key_to_env.items()
47     ]
48     secret_as_env = pb.SecretAsEnv(
49         secret_name=secret_name,
50         key_to_env=key_to_env,
51     )
52 
53     msg.secret_as_env.append(secret_as_env)
54 
55     task.platform_config['kubernetes'] = json_format.MessageToDict(msg)
56 
57     return task
58 
59 
60 def use_secret_as_volume(
61     task: PipelineTask,
62     secret_name: str,
63     mount_path: str,
64 ) -> PipelineTask:
65     """Use a Kubernetes Secret by mounting its data to the task's container as
66     described by the `Kubernetes documentation <https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-files-from-a-pod>`_.
67 
68     Args:
69         task: Pipeline task.
70         secret_name: Name of the Secret.
71         mount_path: Path to which to mount the Secret data.
72 
73     Returns:
74         Task object with updated secret configuration.
75     """
76 
77     msg = common.get_existing_kubernetes_config_as_message(task)
78 
79     secret_as_vol = pb.SecretAsVolume(
80         secret_name=secret_name,
81         mount_path=mount_path,
82     )
83 
84     msg.secret_as_volume.append(secret_as_vol)
85 
86     task.platform_config['kubernetes'] = json_format.MessageToDict(msg)
87 
88     return task
89
```

Path: `kubernetes_platform/python/kfp/kubernetes/config_map.py`
Content:
```
1 # Copyright 2024 The Kubeflow Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 from typing import Dict
16 
17 from google.protobuf import json_format
18 from kfp.dsl import PipelineTask
19 from kfp.kubernetes import common
20 from kfp.kubernetes import kubernetes_executor_config_pb2 as pb
21 
22 
23 def use_config_map_as_env(
24     task: PipelineTask,
25     config_map_name: str,
26     config_map_key_to_env: Dict[str, str],
27 ) -> PipelineTask:
28     """Use a Kubernetes ConfigMap as an environment variable as described by the `Kubernetes documentation
29     https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#define-container-environment-variables-using-configmap-data` _.
30 
31     Args:
32         task: Pipeline task.
33         config_map_name: Name of the ConfigMap.
34         config_map_key_to_env: Dictionary of ConfigMap key to environment variable name. For example, ``{'foo': 'FOO'}`` sets the value of the ConfigMap's foo field to the environment variable ``FOO``.
35 
36     Returns:
37         Task object with updated ConfigMap configuration.
38     """
39 
40     msg = common.get_existing_kubernetes_config_as_message(task)
41 
42     key_to_env = [
43         pb.ConfigMapAsEnv.ConfigMapKeyToEnvMap(
44             config_map_key=config_map_key,
45             env_var=env_var,
46         ) for config_map_key, env_var in config_map_key_to_env.items()
47     ]
48     config_map_as_env = pb.ConfigMapAsEnv(
49         config_map_name=config_map_name,
50         key_to_env=key_to_env,
51     )
52 
53     msg.config_map_as_env.append(config_map_as_env)
54 
55     task.platform_config['kubernetes'] = json_format.MessageToDict(msg)
56 
57     return task
58 
59 
60 def use_config_map_as_volume(
61     task: PipelineTask,
62     config_map_name: str,
63     mount_path: str,
64 ) -> PipelineTask:
65     """Use a Kubernetes ConfigMap by mounting its data to the task's container as
66     described by the `Kubernetes documentation <https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#add-configmap-data-to-a-volume>`_.
67 
68     Args:
69         task: Pipeline task.
70         config_map_name: Name of the ConfigMap.
71         mount_path: Path to which to mount the ConfigMap data.
72 
73     Returns:
74         Task object with updated ConfigMap configuration.
75     """
76 
77     msg = common.get_existing_kubernetes_config_as_message(task)
78 
79     config_map_as_vol = pb.ConfigMapAsVolume(
80         config_map_name=config_map_name,
81         mount_path=mount_path,
82     )
83     msg.config_map_as_volume.append(config_map_as_vol)
84 
85     task.platform_config['kubernetes'] = json_format.MessageToDict(msg)
86 
87     return task
88
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/kubernetes_platform/python/kfp/kubernetes/config_map.py b/kubernetes_platform/python/kfp/kubernetes/config_map.py
--- a/kubernetes_platform/python/kfp/kubernetes/config_map.py
+++ b/kubernetes_platform/python/kfp/kubernetes/config_map.py
@@ -61,6 +61,7 @@
     task: PipelineTask,
     config_map_name: str,
     mount_path: str,
+    optional: bool = False,
 ) -> PipelineTask:
     """Use a Kubernetes ConfigMap by mounting its data to the task's container as
     described by the `Kubernetes documentation <https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#add-configmap-data-to-a-volume>`_.
@@ -69,6 +70,7 @@
         task: Pipeline task.
         config_map_name: Name of the ConfigMap.
         mount_path: Path to which to mount the ConfigMap data.
+        optional: Optional field specifying whether the ConfigMap must be defined.
 
     Returns:
         Task object with updated ConfigMap configuration.
@@ -79,6 +81,7 @@
     config_map_as_vol = pb.ConfigMapAsVolume(
         config_map_name=config_map_name,
         mount_path=mount_path,
+        optional=optional,
     )
     msg.config_map_as_volume.append(config_map_as_vol)
 
diff --git a/kubernetes_platform/python/kfp/kubernetes/secret.py b/kubernetes_platform/python/kfp/kubernetes/secret.py
--- a/kubernetes_platform/python/kfp/kubernetes/secret.py
+++ b/kubernetes_platform/python/kfp/kubernetes/secret.py
@@ -61,6 +61,7 @@
     task: PipelineTask,
     secret_name: str,
     mount_path: str,
+    optional: bool = False,
 ) -> PipelineTask:
     """Use a Kubernetes Secret by mounting its data to the task's container as
     described by the `Kubernetes documentation <https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-files-from-a-pod>`_.
@@ -69,6 +70,7 @@
         task: Pipeline task.
         secret_name: Name of the Secret.
         mount_path: Path to which to mount the Secret data.
+        optional: Optional field specifying whether the Secret must be defined.
 
     Returns:
         Task object with updated secret configuration.
@@ -79,6 +81,7 @@
     secret_as_vol = pb.SecretAsVolume(
         secret_name=secret_name,
         mount_path=mount_path,
+        optional=optional,
     )
 
     msg.secret_as_volume.append(secret_as_vol)

{"golden_diff": "diff --git a/kubernetes_platform/python/kfp/kubernetes/config_map.py b/kubernetes_platform/python/kfp/kubernetes/config_map.py\n--- a/kubernetes_platform/python/kfp/kubernetes/config_map.py\n+++ b/kubernetes_platform/python/kfp/kubernetes/config_map.py\n@@ -61,6 +61,7 @@\n     task: PipelineTask,\n     config_map_name: str,\n     mount_path: str,\n+    optional: bool = False,\n ) -> PipelineTask:\n     \"\"\"Use a Kubernetes ConfigMap by mounting its data to the task's container as\n     described by the `Kubernetes documentation <https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#add-configmap-data-to-a-volume>`_.\n@@ -69,6 +70,7 @@\n         task: Pipeline task.\n         config_map_name: Name of the ConfigMap.\n         mount_path: Path to which to mount the ConfigMap data.\n+        optional: Optional field specifying whether the ConfigMap must be defined.\n \n     Returns:\n         Task object with updated ConfigMap configuration.\n@@ -79,6 +81,7 @@\n     config_map_as_vol = pb.ConfigMapAsVolume(\n         config_map_name=config_map_name,\n         mount_path=mount_path,\n+        optional=optional,\n     )\n     msg.config_map_as_volume.append(config_map_as_vol)\n \ndiff --git a/kubernetes_platform/python/kfp/kubernetes/secret.py b/kubernetes_platform/python/kfp/kubernetes/secret.py\n--- a/kubernetes_platform/python/kfp/kubernetes/secret.py\n+++ b/kubernetes_platform/python/kfp/kubernetes/secret.py\n@@ -61,6 +61,7 @@\n     task: PipelineTask,\n     secret_name: str,\n     mount_path: str,\n+    optional: bool = False,\n ) -> PipelineTask:\n     \"\"\"Use a Kubernetes Secret by mounting its data to the task's container as\n     described by the `Kubernetes documentation <https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-files-from-a-pod>`_.\n@@ -69,6 +70,7 @@\n         task: Pipeline task.\n         secret_name: Name of the Secret.\n         mount_path: Path to which to mount the Secret data.\n+        optional: Optional field specifying whether the Secret must be defined.\n \n     Returns:\n         Task object with updated secret configuration.\n@@ -79,6 +81,7 @@\n     secret_as_vol = pb.SecretAsVolume(\n         secret_name=secret_name,\n         mount_path=mount_path,\n+        optional=optional,\n     )\n \n     msg.secret_as_volume.append(secret_as_vol)\n", "issue": "[sdk]  Missing optional parameter to  v2 `use_config_map_as_volume` and `use_secret_as_volume` \nIn kfp v1 the config map or secret can be marked as optional when used as a volume source, as shown in the example below.\r\n\r\nThis capability is missing in the recent added `use_config_map_as_volume` and `use_secret_as_volume` v2 sdk functions. (https://github.com/kubeflow/pipelines/pull/10400, https://github.com/kubeflow/pipelines/pull/10483)\r\n\r\n```bash\r\nvol = k8s_client.V1Volume(\r\n        name='dummy',\r\n        config_map=k8s_client.V1ConfigMapVolumeSource(name='name', optional=True),\r\n    )\r\n    component.add_pvolumes({mountPoint: vol})\r\n\r\n\r\nvol = k8s_client.V1Volume(\r\n        name='dummy',\r\n        secret=k8s_client.V1SecretVolumeSource(secret_name='name', optional=False),\r\n    )\r\n    component.add_pvolumes({mountPoint: vol})\r\n\r\n```\r\n\r\n\r\nWhat is the use case or pain point?\r\n\r\nMaking the secret or config map an optional resource allows the KFP component to function without failing even if those resources are not present.\r\n\r\n\n", "before_files": [{"content": "# Copyright 2023 The Kubeflow Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom typing import Dict\n\nfrom google.protobuf import json_format\nfrom kfp.dsl import PipelineTask\nfrom kfp.kubernetes import common\nfrom kfp.kubernetes import kubernetes_executor_config_pb2 as pb\n\n\ndef use_secret_as_env(\n    task: PipelineTask,\n    secret_name: str,\n    secret_key_to_env: Dict[str, str],\n) -> PipelineTask:\n    \"\"\"Use a Kubernetes Secret as an environment variable as described by the `Kubernetes documentation\n    https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-environment-variables `_.\n\n    Args:\n        task: Pipeline task.\n        secret_name: Name of the Secret.\n        secret_key_to_env: Dictionary of Secret data key to environment variable name. For example, ``{'password': 'PASSWORD'}`` sets the data of the Secret's password field to the environment variable ``PASSWORD``.\n\n    Returns:\n        Task object with updated secret configuration.\n    \"\"\"\n\n    msg = common.get_existing_kubernetes_config_as_message(task)\n\n    key_to_env = [\n        pb.SecretAsEnv.SecretKeyToEnvMap(\n            secret_key=secret_key,\n            env_var=env_var,\n        ) for secret_key, env_var in secret_key_to_env.items()\n    ]\n    secret_as_env = pb.SecretAsEnv(\n        secret_name=secret_name,\n        key_to_env=key_to_env,\n    )\n\n    msg.secret_as_env.append(secret_as_env)\n\n    task.platform_config['kubernetes'] = json_format.MessageToDict(msg)\n\n    return task\n\n\ndef use_secret_as_volume(\n    task: PipelineTask,\n    secret_name: str,\n    mount_path: str,\n) -> PipelineTask:\n    \"\"\"Use a Kubernetes Secret by mounting its data to the task's container as\n    described by the `Kubernetes documentation <https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-files-from-a-pod>`_.\n\n    Args:\n        task: Pipeline task.\n        secret_name: Name of the Secret.\n        mount_path: Path to which to mount the Secret data.\n\n    Returns:\n        Task object with updated secret configuration.\n    \"\"\"\n\n    msg = common.get_existing_kubernetes_config_as_message(task)\n\n    secret_as_vol = pb.SecretAsVolume(\n        secret_name=secret_name,\n        mount_path=mount_path,\n    )\n\n    msg.secret_as_volume.append(secret_as_vol)\n\n    task.platform_config['kubernetes'] = json_format.MessageToDict(msg)\n\n    return task\n", "path": "kubernetes_platform/python/kfp/kubernetes/secret.py"}, {"content": "# Copyright 2024 The Kubeflow Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom typing import Dict\n\nfrom google.protobuf import json_format\nfrom kfp.dsl import PipelineTask\nfrom kfp.kubernetes import common\nfrom kfp.kubernetes import kubernetes_executor_config_pb2 as pb\n\n\ndef use_config_map_as_env(\n    task: PipelineTask,\n    config_map_name: str,\n    config_map_key_to_env: Dict[str, str],\n) -> PipelineTask:\n    \"\"\"Use a Kubernetes ConfigMap as an environment variable as described by the `Kubernetes documentation\n    https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#define-container-environment-variables-using-configmap-data` _.\n\n    Args:\n        task: Pipeline task.\n        config_map_name: Name of the ConfigMap.\n        config_map_key_to_env: Dictionary of ConfigMap key to environment variable name. For example, ``{'foo': 'FOO'}`` sets the value of the ConfigMap's foo field to the environment variable ``FOO``.\n\n    Returns:\n        Task object with updated ConfigMap configuration.\n    \"\"\"\n\n    msg = common.get_existing_kubernetes_config_as_message(task)\n\n    key_to_env = [\n        pb.ConfigMapAsEnv.ConfigMapKeyToEnvMap(\n            config_map_key=config_map_key,\n            env_var=env_var,\n        ) for config_map_key, env_var in config_map_key_to_env.items()\n    ]\n    config_map_as_env = pb.ConfigMapAsEnv(\n        config_map_name=config_map_name,\n        key_to_env=key_to_env,\n    )\n\n    msg.config_map_as_env.append(config_map_as_env)\n\n    task.platform_config['kubernetes'] = json_format.MessageToDict(msg)\n\n    return task\n\n\ndef use_config_map_as_volume(\n    task: PipelineTask,\n    config_map_name: str,\n    mount_path: str,\n) -> PipelineTask:\n    \"\"\"Use a Kubernetes ConfigMap by mounting its data to the task's container as\n    described by the `Kubernetes documentation <https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#add-configmap-data-to-a-volume>`_.\n\n    Args:\n        task: Pipeline task.\n        config_map_name: Name of the ConfigMap.\n        mount_path: Path to which to mount the ConfigMap data.\n\n    Returns:\n        Task object with updated ConfigMap configuration.\n    \"\"\"\n\n    msg = common.get_existing_kubernetes_config_as_message(task)\n\n    config_map_as_vol = pb.ConfigMapAsVolume(\n        config_map_name=config_map_name,\n        mount_path=mount_path,\n    )\n    msg.config_map_as_volume.append(config_map_as_vol)\n\n    task.platform_config['kubernetes'] = json_format.MessageToDict(msg)\n\n    return task\n", "path": "kubernetes_platform/python/kfp/kubernetes/config_map.py"}], "after_files": [{"content": "# Copyright 2023 The Kubeflow Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom typing import Dict\n\nfrom google.protobuf import json_format\nfrom kfp.dsl import PipelineTask\nfrom kfp.kubernetes import common\nfrom kfp.kubernetes import kubernetes_executor_config_pb2 as pb\n\n\ndef use_secret_as_env(\n    task: PipelineTask,\n    secret_name: str,\n    secret_key_to_env: Dict[str, str],\n) -> PipelineTask:\n    \"\"\"Use a Kubernetes Secret as an environment variable as described by the `Kubernetes documentation\n    https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-environment-variables `_.\n\n    Args:\n        task: Pipeline task.\n        secret_name: Name of the Secret.\n        secret_key_to_env: Dictionary of Secret data key to environment variable name. For example, ``{'password': 'PASSWORD'}`` sets the data of the Secret's password field to the environment variable ``PASSWORD``.\n\n    Returns:\n        Task object with updated secret configuration.\n    \"\"\"\n\n    msg = common.get_existing_kubernetes_config_as_message(task)\n\n    key_to_env = [\n        pb.SecretAsEnv.SecretKeyToEnvMap(\n            secret_key=secret_key,\n            env_var=env_var,\n        ) for secret_key, env_var in secret_key_to_env.items()\n    ]\n    secret_as_env = pb.SecretAsEnv(\n        secret_name=secret_name,\n        key_to_env=key_to_env,\n    )\n\n    msg.secret_as_env.append(secret_as_env)\n\n    task.platform_config['kubernetes'] = json_format.MessageToDict(msg)\n\n    return task\n\n\ndef use_secret_as_volume(\n    task: PipelineTask,\n    secret_name: str,\n    mount_path: str,\n    optional: bool = False,\n) -> PipelineTask:\n    \"\"\"Use a Kubernetes Secret by mounting its data to the task's container as\n    described by the `Kubernetes documentation <https://kubernetes.io/docs/concepts/configuration/secret/#using-secrets-as-files-from-a-pod>`_.\n\n    Args:\n        task: Pipeline task.\n        secret_name: Name of the Secret.\n        mount_path: Path to which to mount the Secret data.\n        optional: Optional field specifying whether the Secret must be defined.\n\n    Returns:\n        Task object with updated secret configuration.\n    \"\"\"\n\n    msg = common.get_existing_kubernetes_config_as_message(task)\n\n    secret_as_vol = pb.SecretAsVolume(\n        secret_name=secret_name,\n        mount_path=mount_path,\n        optional=optional,\n    )\n\n    msg.secret_as_volume.append(secret_as_vol)\n\n    task.platform_config['kubernetes'] = json_format.MessageToDict(msg)\n\n    return task\n", "path": "kubernetes_platform/python/kfp/kubernetes/secret.py"}, {"content": "# Copyright 2024 The Kubeflow Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom typing import Dict\n\nfrom google.protobuf import json_format\nfrom kfp.dsl import PipelineTask\nfrom kfp.kubernetes import common\nfrom kfp.kubernetes import kubernetes_executor_config_pb2 as pb\n\n\ndef use_config_map_as_env(\n    task: PipelineTask,\n    config_map_name: str,\n    config_map_key_to_env: Dict[str, str],\n) -> PipelineTask:\n    \"\"\"Use a Kubernetes ConfigMap as an environment variable as described by the `Kubernetes documentation\n    https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#define-container-environment-variables-using-configmap-data` _.\n\n    Args:\n        task: Pipeline task.\n        config_map_name: Name of the ConfigMap.\n        config_map_key_to_env: Dictionary of ConfigMap key to environment variable name. For example, ``{'foo': 'FOO'}`` sets the value of the ConfigMap's foo field to the environment variable ``FOO``.\n\n    Returns:\n        Task object with updated ConfigMap configuration.\n    \"\"\"\n\n    msg = common.get_existing_kubernetes_config_as_message(task)\n\n    key_to_env = [\n        pb.ConfigMapAsEnv.ConfigMapKeyToEnvMap(\n            config_map_key=config_map_key,\n            env_var=env_var,\n        ) for config_map_key, env_var in config_map_key_to_env.items()\n    ]\n    config_map_as_env = pb.ConfigMapAsEnv(\n        config_map_name=config_map_name,\n        key_to_env=key_to_env,\n    )\n\n    msg.config_map_as_env.append(config_map_as_env)\n\n    task.platform_config['kubernetes'] = json_format.MessageToDict(msg)\n\n    return task\n\n\ndef use_config_map_as_volume(\n    task: PipelineTask,\n    config_map_name: str,\n    mount_path: str,\n    optional: bool = False,\n) -> PipelineTask:\n    \"\"\"Use a Kubernetes ConfigMap by mounting its data to the task's container as\n    described by the `Kubernetes documentation <https://kubernetes.io/docs/tasks/configure-pod-container/configure-pod-configmap/#add-configmap-data-to-a-volume>`_.\n\n    Args:\n        task: Pipeline task.\n        config_map_name: Name of the ConfigMap.\n        mount_path: Path to which to mount the ConfigMap data.\n        optional: Optional field specifying whether the ConfigMap must be defined.\n\n    Returns:\n        Task object with updated ConfigMap configuration.\n    \"\"\"\n\n    msg = common.get_existing_kubernetes_config_as_message(task)\n\n    config_map_as_vol = pb.ConfigMapAsVolume(\n        config_map_name=config_map_name,\n        mount_path=mount_path,\n        optional=optional,\n    )\n    msg.config_map_as_volume.append(config_map_as_vol)\n\n    task.platform_config['kubernetes'] = json_format.MessageToDict(msg)\n\n    return task\n", "path": "kubernetes_platform/python/kfp/kubernetes/config_map.py"}]}