Source code for swh.vault.interface

# Copyright (C) 2017-2020  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 typing import Any, Dict, List, Optional, Tuple

from typing_extensions import Protocol, runtime_checkable

from swh.core.api import remote_api_endpoint
from swh.model.swhids import CoreSWHID


[docs]@runtime_checkable class VaultInterface(Protocol): """ Backend Interface for the Software Heritage vault. """
[docs] @remote_api_endpoint("fetch") def fetch(self, bundle_type: str, swhid: CoreSWHID) -> Optional[bytes]: """Fetch information from a bundle""" ...
[docs] @remote_api_endpoint("cook") def cook( self, bundle_type: str, swhid: CoreSWHID, email: Optional[str] = None ) -> Dict[str, Any]: """Main entry point for cooking requests. This starts a cooking task if needed, and add the given e-mail to the notify list""" ...
[docs] @remote_api_endpoint("progress") def progress(self, bundle_type: str, swhid: CoreSWHID): ...
# Cookers endpoints
[docs] @remote_api_endpoint("set_progress") def set_progress(self, bundle_type: str, swhid: CoreSWHID, progress: str) -> None: """Set the cooking progress of a bundle""" ...
[docs] @remote_api_endpoint("set_status") def set_status(self, bundle_type: str, swhid: CoreSWHID, status: str) -> bool: """Set the cooking status of a bundle""" ...
[docs] @remote_api_endpoint("put_bundle") def put_bundle(self, bundle_type: str, swhid: CoreSWHID, bundle): """Store bundle in vault cache""" ...
[docs] @remote_api_endpoint("send_notif") def send_notif(self, bundle_type: str, swhid: CoreSWHID): """Send all the e-mails in the notification list of a bundle""" ...
# Batch endpoints
[docs] @remote_api_endpoint("batch_cook") def batch_cook(self, batch: List[Tuple[str, str]]) -> int: """Cook a batch of bundles and returns the cooking id.""" ...
[docs] @remote_api_endpoint("batch_progress") def batch_progress(self, batch_id: int) -> Dict[str, Any]: """Fetch information from a batch of bundles""" ...