keep K8s types out of core.py and repository

Author: wesjdj

bases/renku_data_services/data_api/dependencies.py

@@ -263,7 +263,13 @@ def from_env(cls) -> DependencyManager:
                 default_kubeconfig=default_kubeconfig,
                 cluster_repo=cluster_repo,
                 cache=k8s_db_cache,
-                kinds_to_cache=[AMALTHEA_SESSION_GVK, JUPYTER_SESSION_GVK, BUILD_RUN_GVK, TASK_RUN_GVK, KNATIVE_SERVICE_GVK],
+                kinds_to_cache=[
+                    AMALTHEA_SESSION_GVK,
+                    JUPYTER_SESSION_GVK,
+                    BUILD_RUN_GVK,
+                    TASK_RUN_GVK,
+                    KNATIVE_SERVICE_GVK,
+                ],
             ),
         )
         quota_repo = QuotaRepository(K8sResourceQuotaClient(client), K8sPriorityClassClient(client))
@@ -310,7 +316,13 @@ def from_env(cls) -> DependencyManager:
                             default_kubeconfig=default_kubeconfig,
                             cluster_repo=cluster_repo,
                             cache=k8s_db_cache,
-                            kinds_to_cache=[AMALTHEA_SESSION_GVK, JUPYTER_SESSION_GVK, BUILD_RUN_GVK, TASK_RUN_GVK, KNATIVE_SERVICE_GVK],
+                            kinds_to_cache=[
+                                AMALTHEA_SESSION_GVK,
+                                JUPYTER_SESSION_GVK,
+                                BUILD_RUN_GVK,
+                                TASK_RUN_GVK,
+                                KNATIVE_SERVICE_GVK,
+                            ],
                         ),
                     ),
                     namespace=config.k8s_namespace,

components/renku_data_services/renku_apps/core.py

@@ -1,55 +1,31 @@
 """Business logic for Renku apps."""
 
-from datetime import datetime
-
-from renku_data_services.renku_apps import models
-from renku_data_services.renku_apps.cr_knative_service import Condition
-from renku_data_services.renku_apps.crs import KnativeService
+from renku_data_services.renku_apps.models import App, AppRuntimeState, AppStatus
 from renku_data_services.session.models import SessionLauncher
 
 
