import asyncio import json import re def items(x): """List items: x["items"] for a Kubernetes list, x for a list, [] for None.""" if x is None: return [] if isinstance(x, dict): return x.get("items", []) return list(x) def dig(obj, *path, default=None): """Nested lookup: dig(pod, "status.phase") or dig(pod, "status", "phase").""" if path and not isinstance(path[-1], (str, int)): default = path[-1] path = path[:-1] for part in path: for key in str(part).split("."): if isinstance(obj, dict) and key in obj: obj = obj[key] elif isinstance(obj, list) and key.isdigit() and int(key) < len(obj): obj = obj[int(key)] else: return default return obj def names(x): """metadata.name (or name) of every item.""" return [dig(i, "metadata.name", default=dig(i, "name")) for i in items(x)] def count_by(xs, *path): """Count items by a path: count_by(pods, "status.phase").""" counts = {} for i in items(xs): key = dig(i, *path) key = "None" if key is None else str(key) counts[key] = counts.get(key, 0) + 1 return counts def _asyncai_plain(x): if x is None or isinstance(x, (bool, int, float, str)): return x if isinstance(x, dict): return {str(k): _asyncai_plain(v) for k, v in x.items()} if isinstance(x, (list, tuple, set)): return [_asyncai_plain(v) for v in x] return str(x) def to_json(x, indent=None): """json.dumps that turns unknown types into str (Monty's json.dumps has no default=).""" return json.dumps(_asyncai_plain(x), indent=indent) def _asyncai_describe(x): kind = type(x).__name__ if isinstance(x, (list, dict, str, tuple, set)): return kind + "[" + str(len(x)) + "]" return kind