k8s: patch get_kubeconfig to work around deserialization issue

README.md

@@ -216,31 +216,6 @@ docker run -it --rm --name pydo -v $PWD/tests:/tests pydo:dev pytest tests/mocke
 
 > This selection lists the known issues of the client generator.
 
-#### `kubernetes.get_kubeconfig` Does not serialize response content
-
-In the generated Python client, calling client.kubernetes.get_kubeconfig(cluster_id) raises a deserialization error when the response content-type is application/yaml. This occurs because the generator does not correctly handle YAML responses. We should investigate whether the OpenAPI spec or generator configuration can be adjusted to support this content-type. If not, the issue should be reported upstream to improve YAML support in client generation.
-
-Workaround (with std lib httplib):
-
-```python
-from http.client import HTTPSConnection
-
-conn = HTTPSConnection('api.digitalocean.com')
-conn.request(
-    'GET',
-    f'/v2/kubernetes/clusters/{cluster_id}/kubeconfig',
-    headers={'Authorization': f'Bearer {os.environ["DIGITALOCEAN_TOKEN"]}'}
-)
-response = conn.getresponse()
-
-if response.getcode() > 400:
-    msg = 'Unable to get kubeconfig'
-    raise RuntimeError(msg)
-
-kube_config =  response.read().decode('utf-8')
-conn.close()
-```
-
 #### `invoices.get_pdf_by_uuid(invoice_uuid=invoice_uuid_param)` Does not return PDF
 
 In the generated python client, when calling `invoices.get_pdf_by_uuid`, the response returns a Iterator[bytes] that does not format correctly into a PDF.

src/pydo/aio/operations/_patch.py