-def knative_service_to_app(session_launcher: SessionLauncher, knative_service: KnativeService) -> models.App:
-    """Convert a Knative service to an app."""
-    return models.App(
-        name=knative_service.metadata.name,
-        launcher_id=session_launcher.id,
-        project_id=session_launcher.project_id,
-        status=_project_app_status(knative_service),
-        url=_url(knative_service),
-        started=_started_at(knative_service),
-        image=session_launcher.environment.container_image,
+def build_app(launcher: SessionLauncher, runtime: AppRuntimeState) -> App:
+    """Compose an App from its launcher and the runtime state observed in the cluster."""
+    return App(
+        name=runtime.name,
+        launcher_id=launcher.id,
+        project_id=launcher.project_id,
+        status=app_status_from_ready(runtime.ready_status),
+        url=runtime.url,
+        started=runtime.started_at,
+        image=launcher.environment.container_image,
     )
 
 
-def _url(knative_service: KnativeService) -> str | None:
-    """Get the public URL Knative assigned to the service, or None if it is not yet routed."""
-    if knative_service.status is None:
-        return None
-    return knative_service.status.url
-
-
-def _ready_condition(knative_service: KnativeService) -> Condition | None:
-    """Get the Ready condition from a Knative service, or None if it doesn't exist."""
-    if knative_service.status is None or not knative_service.status.conditions:
-        return None
-    return next((c for c in knative_service.status.conditions if c.type == "Ready"), None)
-
-
-def _started_at(knative_service: KnativeService) -> datetime | None:
-    """Get the time the Knative service became Ready, or None if not yet ready."""
-    ready = _ready_condition(knative_service)
-    if ready is None or ready.status != "True" or ready.lastTransitionTime is None:
-        return None
-    return datetime.fromisoformat(ready.lastTransitionTime)
-
+def app_status_from_ready(ready_status: str | None) -> AppStatus:
+    """Map a Kubernetes Ready-condition status value to an app status.
 
-def _project_app_status(knative_service: KnativeService) -> models.AppStatus:
-    """Convert a Knative service's Ready condition into an app status."""
-    ready = _ready_condition(knative_service)
-    if ready is None:
-        return models.AppStatus("pending")
-    if ready.status == "True":
-        return models.AppStatus("ready")
-    if ready.status == "False":
-        return models.AppStatus("failed")
-    return models.AppStatus("pending")
+    Inputs follow the Kubernetes condition convention: "True", "False",
+    "Unknown", or None when the condition is absent. Unknown and absent
+    both collapse to PENDING.
+    """
+    if ready_status == "True":
+        return AppStatus.READY
+    if ready_status == "False":
+        return AppStatus.FAILED
+    return AppStatus.PENDING

components/renku_data_services/renku_apps/k8s_client.py

@@ -1,6 +1,7 @@
 """K8s client wrapper for Renku apps."""
 
 from collections.abc import AsyncGenerator
+from datetime import datetime
 from typing import Any
 
 from renku_data_services.crc.db import ClusterRepository
@@ -9,7 +10,9 @@
 from renku_data_services.k8s.constants import DEFAULT_K8S_CLUSTER, DUMMY_RENKU_APP_USER_ID, ClusterId
 from renku_data_services.k8s.models import GVK, K8sObjectMeta
 from renku_data_services.project.models import Project
+from renku_data_services.renku_apps.cr_knative_service import Condition
 from renku_data_services.renku_apps.crs import KnativeService
+from renku_data_services.renku_apps.models import AppRuntimeState
 from renku_data_services.session.models import SessionLauncher
 
 KNATIVE_SERVICE_GVK = GVK(kind="Service", group="serving.knative.dev", version="v1")
@@ -36,8 +39,8 @@ def __init__(self, client: K8sClusterClientsPool, cluster_repo: ClusterRepositor
 
     async def create_app_deployment(
         self, session_launcher: SessionLauncher, resource_class: ResourceClass | None, project: Project
-    ) -> KnativeService:
-        """Create a deployment for the given app and return the created Knative Service."""
+    ) -> AppRuntimeState:
+        """Create a deployment for the given app and return its observed runtime state."""
         cluster_id: ClusterId = DEFAULT_K8S_CLUSTER
         cluster = await self.__client.cluster_by_id(cluster_id)
         app_name = _generate_app_name(project)
@@ -52,10 +55,10 @@ async def create_app_deployment(
         created = await self.__client.create(
             meta.with_manifest(manifest.model_dump(exclude_none=True, mode="json")), refresh=True
         )
-        return KnativeService.model_validate(created.manifest)
+        return _extract_runtime_state(KnativeService.model_validate(created.manifest))
 
-    async def get_app_deployment(self, app_name: str) -> KnativeService | None:
-        """Get the deployment for the given app name, or None if it does not exist."""
+    async def get_app_deployment(self, app_name: str) -> AppRuntimeState | None:
+        """Get the runtime state for the given app name, or None if it does not exist."""
         cluster_id: ClusterId = DEFAULT_K8S_CLUSTER
         cluster = await self.__client.cluster_by_id(cluster_id)
         meta = K8sObjectMeta(
@@ -68,21 +71,21 @@ async def get_app_deployment(self, app_name: str) -> KnativeService | None:
         obj = await self.__client.get(meta)
         if obj is None:
             return None
-        return KnativeService.model_validate(obj.manifest)
+        return _extract_runtime_state(KnativeService.model_validate(obj.manifest))
 
-    async def get_app_deployment_for_project(self, project: Project) -> KnativeService | None:
-        """Get the app deployment for the given project, or None if it does not exist."""
+    async def get_app_deployment_for_project(self, project: Project) -> AppRuntimeState | None:
+        """Get the runtime state for the given project's app, or None if it does not exist."""
         return await self.get_app_deployment(_generate_app_name(project))
 
     async def delete_app_deployment(self, app_name: str) -> None:
         """Delete the deployment for the given app name. NOT IMPLEMENTED."""
         raise NotImplementedError("Deleting app deployment is not implemented yet")
 
-    async def list_app_deployments(self) -> AsyncGenerator[KnativeService, None]:
+    async def list_app_deployments(self) -> AsyncGenerator[AppRuntimeState, None]:
         """List all app deployments. NOT IMPLEMENTED."""
         raise NotImplementedError("Listing app deployments is not implemented yet")
 
-    async def update_app_deployment(self, app_name: str, session_launcher: SessionLauncher) -> KnativeService:
+    async def update_app_deployment(self, app_name: str, session_launcher: SessionLauncher) -> AppRuntimeState:
         """Update the deployment for the given app name. NOT IMPLEMENTED."""
         raise NotImplementedError("Updating app deployment is not implemented yet")
 
@@ -142,3 +145,38 @@ def _build_app_deployment_manifest(
             },
         }
     )
