Merge pull request #411 from fabianvf/rebase-on-upstream

Use upstream implementation as the base, maintain only diffs

Author: openshift-merge-robot

.travis.yml

@@ -5,8 +5,7 @@ cache:
 - pip
 language: python
 python:
-- '2.7'
-- '3.8'
+- '3.9'
 env:
   global:
     - COVERALLS_PARALLEL=true
@@ -20,21 +19,21 @@ after_success:
 jobs:
   include:
   - stage: lint
-    python: '2.7'
+    python: '3.9'
     install:
     - pip install tox-travis
     script: tox -e py27-lint
     env:
     - TEST_SUITE=lint
-  - python: '3.8'
+  - python: '3.9'
     install:
     - pip install tox-travis
     script: tox -e py35-lint
     env:
     - TEST_SUITE=lint
   - stage: deploy
     script: skip
-    python: '3.8'
+    python: '3.9'
     deploy:
       provider: pypi
       user: openshift
@@ -45,7 +44,7 @@ jobs:
         repo: openshift/openshift-restclient-python
         condition: "$TRAVIS_TAG =~ ^v[0-9]+\\.[0-9]+\\.[0-9]+(([ab]|dev|rc)[0-9]+)?$"
   - stage: test-deploy
-    python: '3.8'
+    python: '3.9'
     script: python -c "import openshift ; print(openshift.__version__)"
     install:
       - pip install openshift

openshift/dynamic/client.py

@@ -1,16 +1,13 @@
-import six
-import json
-
-from kubernetes import watch
-from kubernetes.client.rest import ApiException
+from kubernetes.dynamic.client import meta_request  # noqa
+from kubernetes.dynamic.client import DynamicClient as K8sDynamicClient
 
 from .apply import apply
 from .discovery import EagerDiscoverer, LazyDiscoverer
-from .exceptions import api_exception, KubernetesValidateMissing, ApplyException
 from .resource import Resource, ResourceList, Subresource, ResourceInstance, ResourceField
+from .exceptions import ApplyException
 
 try:
-    import kubernetes_validate
+    import kubernetes_validate  # noqa
     HAS_KUBERNETES_VALIDATE = True
 except ImportError:
     HAS_KUBERNETES_VALIDATE = False
@@ -33,103 +30,14 @@ class VersionNotSupportedError(NotImplementedError):
 ]
 
 
-def meta_request(func):
-    """ Handles parsing response structure and translating API Exceptions """
-    def inner(self, *args, **kwargs):
-        serialize_response = kwargs.pop('serialize', True)
-        serializer = kwargs.pop('serializer', ResourceInstance)
-        try:
-            resp = func(self, *args, **kwargs)
-        except ApiException as e:
-            raise api_exception(e)
-        if serialize_response:
-            try:
-                if six.PY2:
-                    return serializer(self, json.loads(resp.data))
-                return serializer(self, json.loads(resp.data.decode('utf8')))
-            except ValueError:
-                if six.PY2:
-                    return resp.data
-                return resp.data.decode('utf8')
-        return resp
-
-    return inner
-
-
-class DynamicClient(object):
+class DynamicClient(K8sDynamicClient):
     """ A kubernetes client that dynamically discovers and interacts with
         the kubernetes API
     """
 
     def __init__(self, client, cache_file=None, discoverer=None):
-        # Setting default here to delay evaluation of LazyDiscoverer class
-        # until constructor is called
         discoverer = discoverer or LazyDiscoverer
