25400571889

Committed 05 May 2026 08:29PM UTC coverage: 90.612% (-0.4%) from 90.986%

Build # 25400571889

Build Type

Pull #643

github

Committed by

web-flow

Commit Message

Merge 4e812ec20 into 15d676845

Pull Request Pull Request #643: Adding new backfill option

Coverage Stats

8291 of 9150 relevant lines covered (90.61%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

56.25

back/admin/integrations/utils.py

def _tokenize_notation(notation):
    # split on '.' but keep [...] groups intact, so values inside a filter
    # expression (which may themselves contain '.', e.g. emails) aren't split
    tokens = []
    buf = ""
    depth = 0
    for ch in notation:
        if ch == "[":
            depth += 1
            buf += ch
        elif ch == "]":
            depth -= 1
            buf += ch
        elif ch == "." and depth == 0:
            if buf:
                tokens.append(buf)
                buf = ""
        else:
            buf += ch
    if buf:
        tokens.append(buf)
    return tokens


def get_value_from_notation(notation, value):
    # if we don't need to go into props, then just return the value
    if notation == "":
        return value

    for token in _tokenize_notation(notation):
        # filter form: optional_key[field=expected] - pick first list entry
        # whose `field` equals `expected`. Useful when the upstream API returns
        # an unfiltered list (e.g. Bitwarden /public/members).
        if "[" in token and token.endswith("]"):
            list_key, _, filter_expr = token.partition("[")
            filter_expr = filter_expr[:-1]

            if list_key:
                try:
                    value = value[list_key]
                except (KeyError, TypeError):
                    raise KeyError

            if "=" not in filter_expr or not isinstance(value, list):
                raise KeyError

            field, _, expected = filter_expr.partition("=")
            for item in value:
                if isinstance(item, dict) and str(item.get(field, "")) == expected:
                    value = item
                    break
            else:
                raise KeyError
            continue

        try:
            value = value[token]
        except TypeError:
            if not isinstance(value, list):
                raise KeyError

            try:
                index = int(token)
            except (TypeError, ValueError):
                raise KeyError

            try:
                value = value[index]
            except (TypeError, ValueError, IndexError):
                raise KeyError

    return value


def convert_array_to_object(arr):
    return {item["key"]: item["value"] for item in arr}


def convert_object_to_array(obj):
    return [{"key": key, "value": value} for key, value in obj.items()]


def prepare_initial_data(obj):
    if isinstance(obj, dict):
        if obj.get("headers") and not isinstance(obj["headers"], list):
            obj["headers"] = convert_object_to_array(obj["headers"])
        return obj

    return obj

1	def _tokenize_notation(notation):	1✔
2	# split on '.' but keep [...] groups intact, so values inside a filter
3	# expression (which may themselves contain '.', e.g. emails) aren't split
4	tokens = []	1✔
5	buf = ""	1✔
6	depth = 0	1✔
7	for ch in notation:	1✔
8	if ch == "[":	1✔
9	depth += 1	×
10	buf += ch	×
11	elif ch == "]":	1✔
12	depth -= 1	×
13	buf += ch	×
14	elif ch == "." and depth == 0:	1✔
15	if buf:	1✔
16	tokens.append(buf)	1✔
17	buf = ""	1✔
18	else:
19	buf += ch	1✔
20	if buf:	1✔
21	tokens.append(buf)	1✔
22	return tokens	1✔
23
24
25	def get_value_from_notation(notation, value):	1✔
26	# if we don't need to go into props, then just return the value
27	if notation == "":	1✔
28	return value	1✔
29
30	for token in _tokenize_notation(notation):	1✔
31	# filter form: optional_key[field=expected] - pick first list entry
32	# whose `field` equals `expected`. Useful when the upstream API returns
33	# an unfiltered list (e.g. Bitwarden /public/members).
34	if "[" in token and token.endswith("]"):	1✔
35	list_key, _, filter_expr = token.partition("[")	×
36	filter_expr = filter_expr[:-1]	×
37
38	if list_key:	×
39	try:	×
40	value = value[list_key]	×
41	except (KeyError, TypeError):	×
42	raise KeyError	×
43
44	if "=" not in filter_expr or not isinstance(value, list):	×
45	raise KeyError	×
46
47	field, _, expected = filter_expr.partition("=")	×
48	for item in value:	×
49	if isinstance(item, dict) and str(item.get(field, "")) == expected:	×
50	value = item	×
51	break	×
52	else:
53	raise KeyError	×
54	continue	×
55
56	try:	1✔
57	value = value[token]	1✔
58	except TypeError:	1✔
59	if not isinstance(value, list):	1✔
60	raise KeyError	×
61
62	try:	1✔
63	index = int(token)	1✔
64	except (TypeError, ValueError):	1✔
65	raise KeyError	1✔
66
67	try:	1✔
68	value = value[index]	1✔
69	except (TypeError, ValueError, IndexError):	1✔
70	raise KeyError	1✔
71
72	return value	1✔
73
74
75	def convert_array_to_object(arr):	1✔
76	return {item["key"]: item["value"] for item in arr}	×
77
78
79	def convert_object_to_array(obj):	1✔
80	return [{"key": key, "value": value} for key, value in obj.items()]	×
81
82
83	def prepare_initial_data(obj):	1✔
84	if isinstance(obj, dict):	×
85	if obj.get("headers") and not isinstance(obj["headers"], list):	×
86	obj["headers"] = convert_object_to_array(obj["headers"])	×
87	return obj	×
88
89	return obj	×

chiefonboarding / ChiefOnboarding / 25400571889

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous