internet/kubernetes.core
3
0
Fork 0
mirror of https://github.com/ansible-collections/kubernetes.core.git synced 2026-07-29 02:44:41 +00:00
Code Issues Projects Releases Wiki Activity

Enable black formatting test (#259)

Browse Source 
Enable black formatting test

SUMMARY
Signed-off-by: Abhijeet Kasurde akasurde@redhat.com
ISSUE TYPE

Bugfix Pull Request

COMPONENT NAME
plugins/action/k8s_info.py
plugins/connection/kubectl.py
plugins/doc_fragments/helm_common_options.py
plugins/doc_fragments/k8s_auth_options.py
plugins/doc_fragments/k8s_delete_options.py
plugins/doc_fragments/k8s_name_options.py
plugins/doc_fragments/k8s_resource_options.py
plugins/doc_fragments/k8s_scale_options.py
plugins/doc_fragments/k8s_state_options.py
plugins/doc_fragments/k8s_wait_options.py
plugins/filter/k8s.py
plugins/inventory/k8s.py
plugins/lookup/k8s.py
plugins/lookup/kustomize.py
plugins/module_utils/ansiblemodule.py
plugins/module_utils/apply.py
plugins/module_utils/args_common.py
plugins/module_utils/client/discovery.py
plugins/module_utils/client/resource.py
plugins/module_utils/common.py
plugins/module_utils/exceptions.py
plugins/module_utils/hashes.py
plugins/module_utils/helm.py
plugins/module_utils/k8sdynamicclient.py
plugins/module_utils/selector.py
plugins/modules/helm.py
plugins/modules/helm_info.py
plugins/modules/helm_plugin.py
plugins/modules/helm_plugin_info.py
plugins/modules/helm_repository.py
plugins/modules/helm_template.py
plugins/modules/k8s.py
plugins/modules/k8s_cluster_info.py
plugins/modules/k8s_cp.py
plugins/modules/k8s_drain.py
plugins/modules/k8s_exec.py
plugins/modules/k8s_info.py
plugins/modules/k8s_json_patch.py
plugins/modules/k8s_log.py
plugins/modules/k8s_rollback.py
plugins/modules/k8s_scale.py
plugins/modules/k8s_service.py
tests/integration/targets/kubernetes/library/test_tempfile.py
tests/unit/module_utils/test_apply.py
tests/unit/module_utils/test_common.py
tests/unit/module_utils/test_discoverer.py
tests/unit/module_utils/test_hashes.py
tests/unit/module_utils/test_marshal.py
tests/unit/module_utils/test_selector.py
tox.ini

Reviewed-by: None <None>
Reviewed-by: Mike Graves <mgraves@redhat.com>
Reviewed-by: None <None>
This commit is contained in:
Abhijeet Kasurde
2021-10-18 21:02:05 +05:30
committed by GitHub
parent 4010987d1f
commit 91b80b1d1d
50 changed files with 3453 additions and 2175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
plugins/lookup/k8s.py
View File

@@ -3,11 +3,11 @@
#
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import (absolute_import, division, print_function)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = '''
DOCUMENTATION = """
lookup: k8s
short_description: Query the K8s API
@@ -117,7 +117,7 @@ DOCUMENTATION = '''
- "python >= 3.6"
- "kubernetes >= 12.0.0"
- "PyYAML >= 3.11"
'''
"""
EXAMPLES = """
- name: Fetch a list of namespaces
@@ -187,11 +187,15 @@ from ansible.errors import AnsibleError
from ansible.module_utils.common._collections_compat import KeysView
from ansible.plugins.lookup import LookupBase
from ansible_collections.kubernetes.core.plugins.module_utils.common import K8sAnsibleMixin, get_api_client
from ansible_collections.kubernetes.core.plugins.module_utils.common import (
K8sAnsibleMixin,
get_api_client,
)
try:
from kubernetes.dynamic.exceptions import NotFoundError
HAS_K8S_MODULE_HELPER = True
k8s_import_exception = None
except ImportError as e:
@@ -200,12 +204,13 @@ except ImportError as e:
class KubernetesLookup(K8sAnsibleMixin):
def __init__(self):
if not HAS_K8S_MODULE_HELPER:
raise Exception(
"Requires the Kubernetes Python client. Try `pip install kubernetes`. Detail: {0}".format(k8s_import_exception)
"Requires the Kubernetes Python client. Try `pip install kubernetes`. Detail: {0}".format(
k8s_import_exception
)
)
self.kind = None
@@ -226,31 +231,33 @@ class KubernetesLookup(K8sAnsibleMixin):
self.params = kwargs
self.client = get_api_client(**kwargs)
cluster_info = kwargs.get('cluster_info')
if cluster_info == 'version':
cluster_info = kwargs.get("cluster_info")
if cluster_info == "version":
return [self.client.version]
if cluster_info == 'api_groups':
if cluster_info == "api_groups":
if isinstance(self.client.resources.api_groups, KeysView):
return [list(self.client.resources.api_groups)]
return [self.client.resources.api_groups]
self.kind = kwargs.get('kind')
self.name = kwargs.get('resource_name')
self.namespace = kwargs.get('namespace')
self.api_version = kwargs.get('api_version', 'v1')
self.label_selector = kwargs.get('label_selector')
self.field_selector = kwargs.get('field_selector')
self.include_uninitialized = kwargs.get('include_uninitialized', False)
self.kind = kwargs.get("kind")
self.name = kwargs.get("resource_name")
self.namespace = kwargs.get("namespace")
self.api_version = kwargs.get("api_version", "v1")
self.label_selector = kwargs.get("label_selector")
self.field_selector = kwargs.get("field_selector")
self.include_uninitialized = kwargs.get("include_uninitialized", False)
resource_definition = kwargs.get('resource_definition')
src = kwargs.get('src')
resource_definition = kwargs.get("resource_definition")
src = kwargs.get("src")
if src:
resource_definition = self.load_resource_definitions(src)[0]
if resource_definition:
self.kind = resource_definition.get('kind', self.kind)
self.api_version = resource_definition.get('apiVersion', self.api_version)
self.name = resource_definition.get('metadata', {}).get('name', self.name)
self.namespace = resource_definition.get('metadata', {}).get('namespace', self.namespace)
self.kind = resource_definition.get("kind", self.kind)
self.api_version = resource_definition.get("apiVersion", self.api_version)
self.name = resource_definition.get("metadata", {}).get("name", self.name)
self.namespace = resource_definition.get("metadata", {}).get(
"namespace", self.namespace
)
if not self.kind:
raise AnsibleError(
@@ -260,17 +267,21 @@ class KubernetesLookup(K8sAnsibleMixin):
resource = self.find_resource(self.kind, self.api_version, fail=True)
try:
k8s_obj = resource.get(name=self.name, namespace=self.namespace, label_selector=self.label_selector, field_selector=self.field_selector)
k8s_obj = resource.get(
name=self.name,
namespace=self.namespace,
label_selector=self.label_selector,
field_selector=self.field_selector,
)
except NotFoundError:
return []
if self.name:
return [k8s_obj.to_dict()]
return k8s_obj.to_dict().get('items')
return k8s_obj.to_dict().get("items")
class LookupModule(LookupBase):
def run(self, terms, variables=None, **kwargs):
return KubernetesLookup().run(terms, variables=variables, **kwargs)