INNER CODE UNIT · Python

_build_selector_query

robusta-dev/krr · robusta_krr/core/integrations/kubernetes/__init__.py:203

    def _build_selector_query(selector: Any) -> Union[str, None]:
        label_filters = []

        if selector.match_labels is not None:
            label_filters += [f"{label[0]}={label[1]}" for label in selector.match_labels.items()]

        if selector.match_expressions is not None:
            label_filters += [
                ClusterLoader._get_match_expression_filter(expression) for expression in selector.match_expressions
            ]

        # normally the kubernetes API client renames matchLabels to match_labels in python
        # but for CRDs like ArgoRollouts that renaming doesn't happen and we have selector={'matchLabels': {'app': 'test-app'}}
        if getattr(selector, "matchLabels", None):
            label_filters += [f"{label[0]}={label[1]}" for label in getattr(selector, "matchLabels").items()]
        if getattr(selector, "matchExpressions", None):
            label_filters += [
                ClusterLoader._get_match_expression_filter(expression)

View source record →