def boolean()

in lib/kql/kql/dsl.py [0:0]

• 42 lines of code
• 22 McCabe index (conditional complexity)

def boolean(**kwargs):
    """Wrap a query in a boolean term and optimize while building."""
    assert len(kwargs) == 1
    [(boolean_type, children)] = kwargs.items()

    if not isinstance(children, list):
        children = [children]

    dsl = defaultdict(list)

    if boolean_type in ("must", "filter"):
        # safe to convert and(and(x), y) -> and(x, y)
        for child in children:
            if list(child) == ["bool"]:
                for child_type, child_terms in child["bool"].items():
                    if child_type in ("must", "filter"):
                        dsl[child_type].extend(child_terms)
                    elif child_type == "should":
                        if "should" not in dsl:
                            dsl[child_type].extend(child_terms)
                        else:
                            dsl[boolean_type].append(boolean(should=child_terms))
                    elif child_type == "must_not":
                        dsl[child_type].extend(child_terms)
                    elif child_type != "minimum_should_match":
                        raise ValueError("Unknown term {}: {}".format(child_type, child_terms))
            else:
                dsl[boolean_type].append(child)

    elif boolean_type == "should":
        # can flatten `should` of `should`
        for child in children:
            if list(child) == ["bool"] and set(child["bool"]).issubset({"should", "minimum_should_match"}):
                dsl["should"].extend(child["bool"]["should"])
            else:
                dsl[boolean_type].append(child)

    elif boolean_type == "must_not" and len(children) == 1:
        # must_not: [{bool: {must: x}}] -> {must_not: x}
        child = children[0]
        if list(child) == ["bool"] and list(child["bool"]) in (["filter"], ["must"]):
            negated, = child["bool"].values()
            dsl = {"must_not": negated}
        else:
            dsl = {"must_not": children}

    else:
        dsl = dict(kwargs)

    if "should" in dsl:
        dsl.update(minimum_should_match=1)

    dsl = {"bool": dict(dsl)}
    return dsl