+
+
+def _url(knative_service: KnativeService) -> str | None:
+    """Get the public URL Knative assigned to the service, or None if it is not yet routed."""
+    if knative_service.status is None:
+        return None
+    return knative_service.status.url
+
+
+def _ready_condition(knative_service: KnativeService) -> Condition | None:
+    """Get the Ready condition from a Knative service, or None if it doesn't exist."""
+    if knative_service.status is None or not knative_service.status.conditions:
+        return None
+    return next((c for c in knative_service.status.conditions if c.type == "Ready"), None)
+
+
+def _started_at(knative_service: KnativeService) -> datetime | None:
+    """Get the time the Knative service became Ready, or None if not yet ready."""
+    ready = _ready_condition(knative_service)
+    if ready is None or ready.status != "True" or ready.lastTransitionTime is None:
+        return None
+    return datetime.fromisoformat(ready.lastTransitionTime)
+
+
+def _extract_runtime_state(knative_service: KnativeService) -> AppRuntimeState:
+    """Read app runtime state primitives off a Knative Service."""
+    ready = _ready_condition(knative_service)
+    return AppRuntimeState(
+        name=knative_service.metadata.name,
+        launcher_id=knative_service.launcher_id,
+        project_id=knative_service.project_id,
+        ready_status=ready.status if ready is not None else None,
+        url=_url(knative_service),
+        started_at=_started_at(knative_service),
+    )

components/renku_data_services/renku_apps/models.py

@@ -40,3 +40,21 @@ def as_apispec(self) -> apispec.AppResponse:
             started=self.started,
             image=self.image,
         )
+
+
+@dataclass(frozen=True, kw_only=True)
+class AppRuntimeState:
+    """Runtime state of an app deployment, as observed in the cluster.
+
+    Carries the primitives that the K8s adapter extracts from a Knative Service
+    so that domain logic can compose an App without depending on K8s types.
+    The ready_status field holds the raw Kubernetes Ready-condition status value
+    ("True", "False", "Unknown", or None if the condition is absent).
+    """
+
+    name: str
+    launcher_id: ULID
+    project_id: ULID
+    ready_status: str | None
+    url: str | None
+    started_at: datetime | None

components/renku_data_services/renku_apps/repository.py

@@ -9,7 +9,7 @@
 from renku_data_services.crc.db import ResourcePoolRepository
 from renku_data_services.crc.models import ResourceClass
 from renku_data_services.project.db import ProjectRepository
-from renku_data_services.renku_apps.core import knative_service_to_app
+from renku_data_services.renku_apps.core import build_app
 from renku_data_services.renku_apps.k8s_client import RenkuAppsK8sClient
 from renku_data_services.renku_apps.models import App
 from renku_data_services.session.db import SessionRepository
@@ -52,16 +52,16 @@ async def create_app(self, user: base_models.APIUser, launcher_id: ULID) -> App:
         project = await self.project_repo.get_project(user, launcher.project_id)
         if await self.k8s_client.get_app_deployment_for_project(project) is not None:
             raise errors.ConflictError(message=f"An app already exists for project '{launcher.project_id}'.")
-        service = await self.k8s_client.create_app_deployment(launcher, resource_class, project)
-        return knative_service_to_app(launcher, service)
+        runtime_state = await self.k8s_client.create_app_deployment(launcher, resource_class, project)
+        return build_app(launcher, runtime_state)
 
     async def get_app(self, user: base_models.APIUser, app_name: str) -> App:
         """Retrieve an app by its name."""
-        service = await self.k8s_client.get_app_deployment(app_name)
-        if service is None:
+        runtime_state = await self.k8s_client.get_app_deployment(app_name)
+        if runtime_state is None:
             raise errors.MissingResourceError(
                 message=f"App with name '{app_name}' does not exist or you do not have access to it."
             )
 
-        launcher = await self.session_repo.get_launcher(user, service.launcher_id)
-        return knative_service_to_app(launcher, service)
+        launcher = await self.session_repo.get_launcher(user, runtime_state.launcher_id)
+        return build_app(launcher, runtime_state)