Source code for

# Copyright (C) 2020-2022  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 Any, Dict, Iterable, List, Optional, TypeVar

from typing_extensions import Protocol, 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]


[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(Protocol):
[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("origin/get") def origin_get(self, url: str) -> Optional[Dict[str, Any]]: """Returns the full documents associated to the given origin URLs. Order is arbitrary; unknown origins are not returned. """
[docs] @remote_api_endpoint("origin/delete") def origin_delete(self, url: str) -> bool: """Remove the documents associated with the given origin URL. Returns: True if the document was removed, False if it could not be found. """ ...
[docs] @remote_api_endpoint("visit_types_count") def visit_types_count(self) -> Counter: """Returns origin counts per visit type (git, hg, svn, ...).""" ...