client: add Kubernetes-aware retry helpers

kubernetes/aio/utils/__init__.py

@@ -18,3 +18,10 @@
     FailToCreateError, create_from_dict, create_from_yaml,
     create_from_yaml_single_item,
 )
+from .retry import (Backoff, DEFAULT_BACKOFF, DEFAULT_RETRY,
+                    DEFAULT_RETRY_AFTER_BACKOFF, async_on_error,
+                    async_on_retry_after_error, async_retry_on_conflict,
+                    is_conflict, is_retry_after_response,
+                    is_too_many_requests, on_error, on_retry_after_error,
+                    retry_after_backoff, retry_after_max_retries,
+                    retry_on_conflict, retry_after_seconds)

kubernetes/aio/utils/retry.py

@@ -0,0 +1 @@
+../../utils/retry.py

kubernetes/aio/utils/retry_test.py

@@ -0,0 +1,72 @@
+# Copyright 2026 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+from kubernetes.aio.utils.retry import (
+    Backoff,
+    DEFAULT_RETRY,
+    async_retry_on_conflict,
+    retry_after_seconds,
+)
+
+
+class FakeError(Exception):
+
+    def __init__(self, status, headers=None):
+        super().__init__("status {0}".format(status))
+        self.status = status
+        self.headers = headers or {}
+
+
+class AioRetryTest(unittest.IsolatedAsyncioTestCase):
+
+    def test_default_retry_matches_client_go(self):
+        self.assertEqual(
+            DEFAULT_RETRY,
+            Backoff(steps=5, duration=0.01, factor=1.0, jitter=0.1),
+        )
+
+    def test_retry_after_seconds_parses_delay_seconds(self):
+        error = FakeError(429, {"Retry-After": "7"})
+
+        self.assertEqual(retry_after_seconds(error), 7.0)
+
+    async def test_async_retry_on_conflict_retries(self):
+        attempts = []
+        sleeps = []
+
+        async def fn():
+            attempts.append(1)
+            if len(attempts) < 3:
+                raise FakeError(409)
+            return "updated"
+
+        async def sleep(delay):
+            sleeps.append(delay)
+
+        result = await async_retry_on_conflict(
+            fn,
+            backoff=Backoff(steps=3, duration=1.0, factor=1.0),
+            sleep_func=sleep,
+            random_func=lambda: 0.0,
+        )
+
+        self.assertEqual(result, "updated")
+        self.assertEqual(len(attempts), 3)
+        self.assertEqual(sleeps, [1.0, 1.0])
+
+
+if __name__ == "__main__":
+    unittest.main()

kubernetes/utils/__init__.py

@@ -19,3 +19,10 @@
 from .duration import parse_duration
 from .metrics import (get_nodes_metrics, get_pods_metrics,
                       get_pods_metrics_in_all_namespaces)
+from .retry import (Backoff, DEFAULT_BACKOFF, DEFAULT_RETRY,
+                    DEFAULT_RETRY_AFTER_BACKOFF, async_on_error,
+                    async_on_retry_after_error, async_retry_on_conflict,
+                    is_conflict, is_retry_after_response,
+                    is_too_many_requests, on_error, on_retry_after_error,
+                    retry_after_backoff, retry_after_max_retries,
+                    retry_on_conflict, retry_after_seconds)