Iterate over all iterables (except dicts) in hierarchy functions
"""
All the functions to properly build the object hierarchies.
"""
+import collections.abc
def build_object_reference(body):
@@ -40,11 +41,8 @@ def append_owner_reference(objs, owner):
Note: the owned objects are usually not the one being processed,
so the whole body can be modified, no patches are needed.
"""
- if not isinstance(objs, (list, tuple)):
- objs = [objs]
-
owner = build_owner_reference(owner)
- for obj in objs:
+ for obj in _iter_objects(objs):
refs = obj.setdefault('metadata', {}).setdefault('ownerReferences', [])
matching = [ref for ref in refs if ref['uid'] == owner['uid']]
if not matching:
@@ -58,11 +56,8 @@ def remove_owner_reference(objs, owner):
Note: the owned objects are usually not the one being processed,
so the whole body can be modified, no patches are needed.
"""
- if not isinstance(objs, (list, tuple)):
- objs = [objs]
-
owner = build_owner_reference(owner)
- for obj in objs:
+ for obj in _iter_objects(objs):
refs = obj.setdefault('metadata', {}).setdefault('ownerReferences', [])
matching = [ref for ref in refs if ref['uid'] == owner['uid']]
for ref in matching:
@@ -74,10 +69,7 @@ def label(objs, labels, force=False):
"""
Apply the labels to the object(s).
"""
- if not isinstance(objs, (list, tuple)):
- objs = [objs]
-
- for obj in objs:
+ for obj in _iter_objects(objs):
obj_labels = obj.setdefault('metadata', {}).setdefault('labels', {})
for key, val in labels.items():
if force:
@@ -101,10 +93,7 @@ def harmonize_naming(objs, name=None, strict=False):
If the objects already have their own names, auto-naming is not applied,
and the existing names are used as is.
"""
- if not isinstance(objs, (list, tuple)):
- objs = [objs]
-
- for obj in objs:
+ for obj in _iter_objects(objs):
if obj.get('metadata', {}).get('name', None) is None:
if strict:
obj.setdefault('metadata', {}).setdefault('name', name)
@@ -121,10 +110,7 @@ def adjust_namespace(objs, namespace=None):
It is a common practice to keep the children objects in the same
namespace as their owner, unless explicitly overridden at time of creation.
"""
- if not isinstance(objs, (list, tuple)):
- objs = [objs]
-
- for obj in objs:
+ for obj in _iter_objects(objs):
obj.setdefault('metadata', {}).setdefault('namespace', namespace)
@@ -136,3 +122,17 @@ def adopt(objs, owner):
harmonize_naming(objs, name=owner.get('metadata', {}).get('name', None))
adjust_namespace(objs, namespace=owner.get('metadata', {}).get('namespace', None))
label(objs, labels=owner.get('metadata', {}).get('labels', {}))
+
+
+def _iter_objects(objs):
+ """
+ A helper to iterate over one or many objects, as used in these functions.
+ """
+ if objs is None:
+ return
+ elif isinstance(objs, collections.abc.Mapping): # also iterable, intercept it.
+ yield objs
+ elif isinstance(objs, collections.abc.Iterable):
+ yield from objs
+ else:
+ yield objs
