Add python client library.

client/python/babylon_async/LICENSE

@@ -0,0 +1,7 @@
+Copyright © 2026 Red Hat
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

client/python/babylon_async/README.md

@@ -0,0 +1 @@
+# Babylon Asynchronous Client Library

client/python/babylon_async/pyproject.toml

@@ -0,0 +1,23 @@
+[build-system]
+requires = ["hatchling >= 1.26"]
+build-backend = "hatchling.build"
+
+[project]
+name = "babylon_async"
+version = "0.0.1"
+authors = [
+  { name="Johnathan Kupferer", email="jkupfere@redhat.com" },
+]
+description = "Babylon asynchronous library package"
+readme = "README.md"
+requires-python = ">=3.9"
+classifiers = [
+    "Programming Language :: Python :: 3",
+    "Operating System :: OS Independent",
+]
+license = "MIT"
+license-files = ["LICENSE"]
+
+[project.urls]
+Homepage = "https://github.com/redhat-cop/babylon"
+Issues = "https://github.com/redhat-cop/babylon/issues"

client/python/babylon_async/src/babylon_async/__init__.py

@@ -0,0 +1,9 @@
+from .client import BabylonClient
+from .exceptions import BabylonApiException
+
+from .anarchyrun import AnarchyRun
+from .anarchysubject import AnarchySubject
+from .catalogitem import CatalogItem
+from .tenantclusterpool import TenantClusterPool
+from .resourceclaim import ResourceClaim
+from .resourcepool import ResourcePool

client/python/babylon_async/src/babylon_async/anarchygovernor.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+from typing import List, Mapping
+
+from .k8s_object import K8sObject
+
+class AnarchyGovernor(K8sObject):
+    api_group = "anarchy.gpte.redhat.com"
+    api_version = "v1"
+    kind = "AnarchyGovernor"
+    plural = "anarchygovernors"
+    api_group_version = f"{api_group}/{api_version}"
+
+    def get_sandboxes(self) -> List[Mapping]:
+        return [
+            SandboxSpec(item)
+            for item in self.spec.job_vars['__meta__'].get('sandboxes', [])
+        ]
+
+    @property
+    def job_vars(self) -> Mapping:
+        return self.spec.job_vars
+
+    @property
+    def spec(self) -> AnarchyGovernorSpec:
+        return AnarchyGovernorSpec(self._definition)
+
+    @property
+    def vars(self) -> Mapping:
+        return self.spec.vars
+
+class AnarchyGovernorSpec:
+    def __init__(self, definition):
+        self._definition = definition
+
+    @property
+    def job_vars(self) -> Mapping:
+        return self._definition.get('vars', {}).get('job_vars', {})
+
+    @property
+    def vars(self) -> Mapping:
+        return self._definition.get('vars', {})
+
+class SandboxSpec:
+    def __init__(self, definition):
+        self._definition = definition
+
+    @property
+    def alias(self) -> str|None:
+        return self._definition.get('alias')
+
+    @property
+    def annotations(self) -> Mapping[str, str]|None:
+        return self._definition.get('annotations')
+
+    @property
+    def cloud_selector(self) -> Mapping[str, str]|None:
+        return self._definition.get('cloud_selector')
+
+    @property
+    def kind(self) -> str:
+        return self._definition['kind']
+
+    @property
+    def namespace_suffix(self) -> str|None:
+        return self._definition.get('namespace_suffix')
+
+    @property
+    def quota(self) -> Mapping[str, str]|None:
+        return self._definition.get('quota')
+
+    @property
+    def var(self) -> str|None:
+        return self._definition.get('var')

