[CDF-27549] Add AppIO resource loader for custom app deployment via App Hosting

cognite_toolkit/_cdf_tk/client/_toolkit_client.py

@@ -10,6 +10,8 @@
 from .api.agents import AgentsAPI
 from .api.alerts import AlertsAPI
 from .api.annotations import AnnotationsAPI
+from .api.app_versions import AppVersionsAPI
+from .api.apps import AppsAPI
 from .api.assets import AssetsAPI
 from .api.canvas import IndustrialCanvasAPI
 from .api.cognite_files import CogniteFilesAPI
@@ -64,6 +66,8 @@ class ToolAPI:
     def __init__(self, http_client: HTTPClient, console: Console) -> None:
         self.http_client = http_client
         self.agents = AgentsAPI(http_client)
+        self.apps = AppsAPI(http_client)
+        self.app_versions = AppVersionsAPI(http_client)
         self.annotations = AnnotationsAPI(http_client)
         self.assets = AssetsAPI(http_client)
         self.cognite_files = CogniteFilesAPI(http_client)

cognite_toolkit/_cdf_tk/client/api/app_versions.py

@@ -0,0 +1,119 @@
+"""AppVersionsAPI: Version management for custom apps via the CDF App Hosting API."""
+
+import json
+from collections.abc import Iterable, Sequence
+
+from cognite_toolkit._cdf_tk.client.http_client import HTTPClient, RequestMessage, ToolkitAPIError
+from cognite_toolkit._cdf_tk.client.http_client._data_classes import FailedResponse, SuccessResponse
+from cognite_toolkit._cdf_tk.client.identifiers import AppVersionId
+from cognite_toolkit._cdf_tk.client.resource_classes.app_version import AppVersionResponse
+
+
+class AppVersionsAPI:
+    """Client for the CDF App Hosting Versions API (POST /apphosting/apps/{externalId}/versions/...)."""
+
+    def __init__(self, http_client: HTTPClient) -> None:
+        self._http_client = http_client
+
+    def _url(self, path: str) -> str:
+        return self._http_client.config.create_api_url(path)
+
+    def upload(
+        self,
+        external_id: str,
+        version: str,
+        entrypoint: str,
+        zip_bytes: bytes,
+    ) -> None:
+        """POST /apphosting/apps/{externalId}/versions — multipart/form-data upload of the zipped app."""
+        result = self._http_client.request_multipart_retries(
+            url=self._url(f"/apphosting/apps/{external_id}/versions"),
+            files={"file": ("app.zip", zip_bytes, "application/zip")},
+            form_fields={"version": version, "entryPath": entrypoint},
+        )
+        if isinstance(result, FailedResponse):
+            raise ToolkitAPIError(message=result.body, code=result.status_code)
+
+    def update(self, external_id: str, version: str, patch: dict) -> None:
+        """POST /apphosting/apps/{externalId}/versions/update — apply a lifecycle/alias patch to a version."""
+        request = RequestMessage(
+            endpoint_url=self._url(f"/apphosting/apps/{external_id}/versions/update"),
+            method="POST",
+            body_content={"items": [{"version": version, "update": patch}]},
+        )
+        self._http_client.request_single_retries(request).get_success_or_raise(request)
+
+    def retrieve(self, items: Sequence[AppVersionId], ignore_unknown_ids: bool = False) -> list[AppVersionResponse]:
+        """GET /apphosting/apps/{externalId}/versions/{version} — retrieve version metadata."""
+        results: list[AppVersionResponse] = []
+        for item in items:
+            request = RequestMessage(
+                endpoint_url=self._url(f"/apphosting/apps/{item.app_external_id}/versions/{item.version}"),
+                method="GET",
+            )
+            result = self._http_client.request_single_retries(request)
+            if not isinstance(result, SuccessResponse):
+                if isinstance(result, FailedResponse) and result.status_code in (400, 404) and ignore_unknown_ids:
+                    continue
+                result.get_success_or_raise(request)
+                continue
+            data = json.loads(result.body)
+            results.append(AppVersionResponse(
+                app_external_id=data.get("appExternalId", item.app_external_id),
+                version=data.get("version", item.version),
+                lifecycle_state=data.get("lifecycleState", "DRAFT"),
+                alias=data.get("alias"),
+                entrypoint=data.get("entrypoint", "index.html"),
+            ))
+        return results
+
+    def iterate(self, limit: int | None = 100) -> Iterable[list[AppVersionResponse]]:
+        """POST /apphosting/versions/list — paginated list of all versions across all apps."""
+        cursor: str | None = None
+        page_limit = min(limit, 1000) if limit is not None else 1000
+        fetched = 0
+        while True:
+            body: dict = {"limit": page_limit}
+            if cursor:
+                body["cursor"] = cursor
+            request = RequestMessage(
+                endpoint_url=self._url("/apphosting/versions/list"),
+                method="POST",
+                body_content=body,
+            )
+            result = self._http_client.request_single_retries(request)
+            if not isinstance(result, SuccessResponse):
+                result.get_success_or_raise(request)
+                break
+
+            data = json.loads(result.body)
+            page_items = [
+                AppVersionResponse(
+                    app_external_id=item["appExternalId"],
+                    version=item["version"],
+                    lifecycle_state=item.get("lifecycleState", "DRAFT"),
+                    alias=item.get("alias"),
+                    entrypoint=item.get("entrypoint", "index.html"),
+                )
+                for item in data.get("items", [])
+            ]
+            if page_items:
+                yield page_items
+                fetched += len(page_items)
+
+            cursor = data.get("nextCursor")
+            if not cursor or (limit is not None and fetched >= limit):
+                break
+
+    def delete(self, versions: Sequence[AppVersionId]) -> None:
+        """POST /apphosting/apps/{externalId}/versions/delete — delete specific versions, grouped by app."""
+        by_app: dict[str, list[AppVersionId]] = {}
+        for version_id in versions:
+            by_app.setdefault(version_id.app_external_id, []).append(version_id)
+        for app_external_id, app_versions in by_app.items():
+            request = RequestMessage(
+                endpoint_url=self._url(f"/apphosting/apps/{app_external_id}/versions/delete"),
+                method="POST",
+                body_content={"items": [{"version": v.version} for v in app_versions]},
+            )
+            self._http_client.request_single_retries(request).get_success_or_raise(request)

