Allowing **kwargs to be passed to modify_and_apply()

Author: bradmwilliams

ansible/rebuild_module.digest

@@ -1 +1 @@
-40a6f692df33c354be1fae814d8429d5  -
+a47abb7772c34b91b10eab8657dabc12  -

ansible/roles/openshift_client_python/library/openshift_client_python.py

@@ -0,0 +1,120 @@
+#!/usr/bin/python
+
+import argparse
+import json
+import logging
+import sys
+import traceback
+
+import openshift_client as oc
+from openshift_client import OpenShiftPythonException
+from openshift_client.decorators import ephemeral_project
+
+logging.basicConfig(level=logging.INFO, stream=sys.stdout, format='%(message)s')
+logger = logging.getLogger('ModifyAndApply')
+
+
+def validate_server_connection(ctx):
+    with oc.options(ctx), oc.tracking(), oc.timeout(60):
+        try:
+            username = oc.whoami()
+            version = oc.get_server_version()
+            logger.debug(f'Connected to APIServer running version: {version}, as: {username}')
+        except (ValueError, OpenShiftPythonException, Exception) as e:
+            logger.error(f"Unable to verify cluster connection using context: \"{ctx['context']}\"")
+            raise e
+
+
+def test_update_dynamic_keyword_args(obj):
+    def update_dynamic_keyword_args(apiobj, **kwargs):
+        logger.info(f'Updating object: {apiobj.name()} with: {json.dumps(kwargs, indent=4, default=str)}')
+        return False
+
+    r, success = obj.modify_and_apply(update_dynamic_keyword_args, retries=0)
+    assert len(r.actions()) == 0
+    assert success == False
+
+    r, success = obj.modify_and_apply(update_dynamic_keyword_args, retries=0, param1='foo')
+    assert len(r.actions()) == 0
+    assert success == False
+
+    r, success = obj.modify_and_apply(update_dynamic_keyword_args, retries=0, param1='foo', param2='bar')
+    assert len(r.actions()) == 0
+    assert success == False
+
+    r, success = obj.modify_and_apply(update_dynamic_keyword_args, retries=0, random1='foo', modnar1='bar')
+    assert len(r.actions()) == 0
+    assert success == False
+
+
+def test_update_named_keyword_args(obj):
+    def update_named_keyword_args(apiobj, param1=None, param2=None):
+        logger.info(f'Updating object: {apiobj.name()} with "param1={param1}" and "param2={param2}"')
+        return False
+
+    r, success = obj.modify_and_apply(update_named_keyword_args, retries=0)
+    assert len(r.actions()) == 0
+    assert success == False
+
+    r, success = obj.modify_and_apply(update_named_keyword_args, retries=0, param1='foo')
+    assert len(r.actions()) == 0
+    assert success == False
+
+    r, success = obj.modify_and_apply(update_named_keyword_args, retries=0, param1='foo', param2='bar')
+    assert len(r.actions()) == 0
+    assert success == False
+
+    try:
+        obj.modify_and_apply(update_named_keyword_args, retries=0, param3='bip')
+    except TypeError as e:
+        if 'got an unexpected keyword argument' in e.__str__():
+            logger.info(f'Unknown parameter specified: {e}')
+        else:
+            raise e
+
+
+@ephemeral_project
+def run(*, project_name=None):
+    logger.info('Running in namespace: {}'.format(project_name))
+    obj = oc.selector('serviceaccount/default').object()
+    test_update_named_keyword_args(obj)
+    test_update_dynamic_keyword_args(obj)
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description='Backup namespace resources')
+
+    config_group = parser.add_argument_group('Configuration Options')
+    config_group.add_argument('-v', '--verbose', help='Enable verbose output', action='store_true')
+
+    ocp_group = parser.add_argument_group('Openshift Cluster Configuration Options')
+    ocp_group.add_argument('-c', '--context', help='The OC context to use', default=None)
+    ocp_group.add_argument('-k', '--kubeconfig', help='The kubeconfig to use (default is "~/.kube/config")', default=None)
+    ocp_group.add_argument('-n', '--namespace', help='The namespace to process', default=None)
+
+    args = vars(parser.parse_args())
+
+    if args['verbose']:
+        logger.setLevel(logging.DEBUG)
+
+    # Validate the connection to the respective cluster
+    context = {}
+    if args['context'] is not None:
+        context.update({'context': args['context']})
+
+    if args['kubeconfig'] is not None:
+        context.update({'kubeconfig': args['kubeconfig']})
+
+    validate_server_connection(context)
+
+    with oc.client_host():
+        with oc.timeout(60 * 10), oc.tracking() as t:
+            with oc.options(context):
+                try:
+                    run()
+                except (ValueError, OpenShiftPythonException, Exception):
+                    # Print out exception stack trace via the traceback module
+                    logger.info('Traceback output:\n{}\n'.format(traceback.format_exc()))
+
+                    # Print out all oc interactions and do not redact secret information
+                    logger.info("OC tracking output:\n{}\n".format(t.get_result().as_json(redact_streams=False)))

examples/modify_and_apply.py

@@ -1,16 +1,17 @@
 from __future__ import absolute_import
 
-import yaml
-import sys
 import copy
+import sys
 
+import yaml
+
+from . import util
 from .action import *
+from .context import cur_context
 from .model import *
-from .result import *
 from .naming import kind_matches
-from .context import cur_context
+from .result import *
 from .selector import selector
-from . import util
 
 _DEFAULT = object()
 
@@ -467,17 +468,18 @@ def print_logs(self, stream=sys.stderr, timestamps=False, previous=False, since=
                         self.logs(timestamps=timestamps, previous=previous, since=since, limit_bytes=limit_bytes,
                                   tail=tail, try_longshots=try_longshots, cmd_args=cmd_args))
 
-    def modify_and_apply(self, modifier_func, retries=2, cmd_args=None):
+    def modify_and_apply(self, modifier_func, retries=2, cmd_args=None, **kwargs):
         """
         Calls the modifier_func with self. The function should modify the model of the apiobj argument
         and return True if it wants this method to try to apply the change via the API. For robust
         implementations, a non-zero number of retries is recommended.
 
-        :param modifier_func: Called before each attempt with an self. The associated model will be refreshed before
+        :param modifier_func: Called before each attempt with a self. The associated model will be refreshed before
             each call if necessary. If the function finds changes it wants to make to the model, it should
             make them directly and return True. If it does not want to make changes, it should return False.
         :param cmd_args: An optional list of additional arguments to pass on the command line
         :param retries: The number of times to retry. A value of 0 means only one attempt will be made.
+        :param kwargs: keyword arguments passed directly into modifier_func
         :return: A Result object containing a record of all attempts AND a boolean. The boolean indicates
         True if a change was applied to a resource (i.e. it will be False if modifier_func suggested no
         change was necessary by returning False).
@@ -488,7 +490,7 @@ def modify_and_apply(self, modifier_func, retries=2, cmd_args=None):
         applied_change = False
         for attempt in reversed(list(range(retries + 1))):
 
-            do_apply = modifier_func(self)
+            do_apply = modifier_func(self, **kwargs)
 
             # Modifier does not want to modify this object -- stop retrying. Retuning None should continue attempts.
             if do_apply is False: