Source code for swh.search.interface

# Copyright (C) 2020-2021  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information

from collections import Counter
from typing import Iterable, List, Optional, TypeVar

from typing_extensions import TypedDict

from swh.core.api import remote_api_endpoint
from swh.core.api.classes import PagedResult as CorePagedResult

TResult = TypeVar("TResult")
PagedResult = CorePagedResult[TResult, str]

SORT_BY_OPTIONS = [
    "nb_visits",
    "last_visit_date",
    "last_eventful_visit_date",
    "last_revision_date",
    "last_release_date",
    "date_created",
    "date_modified",
    "date_published",
]


[docs]class MinimalOriginDict(TypedDict): """Mandatory keys of an :class:`OriginDict`""" url: str
[docs]class OriginDict(MinimalOriginDict, total=False): """Argument passed to :meth:`SearchInterface.origin_update`.""" visit_types: List[str] has_visits: bool
[docs]class SearchInterface:
[docs] @remote_api_endpoint("check") def check(self): """Dedicated method to execute some specific check per implementation. """ ...
[docs] @remote_api_endpoint("flush") def flush(self) -> None: """Blocks until all previous calls to _update() are completely applied. """ ...
[docs] @remote_api_endpoint("origin/update") def origin_update(self, documents: Iterable[OriginDict]) -> None: """Persist documents to the search backend. """ ...
[docs] @remote_api_endpoint("visit_types_count") def visit_types_count(self) -> Counter: """Returns origin counts per visit type (git, hg, svn, ...). """ ...