Changeset View
Changeset View
Standalone View
Standalone View
swh/vault/interface.py
# Copyright (C) 2017-2020 The Software Heritage developers | # Copyright (C) 2017-2020 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
from typing import Any, Dict, List, Optional, Tuple, Union | from typing import Any, Dict, List, Optional, Tuple | ||||
from typing_extensions import Protocol, runtime_checkable | from typing_extensions import Protocol, runtime_checkable | ||||
from swh.core.api import remote_api_endpoint | from swh.core.api import remote_api_endpoint | ||||
from swh.model.identifiers import CoreSWHID | |||||
ObjectId = Union[str, bytes] | |||||
@runtime_checkable | @runtime_checkable | ||||
class VaultInterface(Protocol): | class VaultInterface(Protocol): | ||||
""" | """ | ||||
Backend Interface for the Software Heritage vault. | Backend Interface for the Software Heritage vault. | ||||
""" | """ | ||||
@remote_api_endpoint("fetch") | @remote_api_endpoint("fetch") | ||||
def fetch(self, bundle_type: str, obj_id: ObjectId) -> Optional[bytes]: | def fetch(self, bundle_type: str, swhid: CoreSWHID) -> Optional[bytes]: | ||||
"""Fetch information from a bundle""" | """Fetch information from a bundle""" | ||||
... | ... | ||||
@remote_api_endpoint("cook") | @remote_api_endpoint("cook") | ||||
def cook( | def cook( | ||||
self, bundle_type: str, obj_id: ObjectId, email: Optional[str] = None | self, bundle_type: str, swhid: CoreSWHID, email: Optional[str] = None | ||||
) -> Dict[str, Any]: | ) -> Dict[str, Any]: | ||||
"""Main entry point for cooking requests. This starts a cooking task if | """Main entry point for cooking requests. This starts a cooking task if | ||||
needed, and add the given e-mail to the notify list""" | needed, and add the given e-mail to the notify list""" | ||||
... | ... | ||||
@remote_api_endpoint("progress") | @remote_api_endpoint("progress") | ||||
def progress(self, bundle_type: str, obj_id: ObjectId): | def progress(self, bundle_type: str, swhid: CoreSWHID): | ||||
... | ... | ||||
# Cookers endpoints | # Cookers endpoints | ||||
@remote_api_endpoint("set_progress") | @remote_api_endpoint("set_progress") | ||||
def set_progress(self, bundle_type: str, obj_id: ObjectId, progress: str) -> None: | def set_progress(self, bundle_type: str, swhid: CoreSWHID, progress: str) -> None: | ||||
"""Set the cooking progress of a bundle""" | """Set the cooking progress of a bundle""" | ||||
... | ... | ||||
@remote_api_endpoint("set_status") | @remote_api_endpoint("set_status") | ||||
def set_status(self, bundle_type: str, obj_id: ObjectId, status: str) -> bool: | def set_status(self, bundle_type: str, swhid: CoreSWHID, status: str) -> bool: | ||||
"""Set the cooking status of a bundle""" | """Set the cooking status of a bundle""" | ||||
... | ... | ||||
@remote_api_endpoint("put_bundle") | @remote_api_endpoint("put_bundle") | ||||
def put_bundle(self, bundle_type: str, obj_id: ObjectId, bundle): | def put_bundle(self, bundle_type: str, swhid: CoreSWHID, bundle): | ||||
"""Store bundle in vault cache""" | """Store bundle in vault cache""" | ||||
... | ... | ||||
@remote_api_endpoint("send_notif") | @remote_api_endpoint("send_notif") | ||||
def send_notif(self, bundle_type: str, obj_id: ObjectId): | def send_notif(self, bundle_type: str, swhid: CoreSWHID): | ||||
"""Send all the e-mails in the notification list of a bundle""" | """Send all the e-mails in the notification list of a bundle""" | ||||
... | ... | ||||
# Batch endpoints | # Batch endpoints | ||||
@remote_api_endpoint("batch_cook") | @remote_api_endpoint("batch_cook") | ||||
def batch_cook(self, batch: List[Tuple[str, str]]) -> int: | def batch_cook(self, batch: List[Tuple[str, str]]) -> int: | ||||
"""Cook a batch of bundles and returns the cooking id.""" | """Cook a batch of bundles and returns the cooking id.""" | ||||
... | ... | ||||
@remote_api_endpoint("batch_progress") | @remote_api_endpoint("batch_progress") | ||||
def batch_progress(self, batch_id: int) -> Dict[str, Any]: | def batch_progress(self, batch_id: int) -> Dict[str, Any]: | ||||
"""Fetch information from a batch of bundles""" | """Fetch information from a batch of bundles""" | ||||
... | ... |