k8s-manifest-sorter.py

import operator
import sys
import yaml
from typing import Any, Callable, Dict, Union


def deep_sort(obj: dict, sorters: Dict[str, Callable] = None) -> Union[dict, list, Any]:
    sorters = sorters or {}

    result: Any = obj
    if isinstance(obj, dict):
        for key, func in sorters.items():
            if key in obj:
                obj[key].sort(key=func)
        sorted_tmp: list = [(k, deep_sort(v)) for k, v in obj.items()]
        sorted_tmp.sort()
        result: dict = dict(sorted_tmp)
    elif isinstance(obj, list):
        result: list = [deep_sort(v) for v in obj]
    return result


class KubernetesSortDumper(yaml.Dumper):

    name_getter = operator.itemgetter('name')
    sorters = {
        'containers': name_getter,
        'env': name_getter
    }

    @classmethod
    def represent_dict(cls, dumper, data):
        return dumper.represent_mapping('tag:yaml.org,2002:map', deep_sort(data, cls.sorters))

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.add_representer(dict, self.represent_dict)


if __name__ == '__main__':
    try:
        stream = sys.stdin
        if len(sys.argv) > 1 and sys.argv[1] != '-':
            stream = open(sys.argv[1])

        print(yaml.dump(yaml.safe_load(stream), Dumper=KubernetesSortDumper).rstrip('\n'))
    finally:
        if stream is not sys.stdin:
            stream.close()