From e8cd03b37f8abeb3d77ca0e7b0c2207b4b4febdc Mon Sep 17 00:00:00 2001
From: nate nowack <thrast36@gmail.com>
Date: Thu, 10 Oct 2024 16:35:27 -0500
Subject: [PATCH] correctly await `config.load_kube_config` (#15655)

---
 .../prefect-kubernetes/prefect_kubernetes/credentials.py   | 2 +-
 .../prefect-kubernetes/prefect_kubernetes/jobs.py          | 7 ++++---
 src/integrations/prefect-kubernetes/tests/test_jobs.py     | 6 ++++++
 3 files changed, 11 insertions(+), 4 deletions(-)

diff --git a/src/integrations/prefect-kubernetes/prefect_kubernetes/credentials.py b/src/integrations/prefect-kubernetes/prefect_kubernetes/credentials.py
index 26353045bb70..802edc9b5007 100644
--- a/src/integrations/prefect-kubernetes/prefect_kubernetes/credentials.py
+++ b/src/integrations/prefect-kubernetes/prefect_kubernetes/credentials.py
@@ -186,7 +186,7 @@ async def get_client(
                 config.load_incluster_config(client_configuration=client_configuration)
             except config.ConfigException:
                 # If in-cluster config fails, load the local kubeconfig
-                config.load_kube_config(
+                await config.load_kube_config(
                     client_configuration=client_configuration,
                 )
         async with ApiClient(configuration=client_configuration) as api_client:
diff --git a/src/integrations/prefect-kubernetes/prefect_kubernetes/jobs.py b/src/integrations/prefect-kubernetes/prefect_kubernetes/jobs.py
index 2b3978684823..9536fd39dc82 100644
--- a/src/integrations/prefect-kubernetes/prefect_kubernetes/jobs.py
+++ b/src/integrations/prefect-kubernetes/prefect_kubernetes/jobs.py
@@ -7,7 +7,7 @@
 import yaml
 from kubernetes_asyncio.client.models import V1DeleteOptions, V1Job, V1JobList, V1Status
 from pydantic import Field
-from typing_extensions import Self
+from typing_extensions import Self, TypeAlias
 
 from prefect import task
 from prefect.blocks.abstract import JobBlock, JobRun
@@ -16,7 +16,7 @@
 from prefect_kubernetes.exceptions import KubernetesJobTimeoutError
 from prefect_kubernetes.pods import list_namespaced_pod, read_namespaced_pod_log
 
-KubernetesManifest = Union[Dict, Path, str]
+KubernetesManifest: TypeAlias = Union[Dict, Path, str]
 
 
 @task
@@ -502,7 +502,8 @@ class KubernetesJob(JobBlock):
         examples=[{"pretty": "true"}],
     )
     credentials: KubernetesCredentials = Field(
-        default=..., description="The credentials to configure a client from."
+        default_factory=KubernetesCredentials,
+        description="The credentials to configure a client from.",
     )
     delete_after_completion: bool = Field(
         default=True,
diff --git a/src/integrations/prefect-kubernetes/tests/test_jobs.py b/src/integrations/prefect-kubernetes/tests/test_jobs.py
index b7b089b876a0..e2c9d3aa28d0 100644
--- a/src/integrations/prefect-kubernetes/tests/test_jobs.py
+++ b/src/integrations/prefect-kubernetes/tests/test_jobs.py
@@ -210,3 +210,9 @@ async def test_job_block_wait_never_called_raises(
         ValueError, match="The Kubernetes Job run is not in a completed state"
     ):
         await job_run.fetch_result()
+
+
+async def test_job_block_generates_default_credentials():
+    job_block = KubernetesJob(v1_job=dict(metadata=dict(name="test-job")))
+    assert job_block.credentials is not None
+    assert job_block.credentials.cluster_config is None