distro-tools/apollo/server/routes/advisories.py

from math import ceil

from tortoise import connections

from fastapi import APIRouter, Request, Depends
from fastapi.responses import HTMLResponse
from fastapi_pagination import Params
from fastapi_pagination.ext.tortoise import paginate, create_page

from apollo.db import Advisory
from apollo.server.utils import templates

router = APIRouter(tags=["non-api"])


@router.get(
    "/",
    response_class=HTMLResponse,
)
async def list_advisories(
    request: Request,
    params: Params = Depends(),
    search: str = None,
):
    params.size = 50
    if search:
        a = """
        with vars (search, size, page_offset) as (
            values ($1 :: text, $2 :: bigint, $3 :: bigint)
        )
        select
            a.id,
            a.created_at,
            a.updated_at,
            a.published_at,
            a.name,
            a.synopsis,
            a.description,
            a.kind,
            a.severity,
            a.red_hat_advisory_id,
            count(a.*) over () as total
        from
            advisories a
        left outer join advisory_affected_products ap on ap.advisory_id = a.id
        left outer join advisory_cves c on c.advisory_id = a.id
        left outer join advisory_fixes f on f.advisory_id = a.id
        where
            (select search from vars) is null or
            ap.name ilike '%' || (select search from vars) || '%' or
            a.synopsis ilike '%' || (select search from vars) || '%' or
            a.description ilike '%' || (select search from vars) || '%' or
            exists (select cve from advisory_cves where advisory_id = a.id and cve ilike '%' || (select search from vars) || '%') or
            exists (select ticket_id from advisory_fixes where advisory_id = a.id and ticket_id ilike '%' || (select search from vars) || '%') or
            a.name ilike '%' || (select search from vars) || '%'
        group by a.id
        order by a.published_at desc
        limit (select size from vars) offset (select page_offset from vars)
        """

        connection = connections.get("default")
        results = await connection.execute_query(
            a, [search, params.size, params.size * (params.page - 1)]
        )
        count = 0
        if results:
            if results[1]:
                count = results[1][0]["total"]

        advisories = create_page(
            results[1],
            count,
            params,
        )
    else:
        advisories = await paginate(
            Advisory.all().order_by("-published_at"),
            params=params,
        )
    return templates.TemplateResponse(
        "advisories.jinja", {
            "request": request,
            "params": params,
            "search": search if search else "",
            "advisories": advisories,
            "advisories_pages": ceil(advisories.total / advisories.size)
        }
    )


@router.get(
    "/{advisory_name}",
    response_class=HTMLResponse,
)
async def get_advisory(request: Request, advisory_name: str):
    advisory = await Advisory.get_or_none(name=advisory_name,
                                         ).prefetch_related(
                                             "red_hat_advisory",
                                             "packages",
                                             "cves",
                                             "fixes",
                                             "affected_products",
                                         )
    if advisory is None:
        return templates.TemplateResponse(
            "error.jinja", {
                "request": request,
                "message": "Requested advisory not found",
            }
        )

    package_map = {}
    for package in advisory.packages:
        name = f"{package.product_name} - {package.repo_name}"
        if name not in package_map:
            package_map[name] = []

        package_map[name].append(package.nevra)

    return templates.TemplateResponse(
        "advisory.jinja", {
            "request": request,
            "title": f"Advisory {advisory.id}",
            "advisory": advisory,
            "package_map": package_map,
        }
    )