cognite_toolkit/_cdf_tk/client/api/apps.py

@@ -0,0 +1,41 @@
+"""AppsAPI: Custom apps deployed via the CDF App Hosting API."""
+
+from collections.abc import Sequence
+
+from cognite_toolkit._cdf_tk.client.cdf_client import CDFResourceAPI, PagedResponse
+from cognite_toolkit._cdf_tk.client.cdf_client.api import Endpoint
+from cognite_toolkit._cdf_tk.client.http_client import HTTPClient, ItemsSuccessResponse, RequestMessage, SuccessResponse
+from cognite_toolkit._cdf_tk.client.http_client._data_classes import FailedResponse
+from cognite_toolkit._cdf_tk.client.resource_classes.app import AppRequest, AppResponse
+
+
+class AppsAPI(CDFResourceAPI[AppResponse]):
+    """Client for the CDF App Hosting API (/apphosting/apps)."""
+
+    def __init__(self, http_client: HTTPClient) -> None:
+        super().__init__(
+            http_client=http_client,
+            method_endpoint_map={
+                "create": Endpoint(method="POST", path="/apphosting/apps", item_limit=1),
+            },
+        )
+
+    def _validate_page_response(
+        self, response: SuccessResponse | ItemsSuccessResponse
+    ) -> PagedResponse[AppResponse]:
+        return PagedResponse[AppResponse].model_validate_json(response.body)
+
+    def create(self, items: Sequence[AppRequest]) -> list[AppResponse]:
+        """POST /apphosting/apps — create apps."""
+        return self._request_item_response(items, "create")
+
+    def retrieve(self, external_id: str) -> AppResponse | None:
+        """GET /apphosting/apps/{externalId} — fetch app-level metadata (name, description)."""
+        request = RequestMessage(
+            endpoint_url=self._make_url(f"/apphosting/apps/{external_id}"),
+            method="GET",
+        )
+        result = self._http_client.request_single_retries(request)
+        if isinstance(result, FailedResponse) and result.status_code == 404:
+            return None
+        return AppResponse.model_validate_json(result.get_success_or_raise(request).body)

cognite_toolkit/_cdf_tk/client/http_client/_client.py

@@ -259,38 +259,26 @@ def _handle_error_single(self, e: Exception, request: RequestMessage) -> Request
 
             return FailedRequest(error=error_msg)
 
-    def request_raw_retries(
+    def _execute_raw_with_retries(
         self,
-        method: Literal["GET", "POST", "PUT", "DELETE"],
+        method: str,
         url: str,
-        content: bytes | Iterable[bytes],
+        max_retries: int,
+        content: bytes | Iterable[bytes] | None = None,
+        files: dict[str, tuple[str, bytes, str]] | None = None,
+        data: dict[str, str] | None = None,
         headers: dict[str, str] | None = None,
-        max_retries: int | None = None,
     ) -> SuccessResponse | FailedResponse:
-        """Send a raw HTTP request with retry logic but without authentication headers.
-
-        This is useful for uploading to signed URLs (e.g., GCS signed URLs) where
-        authentication is embedded in the URL and adding auth headers would cause errors.
-
-        Args:
-            method: HTTP method to use.
-            url: The URL to send the request to.
-            content: The content to send. Can be bytes or an iterable of bytes for streaming.
-            headers: Optional headers to include in the request.
-            max_retries: Maximum number of retries. Defaults to the client's max_retries setting.
-
-        Returns:
-            HTTPResult: The result of the HTTP request, either SuccessResponse or FailedResponse.
-        """
-        retries = max_retries if max_retries is not None else self._max_retries
         attempt = 0
         last_error_code: int = -1
-        while attempt <= retries:
+        while attempt <= max_retries:
             try:
                 response = self.session.request(
                     method=method,
                     url=url,
                     content=content,
+                    files=files,
+                    data=data,
                     headers=headers,
                     follow_redirects=False,
                 )
@@ -301,22 +289,16 @@ def request_raw_retries(
                         content=response.content,
                     )
                 last_error_code = response.status_code
-                # Check if we should retry based on status code
                 if response.status_code in self._retry_status_codes:
                     retry_after = self._get_retry_after_in_header(response)
-                    if retry_after is not None:
-                        time.sleep(retry_after)
-                    else:
-                        time.sleep(self._backoff_time(attempt))
+                    time.sleep(retry_after if retry_after is not None else self._backoff_time(attempt))
                     attempt += 1
                     continue
-                # Non-retryable error
                 return FailedResponse(
                     status_code=response.status_code,
                     body=response.text,
-                    error=ErrorDetails(code=response.status_code, message=response.text),
+                    error=ErrorDetails.from_response(response),
                 )
-
             except (
                 httpx.ReadTimeout,
                 httpx.TimeoutException,
@@ -325,22 +307,64 @@ def request_raw_retries(
                 httpx.ConnectTimeout,
             ) as e:
                 attempt += 1
-                if attempt <= retries:
+                if attempt <= max_retries:
                     time.sleep(self._backoff_time(attempt))
                     continue
                 return FailedResponse(
                     status_code=last_error_code,
                     body=f"Request failed after {attempt} attempts: {e!s}",
-                    error=ErrorDetails(code=last_error_code, message=f"Request failed after {attempt} attempts: {e!s}"),
+                    error=ErrorDetails(
+                        code=last_error_code,
+                        message=f"Request failed after {attempt} attempts: {e!s}",
+                    ),
                 )
-
-        # Should not reach here, but just in case
         return FailedResponse(
             status_code=last_error_code,
             body=f"Request failed after {attempt} attempts.",
             error=ErrorDetails(code=last_error_code, message=f"Request failed after {attempt} attempts."),
         )
 
+    def request_raw_retries(
+        self,
+        method: Literal["GET", "POST", "PUT", "DELETE"],
+        url: str,
+        content: bytes | Iterable[bytes],
+        headers: dict[str, str] | None = None,
+        max_retries: int | None = None,
+    ) -> SuccessResponse | FailedResponse:
+        """Send a raw HTTP request with retry logic but without authentication headers.
+
+        This is useful for uploading to signed URLs (e.g., GCS signed URLs) where
+        authentication is embedded in the URL and adding auth headers would cause errors.
+        """
+        retries = max_retries if max_retries is not None else self._max_retries
+        return self._execute_raw_with_retries(method, url, retries, content=content, headers=headers)
+
+    def request_multipart_retries(
+        self,
+        url: str,
+        files: dict[str, tuple[str, bytes, str]],
+        form_fields: dict[str, str],
+        api_version: str | None = None,
+    ) -> SuccessResponse | FailedResponse:
+        """POST multipart/form-data to a CDF endpoint with auth headers and retry logic.
+
+        Uses httpx's native multipart encoder — Content-Type (with boundary) and
+        Content-Length are set automatically. Unlike request_raw_retries, CDF auth
+        headers are included because this method targets CDF endpoints, not signed URLs.
+        """
+        auth_name, auth_value = self.config.credentials.authorization_header()
+        # Content-Type is intentionally absent — httpx sets it from the multipart body (including boundary).
+        headers: dict[str, str] = {
+            "User-Agent": f"httpx/{httpx.__version__} {get_user_agent()}",
+            auth_name: auth_value,
+            "accept": "application/json",
+            "x-cdp-sdk": f"CogniteToolkit:{get_current_toolkit_version()}",
+            "x-cdp-app": self.config.client_name,
+            "cdf-version": api_version or self.config.api_subversion,
+        }
+        return self._execute_raw_with_retries("POST", url, self._max_retries, files=files, data=form_fields, headers=headers)
+
     def request_items(self, message: ItemsRequest) -> Sequence[ItemsRequest | ItemsResultMessage]:
         """Send an HTTP request with multiple items and return the response.
 

cognite_toolkit/_cdf_tk/client/resource_classes/app.py

@@ -1,40 +1,22 @@
-from typing import Any, Literal
-
 from cognite_toolkit._cdf_tk.client._resource_base import BaseModelObject, RequestResource, ResponseResource
-from cognite_toolkit._cdf_tk.client.identifiers import AppVersionId
+from cognite_toolkit._cdf_tk.client.identifiers import ExternalId
 
 
 class App(BaseModelObject):
     external_id: str
-    version: str
     name: str
     description: str | None = None
-    lifecycle_state: Literal["DRAFT", "PUBLISHED", "DEPRECATED", "ARCHIVED"] = "PUBLISHED"
-    alias: Literal["ACTIVE", "PREVIEW"] | None = None
-    entrypoint: str = "index.html"
 
 
 class AppRequest(App, RequestResource):
-    """Local representation of a custom app version for App Hosting deployment."""
-
-    def as_id(self) -> AppVersionId:
-        return AppVersionId(app_external_id=self.external_id, version=self.version)
+    """Write resource for POST /apphosting/apps."""
 
-    def dump(
-        self, camel_case: bool = True, exclude_extra: bool = False, context: Literal["api", "toolkit"] = "api"
-    ) -> dict[str, Any]:
-        if context == "toolkit":
-            return super().dump(camel_case=camel_case, exclude_extra=exclude_extra)
-        # Body for POST /apphosting/apps (ensure-app call)
-        key = "externalId" if camel_case else "external_id"
-        body: dict[str, Any] = {key: self.external_id, "name": self.name}
-        if self.description is not None:
-            body["description"] = self.description
-        return body
+    def as_id(self) -> ExternalId:
+        return ExternalId(external_id=self.external_id)
 
 
 class AppResponse(App, ResponseResource[AppRequest]):
-    """Response from App Hosting after a successful deploy."""
+    """Response from GET/POST /apphosting/apps."""
 
     @classmethod
     def request_cls(cls) -> type[AppRequest]: