def _get_aggregate_response()

in libmozdata/buildhub.py [0:0]

• 48 lines of code
• 10 McCabe index (conditional complexity)

def _get_aggregate_response(field, **kwargs):
    search_url = kwargs.pop("_search_url", SEARCH_URL)
    verbose = kwargs.pop("_verbose", False)
    size = kwargs.pop("_size", 1000)

    filter_ = defaultdict(dict)
    filters = []

    if "product" in kwargs:
        # To avoid the rist of someone search for `product="Firefox"` when
        # they should have done `product="firefox"`.
        product = kwargs["product"]
        assert isinstance(product, str), type(product)
        product = product.lower()
        filters.append({"source.product": product})

    if "channel" in kwargs:
        channel = kwargs["channel"]
        assert isinstance(channel, str), type(channel)
        channel = channel.lower()
        filters.append({"target.channel": channel})

    include = ".*"
    if "startswith" in kwargs:
        include = kwargs.pop("startswith")
        include = "{}.*".format(re.escape(include))

    if len(filters) > 1:
        filter_ = {"bool": {"must": [{"term": x} for x in filters]}}
    elif filters:
        filter_ = {"term": filters[0]}
    else:
        filter_ = {"match_all": {}}

    search = {
        "aggs": {
            "myaggs": {
                "filter": filter_ or {"match_all": {}},
                "aggs": {
                    field: {
                        "terms": {
                            "field": field,
                            "size": size,
                            "order": {"_term": "desc"},
                            "include": include,
                        }
                    },
                    # "target.version_count": {
                    #     "cardinality": {"field": "target.version"}
                    # },
                },
            }
        },
        "size": 0,
    }

    if verbose:
        print(json.dumps(search, indent=3))

    response = fetch(search_url, search)
    return response