Fixes env handling to allow hard coding of environment variable in base job template

CHANGELOG.md

@@ -9,8 +9,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Handling for spot instance eviction - [#85](https://github.com/PrefectHQ/prefect-kubernetes/pull/85)
-
 ### Changed
 
 ### Deprecated
@@ -21,6 +19,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security
 
+## 0.2.12
+
+Released September 18th, 2023.
+
+### Added
+
+- Handling for spot instance eviction - [#85](https://github.com/PrefectHQ/prefect-kubernetes/pull/85)
+- Logging recent job events when pod scheduling fails - [#88](https://github.com/PrefectHQ/prefect-kubernetes/pull/88)
+- Event logging for pod events - [#91](https://github.com/PrefectHQ/prefect-kubernetes/pull/91)
+
+
+### Fixed
+
+- `env` handling to allow hard coding of environment variable in base job template - [#94](https://github.com/PrefectHQ/prefect-kubernetes/pull/94)
+
 ## 0.2.11
 
 Released July 20th, 2023.

prefect_kubernetes/worker.py

@@ -94,7 +94,6 @@
 
 import anyio.abc
 from prefect.blocks.kubernetes import KubernetesClusterConfig
-from prefect.docker import get_prefect_image_name
 from prefect.exceptions import (
     InfrastructureError,
     InfrastructureNotAvailable,
@@ -103,6 +102,7 @@
 from prefect.server.schemas.core import Flow
 from prefect.server.schemas.responses import DeploymentResponse
 from prefect.utilities.asyncutils import run_sync_in_worker_thread
+from prefect.utilities.dockerutils import get_prefect_image_name
 from prefect.utilities.importtools import lazy_import
 from prefect.utilities.pydantic import JsonPatch
 from prefect.utilities.templating import find_placeholders
@@ -307,16 +307,37 @@ def prepare_for_flow_run(
         super().prepare_for_flow_run(flow_run, deployment, flow)
         # Update configuration env and job manifest env
         self._update_prefect_api_url_if_local_server()
-        self.job_manifest["spec"]["template"]["spec"]["containers"][0]["env"] = [
-            {"name": k, "value": v} for k, v in self.env.items()
-        ]
+        self._populate_env_in_manifest()
         # Update labels in job manifest
         self._slugify_labels()
         # Add defaults to job manifest if necessary
         self._populate_image_if_not_present()
         self._populate_command_if_not_present()
         self._populate_generate_name_if_not_present()
 
+    def _populate_env_in_manifest(self):
+        transformed_env = [{"name": k, "value": v} for k, v in self.env.items()]
+
+        template_env = self.job_manifest["spec"]["template"]["spec"]["containers"][
+            0
+        ].get("env")
+
+        # If user has removed `{{ env }}` placeholder and hard coded a value for `env`,
+        # we need to prepend our environment variables to the list to ensure Prefect
+        # setting propagation.
+        if isinstance(template_env, list):
+            self.job_manifest["spec"]["template"]["spec"]["containers"][0]["env"] = [
+                *transformed_env,
+                *template_env,
+            ]
+        # Current templating adds `env` as a dict when the kubernetes manifest requires
+        # a list of dicts. Might be able to improve this in the future with a better
+        # default `env` value and better typing.
+        else:
+            self.job_manifest["spec"]["template"]["spec"]["containers"][0][
+                "env"
+            ] = transformed_env
+
     def _update_prefect_api_url_if_local_server(self):
         """If the API URL has been set by the base environment rather than the by the
         user, update the value to ensure connectivity when using a bridge network by

requirements.txt

@@ -1,2 +1,2 @@
-prefect>=2.10.9
+prefect>=2.13.0
 kubernetes >= 24.2.0