@@ -6,15 +6,116 @@
 
 Follow our quickstart for examples: https://aka.ms/azsdk/python/dpcodegen/python/customize
 """
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, cast
+
+from azure.core.exceptions import (
+    ClientAuthenticationError,
+    HttpResponseError,
+    ResourceExistsError,
+    ResourceNotFoundError,
+    ResourceNotModifiedError,
+    map_error,
+)
+from azure.core.pipeline import PipelineResponse
+from azure.core.tracing.decorator_async import distributed_trace_async
+
+from ._operations import (
+    KubernetesOperations as _KubernetesOperations,
+    build_kubernetes_get_kubeconfig_request,
+)
 
 if TYPE_CHECKING:
     # pylint: disable=unused-import,ungrouped-imports
-    from typing import List
+    from typing import MutableMapping, Type
+
+__all__ = ("KubernetesOperations",)
+
+
+# Override: generated client expects JSON but this endpoint returns application/yaml;
+# we return the response body as str instead of deserializing.
+class KubernetesOperations(_KubernetesOperations):
+    """Kubernetes operations."""
+
+    @distributed_trace_async
+    async def get_kubeconfig(
+        self, cluster_id: str, *, expiry_seconds: int = 0, **kwargs: Any
+    ) -> str:
+        """Retrieve the kubeconfig for a Kubernetes Cluster.
+
+        This endpoint returns a kubeconfig file in YAML format. It can be used to
+        connect to and administer the cluster using the Kubernetes command line tool,
+        ``kubectl``, or other programs supporting kubeconfig files (e.g., client libraries).
+
+        The resulting kubeconfig file uses token-based authentication for clusters
+        supporting it, and certificate-based authentication otherwise. For a list of
+        supported versions and more information, see "How to Connect to a DigitalOcean
+        Kubernetes Cluster"
+        https://docs.digitalocean.com/products/kubernetes/how-to/connect-to-cluster/
+
+        Clusters supporting token-based authentication may define an expiration by
+        passing a duration in seconds as a query parameter (expiry_seconds).
+        If not set or 0, then the token will have a 7 day expiry. The query parameter
+        has no impact in certificate-based authentication.
+
+        Kubernetes Roles granted to a user with a token-based kubeconfig are derived from that user's
+        DigitalOcean role. Custom roles require additional configuration by a cluster administrator.
+
+        :param cluster_id: A unique ID that can be used to reference a Kubernetes cluster. Required.
+        :type cluster_id: str
+        :keyword expiry_seconds: The duration in seconds that the returned Kubernetes credentials will
+         be valid. If not set or 0, the credentials will have a 7 day expiry. Default value is 0.
+        :paramtype expiry_seconds: int
+        :return: The kubeconfig file contents as a string (YAML).
+        :rtype: str
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: "MutableMapping[int, Type[HttpResponseError]]" = {
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+            401: cast(
+                "Type[HttpResponseError]",
+                lambda response: ClientAuthenticationError(response=response),
+            ),
+            429: HttpResponseError,
+            500: HttpResponseError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        _request = build_kubernetes_get_kubeconfig_request(
+            cluster_id=cluster_id,
+            expiry_seconds=expiry_seconds,
+            headers=_headers,
+            params=_params,
+        )
+        _request.url = self._client.format_url(_request.url)
+
+        # stream=True so the pipeline's content policy skips deserialization (API returns YAML, not JSON)
+        pipeline_response: PipelineResponse = (
+            await self._client._pipeline.run(  # pylint: disable=protected-access
+                _request, stream=True, **kwargs
+            )
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            await response.read()
+            map_error(
+                status_code=response.status_code,
+                response=response,
+                error_map=error_map,
+            )
+            raise HttpResponseError(response=response)
 
-__all__ = (
-    []
-)  # type: List[str]  # Add all objects you want publicly available to users at this package level
+        if hasattr(response, "read"):
+            body = await response.read()
+        else:
+            body = response.content
+        return body.decode("utf-8") if body else ""
 
 
 def patch_sdk():

src/pydo/operations/_patch.py

@@ -6,16 +6,115 @@
 
 Follow our quickstart for examples: https://aka.ms/azsdk/python/dpcodegen/python/customize
 """
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, cast
 
-from ._operations import DropletsOperations as Droplets
+from azure.core.exceptions import (
+    ClientAuthenticationError,
+    HttpResponseError,
+    ResourceExistsError,
+    ResourceNotFoundError,
+    ResourceNotModifiedError,
+    map_error,
+)
+from azure.core.pipeline import PipelineResponse
+from azure.core.tracing.decorator import distributed_trace
+
+from ._operations import (
+    KubernetesOperations as _KubernetesOperations,
+    build_kubernetes_get_kubeconfig_request,
+)
 
 if TYPE_CHECKING:
     # pylint: disable=unused-import,ungrouped-imports
-    pass
+    from typing import MutableMapping, Type
+
+
+__all__ = ["KubernetesOperations"]
+
+
+# Override: generated client expects JSON but this endpoint returns application/yaml;
+# we return the response body as str instead of deserializing.
+class KubernetesOperations(_KubernetesOperations):
+    """Kubernetes operations."""
+
+    @distributed_trace
+    def get_kubeconfig(
+        self, cluster_id: str, *, expiry_seconds: int = 0, **kwargs: Any
+    ) -> str:
+        """Retrieve the kubeconfig for a Kubernetes Cluster.
+
+        This endpoint returns a kubeconfig file in YAML format. It can be used to
+        connect to and administer the cluster using the Kubernetes command line tool,
+        ``kubectl``, or other programs supporting kubeconfig files (e.g., client libraries).
+
+        The resulting kubeconfig file uses token-based authentication for clusters
+        supporting it, and certificate-based authentication otherwise. For a list of
+        supported versions and more information, see "How to Connect to a DigitalOcean
+        Kubernetes Cluster"
+        https://docs.digitalocean.com/products/kubernetes/how-to/connect-to-cluster/
+
+        Clusters supporting token-based authentication may define an expiration by
+        passing a duration in seconds as a query parameter (expiry_seconds).
+        If not set or 0, then the token will have a 7 day expiry. The query parameter
+        has no impact in certificate-based authentication.
+
+        Kubernetes Roles granted to a user with a token-based kubeconfig are derived from that user's
+        DigitalOcean role. Custom roles require additional configuration by a cluster administrator.
+
+        :param cluster_id: A unique ID that can be used to reference a Kubernetes cluster. Required.
+        :type cluster_id: str
+        :keyword expiry_seconds: The duration in seconds that the returned Kubernetes credentials will
+         be valid. If not set or 0, the credentials will have a 7 day expiry. Default value is 0.
+        :paramtype expiry_seconds: int
+        :return: The kubeconfig file contents as a string (YAML).
+        :rtype: str
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: "MutableMapping[int, Type[HttpResponseError]]" = {
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+            401: cast(
+                "Type[HttpResponseError]",
+                lambda response: ClientAuthenticationError(response=response),
+            ),
+            429: HttpResponseError,
+            500: HttpResponseError,
+        }
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        _request = build_kubernetes_get_kubeconfig_request(
+            cluster_id=cluster_id,
+            expiry_seconds=expiry_seconds,
+            headers=_headers,
+            params=_params,
+        )
+        _request.url = self._client.format_url(_request.url)
+
+        # stream=True so the pipeline's content policy skips deserialization (API returns YAML, not JSON).
+        # Without this, the policy raises DecodeError for application/yaml. See test_kubernetes_get_kubeconfig.
+        pipeline_response: PipelineResponse = (
+            self._client._pipeline.run(  # pylint: disable=protected-access
+                _request, stream=True, **kwargs
+            )
+        )
+
+        response = pipeline_response.http_response
 
+        if response.status_code not in [200]:
+            response.read()
+            map_error(
+                status_code=response.status_code,
+                response=response,
+                error_map=error_map,
+            )
+            raise HttpResponseError(response=response)
 
-__all__ = []
+        body = response.read() if hasattr(response, "read") else response.content
+        return body.decode("utf-8") if body else ""
 
 
 def patch_sdk():