Filters

build_filters_for_repeated_operators(must_clauses, should_clauses, must_not_clauses, qdrant_filter)

Flattens the nested lists of clauses by creating separate Filters for each clause of a logical operator.

Parameters:

Name	Type	Description	Default
must_clauses	list[Filter]	A nested list of must clauses or an empty list.	required
should_clauses	list[Filter]	A nested list of should clauses or an empty list.	required
must_not_clauses	list[Filter]	A nested list of must_not clauses or an empty list.	required
qdrant_filter	list[Filter]	A list where the generated Filter objects will be appended. This list will be modified in-place.	required

Returns:

Type	Description
list[Filter]	The modified qdrant_filter list with appended generated Filter objects.

Source code in dynamiq/storages/vector/qdrant/filters.py

def build_filters_for_repeated_operators(
    must_clauses: list[models.Filter],
    should_clauses: list[models.Filter],
    must_not_clauses: list[models.Filter],
    qdrant_filter: list[models.Filter],
) -> list[models.Filter]:
    """
    Flattens the nested lists of clauses by creating separate Filters for each clause of a logical
    operator.

    Args:
        must_clauses: A nested list of must clauses or an empty list.
        should_clauses: A nested list of should clauses or an empty list.
        must_not_clauses: A nested list of must_not clauses or an empty list.
        qdrant_filter: A list where the generated Filter objects will be appended. This list will be
            modified in-place.

    Returns:
        The modified `qdrant_filter` list with appended generated Filter objects.
    """

    if any(isinstance(i, list) for i in must_clauses):
        for i in must_clauses:
            qdrant_filter.append(
                models.Filter(
                    must=i or None,
                    should=should_clauses or None,
                    must_not=must_not_clauses or None,
                )
            )
    if any(isinstance(i, list) for i in should_clauses):
        for i in should_clauses:
            qdrant_filter.append(
                models.Filter(
                    must=must_clauses or None,
                    should=i or None,
                    must_not=must_not_clauses or None,
                )
            )
    if any(isinstance(i, list) for i in must_not_clauses):
        for i in must_clauses:
            qdrant_filter.append(
                models.Filter(
                    must=must_clauses or None,
                    should=should_clauses or None,
                    must_not=i or None,
                )
            )

    return qdrant_filter

convert_filters_to_qdrant(filter_term=None, is_parent_call=True)

Converts Dynamiq filters to the format used by Qdrant.

Parameters:

Name	Type	Description	Default
filter_term	list[dict] | dict | Filter | None	The Dynamiq filter to be converted to Qdrant.	None
is_parent_call	bool	Indicates if this is the top-level call to the function. If True, the function returns a single models.Filter object; if False, it may return a list of filters or conditions for further processing.	True

Returns:

Type	Description
Filter | list[Filter] | list[Condition] | None	A single Qdrant Filter in the parent call or a list of such Filters in recursive calls.

Raises:

Type	Description
VectorStoreFilterException	If the invalid filter criteria is provided or if an unknown operator is encountered.

Source code in dynamiq/storages/vector/qdrant/filters.py

def convert_filters_to_qdrant(
    filter_term: list[dict] | dict | models.Filter | None = None, is_parent_call: bool = True
) -> models.Filter | list[models.Filter] | list[models.Condition] | None:
    """Converts Dynamiq filters to the format used by Qdrant.

    Args:
        filter_term: The Dynamiq filter to be converted to Qdrant.
        is_parent_call: Indicates if this is the top-level call to the function. If True, the function
            returns a single models.Filter object; if False, it may return a list of filters or
            conditions for further processing.

    Returns:
        A single Qdrant Filter in the parent call or a list of such Filters in recursive calls.

    Raises:
        FilterError: If the invalid filter criteria is provided or if an unknown operator is
            encountered.
    """

    if isinstance(filter_term, models.Filter):
        return filter_term
    if not filter_term:
        return None

    must_clauses: list[models.Filter] = []
    should_clauses: list[models.Filter] = []
    must_not_clauses: list[models.Filter] = []
    # Indicates if there are multiple same LOGICAL OPERATORS on each level
    # and prevents them from being combined
    same_operator_flag = False
    conditions, qdrant_filter, current_level_operators = (
        [],
        [],
        [],
    )

    if isinstance(filter_term, dict):
        filter_term = [filter_term]

    # ======== IDENTIFY FILTER ITEMS ON EACH LEVEL ========

    for item in filter_term:
        operator = item.get("operator")

        # Check for repeated similar operators on each level
        same_operator_flag = operator in current_level_operators and operator in LOGICAL_OPERATORS
        if not same_operator_flag:
            current_level_operators.append(operator)

        if operator is None:
            msg = "Operator not found in filters"
            raise FilterError(msg)

        if operator in LOGICAL_OPERATORS and "conditions" not in item:
            msg = f"'conditions' not found for '{operator}'"
            raise FilterError(msg)

        if operator in LOGICAL_OPERATORS:
            # Recursively process nested conditions
            current_filter = convert_filters_to_qdrant(item.get("conditions", []), is_parent_call=False) or []

            # When same_operator_flag is set to True,
            # ensure each clause is appended as an independent list to avoid merging distinct clauses.
            if operator == "AND":
                must_clauses = [must_clauses, current_filter] if same_operator_flag else must_clauses + current_filter
            elif operator == "OR":
                should_clauses = (
                    [should_clauses, current_filter] if same_operator_flag else should_clauses + current_filter
                )
            elif operator == "NOT":
                must_not_clauses = (
                    [must_not_clauses, current_filter] if same_operator_flag else must_not_clauses + current_filter
                )

        elif operator in COMPARISON_OPERATORS:
            field = item.get("field")
            if not field.startswith("metadata."):
                field = f"metadata.{field}"
            value = item.get("value")
            if field is None or value is None:
                msg = f"'field' or 'value' not found for '{operator}'"
                raise FilterError(msg)

            parsed_conditions = _parse_comparison_operation(comparison_operation=operator, key=field, value=value)

            # check if the parsed_conditions are models.Filter or models.Condition
            for condition in parsed_conditions:
                if isinstance(condition, models.Filter):
                    qdrant_filter.append(condition)
                else:
                    conditions.append(condition)

        else:
            msg = f"Unknown operator {operator} used in filters"
            raise FilterError(msg)

    # ======== PROCESS FILTER ITEMS ON EACH LEVEL ========

    # If same logical operators have separate clauses, create separate filters
    if same_operator_flag:
        qdrant_filter = build_filters_for_repeated_operators(
            must_clauses, should_clauses, must_not_clauses, qdrant_filter
        )

    # else append a single Filter for existing clauses
    elif must_clauses or should_clauses or must_not_clauses:
        qdrant_filter.append(
            models.Filter(
                must=must_clauses or None,
                should=should_clauses or None,
                must_not=must_not_clauses or None,
            )
        )

    # In case of parent call, a single Filter is returned
    if is_parent_call:
        # If qdrant_filter has just a single Filter in parent call,
        # then it might be returned instead.
        if len(qdrant_filter) == 1 and isinstance(qdrant_filter[0], models.Filter):
            return qdrant_filter[0]
        else:
            must_clauses.extend(conditions)
            return models.Filter(
                must=must_clauses or None,
                should=should_clauses or None,
                must_not=must_not_clauses or None,
            )

    # Store conditions of each level in output of the loop
    elif conditions:
        qdrant_filter.extend(conditions)

    return qdrant_filter