How to use the kopf.structs.dicts.walk function in kopf

To help you get started, we’ve selected a few kopf examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github zalando-incubator / kopf / tests / dicts / test_walking.py View on Github external
def test_over_a_dict():
    obj = {}
    result = list(walk(obj))
    assert len(result) == 1
    assert result[0] is obj
github zalando-incubator / kopf / tests / dicts / test_walking.py View on Github external
def test_over_a_list_of_dicts():
    obj1 = {}
    obj2 = {}
    result = list(walk([obj1, obj2]))
    assert len(result) == 2
    assert result[0] is obj1
    assert result[1] is obj2
github zalando-incubator / kopf / tests / dicts / test_walking.py View on Github external
def test_simple_nested():
    obj1 = {'field': {'subfield': 'val'}}
    obj2 = {'field': {}}
    result = list(walk([obj1, obj2], nested=['field.subfield']))
    assert len(result) == 3
    assert result[0] is obj1
    assert result[1] == 'val'
    assert result[2] is obj2
github zalando-incubator / kopf / tests / dicts / test_walking.py View on Github external
def test_over_a_tuple_of_dicts():
    obj1 = {}
    obj2 = {}
    result = list(walk((obj1, obj2)))
    assert len(result) == 2
    assert result[0] is obj1
    assert result[1] is obj2
github zalando-incubator / kopf / tests / dicts / test_walking.py View on Github external
def test_over_a_none():
    result = list(walk(None))
    assert len(result) == 0
github zalando-incubator / kopf / kopf / engines / posting.py View on Github external
def event(objs, *, type, reason, message=''):
    for obj in dicts.walk(objs):
        ref = hierarchies.build_object_reference(obj)
        enqueue(ref=ref, type=type, reason=reason, message=message)
github zalando-incubator / kopf / kopf / toolkits / hierarchies.py View on Github external
def label(
        objs: K8sObjects,
        labels: bodies.Labels,
        *,
        force: bool = False,
        nested: Optional[Iterable[dicts.FieldSpec]] = None,
) -> None:
    """
    Apply the labels to the object(s).
    """
    for obj in cast(Iterator[K8sObject], dicts.walk(objs, nested=nested)):
        obj_labels = obj.setdefault('metadata', {}).setdefault('labels', {})
        for key, val in labels.items():
            if force:
                obj_labels[key] = val
            else:
                obj_labels.setdefault(key, val)
github zalando-incubator / kopf / kopf / toolkits / hierarchies.py View on Github external
"""
    Adjust the namespace of the objects.

    If the objects already have the namespace set, it will be preserved.

    It is a common practice to keep the children objects in the same
    namespace as their owner, unless explicitly overridden at time of creation.
    """

    # Try to use the current object being handled if possible.
    if namespace is None:
        real_owner = _guess_owner(None)
        namespace = real_owner.get('metadata', {}).get('namespace', None)

    # Set namespace based on the explicitly specified or guessed namespace.
    for obj in cast(Iterator[K8sObject], dicts.walk(objs)):
        obj.setdefault('metadata', {}).setdefault('namespace', namespace)