File: utils.py

package info (click to toggle)
python-pykube-ng 22.9.0-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 424 kB
  • sloc: python: 2,336; makefile: 44
file content (89 lines) | stat: -rw-r--r-- 2,430 bytes parent folder | download
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
import re
from typing import List

try:
    from jsonpath_ng import parse as jsonpath

    jsonpath_installed = True
except ImportError:
    jsonpath_installed = False

from itertools import zip_longest


empty = object()


def obj_merge(obj, original_obj, is_strategic=True):
    c = {}
    for k, v in obj.items():
        if k not in original_obj:
            c[k] = v
        else:
            c[k] = obj_check(v, original_obj[k], is_strategic)

    if is_strategic is True:
        for k, v in original_obj.items():
            if k not in obj:
                c[k] = v
    return c


def obj_check(obj_value, original_obj_value, is_strategic=True):
    check_result = None
    if not isinstance(obj_value, type(original_obj_value)):
        check_result = obj_value
    else:
        if isinstance(obj_value, dict):
            check_result = obj_merge(obj_value, original_obj_value, is_strategic)

        elif isinstance(obj_value, list):
            if is_strategic:
                res_list = []
                for x, y in zip_longest(obj_value, original_obj_value, fillvalue=empty):
                    if x is empty:
                        res_list.append(y)
                    elif y is empty:
                        res_list.append(x)
                    else:
                        res_list.append(obj_check(x, y, is_strategic))
                check_result = res_list
            else:
                check_result = obj_value
        else:
            check_result = obj_value
    return check_result


def jsonpath_parse(template, obj):
    def repl(m):
        path = m.group(2)
        if not path.startswith("$"):
            path = "$" + path
        return jsonpath(path).find(obj)[0].value

    return re.sub(r"(\{([^\}]*)\})", repl, template)


def join_url_path(*components, join_empty: bool = False) -> str:
    """Join given URL path components and return absolute path starting with '/'."""
    new_comps: List[str] = []
    for comp in components:
        comp = comp.strip("/")
        if comp in ("", "."):
            continue
        else:
            new_comps.append(comp)

    if components and components[-1] == "" and join_empty:
        trailing_slash = True
    elif components and components[-1] == "/":
        trailing_slash = True
    else:
        trailing_slash = False

    if trailing_slash:
        new_comps.append("")

    path = "/".join(new_comps)
    return "/" + path