def _documents_for()

in kitsune/wiki/facets.py [0:0]

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

def _documents_for(user, locale, topics=None, products=None):
    """Returns a list of articles that apply to passed in locale, topics and products."""
    cache_key = _cache_key(locale, topics, products)

    if not user.is_authenticated:
        # For anonymous users, first check the cache.
        documents_cache_key = f"documents_for_v2:{cache_key}"
        documents = cache.get(documents_cache_key)
        if documents is not None:
            return documents

    qs = Document.objects.visible(
        user,
        locale=locale,
        is_archived=False,
        current_revision__isnull=False,
        category__in=settings.IA_DEFAULT_CATEGORIES,
    )
    # speed up query by removing any ordering, since we're doing it in python:
    qs = qs.select_related("current_revision", "parent").order_by()

    if topics:
        topic_ids = [t.id for t in topics]
        # For parent documents: include if they have the requested topics
        # For translations: include ONLY if their parent has the requested topics,
        # completely ignoring any topics directly assigned to the translation
        qs = qs.filter(
            # Either this is a parent document with matching topics
            (Q(parent__isnull=True) & Q(topics__in=topic_ids))
            # OR this is a translation and its parent has matching topics
            | (Q(parent__isnull=False) & Q(parent__topics__in=topic_ids))
        )

    for product in products or []:
        # we need to filter against parent products for localized articles
        qs = qs.filter(Q(products=product) | Q(parent__products=product))

    qs = qs.distinct()

    votes_cache_key = f"votes_for:{cache_key}"
    votes_dict = cache.get(votes_cache_key)
    if votes_dict is None:
        # NOTE: It's important to use "created__range" rather than "created__gt"
        #       with Postgres, otherwise it won't use the index on the "created"
        #       field, and the "HelpfulVote" query will be massively slower.
        votes_query = (
            HelpfulVote.objects.filter(
                revision_id__in=qs.values_list("current_revision_id", flat=True),
                created__range=(Now() - timedelta(days=30), Now()),
                helpful=True,
            )
            .values("revision_id")
            .annotate(count=Count("*"))
            .values("revision_id", "count")
        )
        votes_dict = {row["revision_id"]: row["count"] for row in votes_query}
        # the votes query is rather expensive, and only used for ordering,
        # so we can cache it rather aggressively
        cache.set(votes_cache_key, votes_dict, timeout=settings.CACHE_LONG_TIMEOUT)

    # Annotate each of the documents with its string of product titles. This must
    # be a sub-query in order to free itself from the product filter(s) above.
    qs = qs.annotate(
        product_titles=Subquery(
            Document.objects.filter(pk=OuterRef("pk"))
            .annotate(
                product_titles=Case(
                    When(
                        parent__isnull=False,
                        then=StringAgg(
                            "parent__products__title",
                            delimiter=", ",
                            ordering="parent__products__title",
                        ),
                    ),
                    default=StringAgg(
                        "products__title", delimiter=", ", ordering="products__title"
                    ),
                ),
            )
            .values("product_titles")
        )
    )

    doc_dicts = []
    for d in qs:
        doc_dicts.append(
            dict(
                id=d.id,
                document_title=d.title,
                url=d.get_absolute_url(),
                document_parent_id=d.parent_id,
                created=d.current_revision.created,
                product_titles=d.product_titles,
                document_summary=d.current_revision.summary,
                display_order=d.original.display_order,
                helpful_votes=votes_dict.get(d.current_revision_id, 0),
            )
        )

    # sort the results by ascending display_order and descending votes
    doc_dicts.sort(key=lambda x: (x["display_order"], -x["helpful_votes"]))

    if not user.is_authenticated:
        cache.set(documents_cache_key, doc_dicts)

    return doc_dicts