1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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
