takahe/api/views/search.py

from typing import Literal

from hatchway import Field, api_view

from activities.models import PostInteraction
from activities.services.search import SearchService
from api import schemas
from api.decorators import scope_required


@scope_required("read")
@api_view.get
def search(
    request,
    q: str,
    type: Literal["accounts", "hashtags", "statuses"] | None = None,
    fetch_identities: bool = Field(False, alias="resolve"),
    following: bool = False,
    exclude_unreviewed: bool = False,
    account_id: str | None = None,
    max_id: str | None = None,
    since_id: str | None = None,
    min_id: str | None = None,
    limit: int = 20,
    offset: int = 0,
) -> schemas.Search:
    if limit > 40:
        limit = 40
    result: dict[str, list] = {"accounts": [], "statuses": [], "hashtags": []}
    # We don't support pagination for searches yet
    if max_id or since_id or min_id or offset:
        return schemas.Search(**result)
    # Run search
    searcher = SearchService(q, request.identity)
    search_result = searcher.search_all()
    if type is None or type == "accounts":
        result["accounts"] = [
            schemas.Account.from_identity(i, include_counts=False)
            for i in search_result["identities"]
        ]
    if type is None or type == "hashtag":
        result["hashtags"] = [
            schemas.Tag.from_hashtag(h) for h in search_result["hashtags"]
        ]
    if type is None or type == "statuses":
        interactions = PostInteraction.get_post_interactions(
            search_result["posts"], request.identity
        )
        result["statuses"] = [
            schemas.Status.from_post(
                p, interactions=interactions, identity=request.identity
            )
            for p in search_result["posts"]
        ]
    return schemas.Search(**result)