-
-        self.client = client
-        self.configuration = client.configuration
-        self.__discoverer = discoverer(self, cache_file)
-
-    @property
-    def resources(self):
-        return self.__discoverer
-
-    @property
-    def version(self):
-        return self.__discoverer.version
-
-    def ensure_namespace(self, resource, namespace, body):
-        namespace = namespace or body.get('metadata', {}).get('namespace')
-        if not namespace:
-            raise ValueError("Namespace is required for {}.{}".format(resource.group_version, resource.kind))
-        return namespace
-
-    def serialize_body(self, body):
-        if hasattr(body, 'to_dict'):
-            return body.to_dict()
-        return body or {}
-
-    def get(self, resource, name=None, namespace=None, **kwargs):
-        path = resource.path(name=name, namespace=namespace)
-        return self.request('get', path, **kwargs)
-
-    def create(self, resource, body=None, namespace=None, **kwargs):
-        body = self.serialize_body(body)
-        if resource.namespaced:
-            namespace = self.ensure_namespace(resource, namespace, body)
-        path = resource.path(namespace=namespace)
-        return self.request('post', path, body=body, **kwargs)
-
-    def delete(self, resource, name=None, namespace=None, label_selector=None, field_selector=None, **kwargs):
-        if not (name or label_selector or field_selector):
-            raise ValueError("At least one of name|label_selector|field_selector is required")
-        if resource.namespaced and not (label_selector or field_selector or namespace):
-            raise ValueError("At least one of namespace|label_selector|field_selector is required")
-        path = resource.path(name=name, namespace=namespace)
-        return self.request('delete', path, label_selector=label_selector, field_selector=field_selector, **kwargs)
-
-    def replace(self, resource, body=None, name=None, namespace=None, **kwargs):
-        body = self.serialize_body(body)
-        name = name or body.get('metadata', {}).get('name')
-        if not name:
-            raise ValueError("name is required to replace {}.{}".format(resource.group_version, resource.kind))
-        if resource.namespaced:
-            namespace = self.ensure_namespace(resource, namespace, body)
-        path = resource.path(name=name, namespace=namespace)
-        return self.request('put', path, body=body, **kwargs)
-
-    def patch(self, resource, body=None, name=None, namespace=None, **kwargs):
-        body = self.serialize_body(body)
-        name = name or body.get('metadata', {}).get('name')
-        if not name:
-            raise ValueError("name is required to patch {}.{}".format(resource.group_version, resource.kind))
-        if resource.namespaced:
-            namespace = self.ensure_namespace(resource, namespace, body)
-
-        content_type = kwargs.pop('content_type', 'application/strategic-merge-patch+json')
-        path = resource.path(name=name, namespace=namespace)
-
-        return self.request('patch', path, body=body, content_type=content_type, **kwargs)
+        K8sDynamicClient.__init__(self, client, cache_file=cache_file, discoverer=discoverer)
 
     def apply(self, resource, body=None, name=None, namespace=None):
         body = self.serialize_body(body)
@@ -144,139 +52,3 @@ def apply(self, resource, body=None, name=None, namespace=None):
         except ApplyException as e:
             raise ValueError("Could not apply strategic merge to %s/%s: %s" %
                              (body['kind'], body['metadata']['name'], e))
-
-    def watch(self, resource, namespace=None, name=None, label_selector=None, field_selector=None, resource_version=None, timeout=None):
-        """
-        Stream events for a resource from the Kubernetes API
-
-        :param resource: The API resource object that will be used to query the API
-        :param namespace: The namespace to query
-        :param name: The name of the resource instance to query
-        :param label_selector: The label selector with which to filter results
-        :param field_selector: The field selector with which to filter results
-        :param resource_version: The version with which to filter results. Only events with
-                                 a resource_version greater than this value will be returned
-        :param timeout: The amount of time in seconds to wait before terminating the stream
-
-        :return: Event object with these keys:
-                   'type': The type of event such as "ADDED", "DELETED", etc.
-                   'raw_object': a dict representing the watched object.
-                   'object': A ResourceInstance wrapping raw_object.
-
-        Example:
-            client = DynamicClient(k8s_client)
-            v1_pods = client.resources.get(api_version='v1', kind='Pod')
-
-            for e in v1_pods.watch(resource_version=0, namespace=default, timeout=5):
-                print(e['type'])
-                print(e['object'].metadata)
-        """
-        watcher = watch.Watch()
-        for event in watcher.stream(
-            resource.get,
-            namespace=namespace,
-            name=name,
-            field_selector=field_selector,
-            label_selector=label_selector,
-            resource_version=resource_version,
-            serialize=False,
-            timeout_seconds=timeout
-        ):
-            event['object'] = ResourceInstance(resource, event['object'])
-            yield event
-
-    @meta_request
-    def request(self, method, path, body=None, **params):
-        if not path.startswith('/'):
-            path = '/' + path
-
-        path_params = params.get('path_params', {})
-        query_params = params.get('query_params', [])
-        if params.get('pretty') is not None:
-            query_params.append(('pretty', params['pretty']))
-        if params.get('_continue') is not None:
-            query_params.append(('continue', params['_continue']))
-        if params.get('include_uninitialized') is not None:
-            query_params.append(('includeUninitialized', params['include_uninitialized']))
-        if params.get('field_selector') is not None:
-            query_params.append(('fieldSelector', params['field_selector']))
-        if params.get('label_selector') is not None:
-            query_params.append(('labelSelector', params['label_selector']))
-        if params.get('limit') is not None:
-            query_params.append(('limit', params['limit']))
-        if params.get('resource_version') is not None:
-            query_params.append(('resourceVersion', params['resource_version']))
-        if params.get('timeout_seconds') is not None:
-            query_params.append(('timeoutSeconds', params['timeout_seconds']))
-        if params.get('watch') is not None:
-            query_params.append(('watch', params['watch']))
-        if params.get('dry_run') is not None:
-            query_params.append(('dryRun', params['dry_run']))
-
-        header_params = params.get('header_params', {})
-        form_params = []
-        local_var_files = {}
-
-        # Checking Accept header.
-        new_header_params = dict((key.lower(), value) for key, value in header_params.items())
-        if not 'accept' in new_header_params:
-            header_params['Accept'] = self.client.select_header_accept([
-                'application/json',
-                'application/yaml',
-            ])
-
-        # HTTP header `Content-Type`
-        if params.get('content_type'):
-            header_params['Content-Type'] = params['content_type']
-        else:
-            header_params['Content-Type'] = self.client.select_header_content_type(['*/*'])
-
-        # Authentication setting
-        auth_settings = ['BearerToken']
-
-        return self.client.call_api(
-            path,
-            method.upper(),
-            path_params,
-            query_params,
-            header_params,
-            body=body,
-            post_params=form_params,
-            async_req=params.get('async_req'),
-            files=local_var_files,
-            auth_settings=auth_settings,
-            _preload_content=False,
-            _return_http_data_only=params.get('_return_http_data_only', True)
-        )
-
-    def validate(self, definition, version=None, strict=False):
-        """validate checks a kubernetes resource definition
-
-        Args:
-            definition (dict): resource definition
-            version (str): version of kubernetes to validate against
-            strict (bool): whether unexpected additional properties should be considered errors
-
-        Returns:
-            warnings (list), errors (list): warnings are missing validations, errors are validation failures
-        """
-        if not HAS_KUBERNETES_VALIDATE:
-            raise KubernetesValidateMissing()
-
-        errors = list()
-        warnings = list()
-        try:
-            if version is None:
-                try:
-                    version = self.version['kubernetes']['gitVersion']
-                except KeyError:
-                    version = kubernetes_validate.latest_version()
-            kubernetes_validate.validate(definition, version, strict)
-        except kubernetes_validate.utils.ValidationError as e:
-            errors.append("resource definition validation error at %s: %s" % ('.'.join([str(item) for item in e.path]), e.message))  # noqa: B306
-        except VersionNotSupportedError:
-            errors.append("Kubernetes version %s is not supported by kubernetes-validate" % version)
-        except kubernetes_validate.utils.SchemaNotFoundError as e:
-            warnings.append("Could not find schema for object kind %s with API version %s in Kubernetes version %s (possibly Custom Resource?)" %
-                            (e.kind, e.api_version, e.version))
-        return warnings, errors