client/python/babylon_async/src/babylon_async/anarchyrun.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from kubernetes_asyncio.client import (
+    ApiException as KubernetesApiException
+)
+
+from .k8s_object import K8sObject
+
+class AnarchyRun(K8sObject):
+    api_group = "anarchy.gpte.redhat.com"
+    api_version = "v1"
+    kind = "AnarchyRun"
+    plural = "anarchyruns"
+    api_group_version = f"{api_group}/{api_version}"
+
+    @property
+    def is_active(self) -> bool:
+        value = self.metadata.labels.get('anarchy.gpte.redhat.com/runner')
+        if value.startswith('anarchy-runner-'):
+            return True
+        return self.state in {'failed', 'pending'}
+
+    @property
+    def is_finished(self) -> bool:
+        return self.state in {'canceled', 'successful'}
+
+    @property
+    def is_queued(self) -> bool:
+        return self.state == 'queued'
+
+    @property
+    def runner_pod_name(self) -> str|None:
+        value = self.metadata.labels.get('anarchy.gpte.redhat.com/runner')
+        if value.startswith('anarchy-runner-'):
+            return value
+        return None
+
+    @property
+    def spec(self) -> AnarchyRunSpec:
+        return AnarchyRunSpec(self._definition['spec'])
+
+    @property
+    def state(self) -> str|None:
+        value = self.metadata.labels.get('anarchy.gpte.redhat.com/runner')
+        if value in {'canceled', 'failed', 'pending', 'queued', 'successful'}:
+            return value
+        if value.startswith('anarchy-runner-'):
+            return 'running'
+        return None
+
+    @property
+    def status(self) -> AnarchyRunStatus|None:
+        if 'status' not in self._definition:
+            return None
+        return AnarchyRunStatus(self._definition['status'])
+
+    async def check_runner_pod_exists(self) -> bool:
+        if self.runner_pod_name is None:
+            return True
+        try:
+            await self.client.core_v1_api.read_namespaced_pod(
+                name=self.runner_pod_name,
+                namespace=self.metadata.namespace,
+            )
+            return True
+        except KubernetesApiException as exception:
+            if exception.status == 404:
+                return False
+            raise
+
+    async def set_runner_state(self, value) -> None:
+        await self.patch({
+            "metadata": {
+                "labels": {
+                    "anarchy.gpte.redhat.com/runner": value,
+                }
+            }
+        })
+
+class AnarchyRunSpec:
+    def __init__(self, definition):
+        self._definition = definition
+
+class AnarchyRunStatus:
+    def __init__(self, definition):
+        self._definition = definition
+
+    @property
+    def retry_after_datetime(self) -> datetime|None:
+        ts = self.retry_after_timestamp
+        if ts is None:
+            return None
+        return datetime.strptime(ts, '%Y-%m-%dT%H:%M:%S%z')
+
+    @property
+    def retry_after_timestamp(self) -> str|None:
+        return self._definition.get('retryAfter')
+
+    @property
+    def run_post_datetime(self) -> datetime|None:
+        ts = self.run_post_timestamp
+        if ts is None:
+            return None
+        return datetime.strptime(ts, '%Y-%m-%dT%H:%M:%S%z')
+
+    @property
+    def run_post_timestamp(self) -> str|None:
+        return self._definition.get('runPostTimestamp')

client/python/babylon_async/src/babylon_async/anarchysubject.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+from typing import Generator
+
+from .k8s_object import K8sObject
+from .resourcereference import ResourceReference
+
+from .anarchyrun import AnarchyRun
+
+class AnarchySubject(K8sObject):
+    api_group = "anarchy.gpte.redhat.com"
+    api_version = "v1"
+    kind = "AnarchySubject"
+    plural = "anarchysubjects"
+    api_group_version = f"{api_group}/{api_version}"
+
+    @property
+    def spec(self) -> AnarchySubjectSpec:
+        return AnarchySubjectSpec(self._definition['spec'])
+
+    @property
+    def status(self) -> AnarchySubjectStatus|None:
+        if 'status' not in self._definition:
+            return None
+        return AnarchySubjectStatus(self._definition['status'])
+
+    async def get_anarchy_run(self, name:str) -> AnarchyRun:
+        return await self.client.get_anarchy_run(
+            name=name,
+            namespace=self.namespace,
+        )
+
+    async def list_anarchy_runs(self) -> Generator[AnarchyRun, None, None]:
+        async for anarchy_run in self.client.list_anarchy_runs_for_anarchy_subject(self):
+            yield anarchy_run
+
+class AnarchySubjectSpec:
+    def __init__(self, definition):
+        self._definition = definition
+
+class AnarchySubjectStatus:
+    def __init__(self, definition):
+        self._definition = definition
+
+    @property
+    def runs(self) -> AnarchySubjectStatusRuns:
+        return AnarchySubjectStatusRuns(self._definition.get('runs', {}))
+
+class AnarchySubjectStatusRuns:
+    def __init__(self, definition):
+        self._definition = definition
+
+    @property
+    def active(self) -> list[ResourceReference]:
+        return [
+            ResourceReference(item) for item in self._definition.get('active', [])
+        ]