Source code for swh.vault.api.client

# Copyright (C) 2016-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 swh.core.api import RPCClient
from swh.model import hashutil
from swh.vault.exc import NotFoundExc

[docs]class RemoteVaultClient(RPCClient): """Client to the Software Heritage vault cache.""" reraise_exceptions = [NotFoundExc] # Web API endpoints
[docs] def fetch(self, obj_type, obj_id): hex_id = hashutil.hash_to_hex(obj_id) return self.get("fetch/{}/{}".format(obj_type, hex_id))
[docs] def cook(self, obj_type, obj_id, email=None): hex_id = hashutil.hash_to_hex(obj_id) return "cook/{}/{}".format(obj_type, hex_id), data={}, params=({"email": email} if email else None), )
[docs] def progress(self, obj_type, obj_id): hex_id = hashutil.hash_to_hex(obj_id) return self.get("progress/{}/{}".format(obj_type, hex_id))
# Cookers endpoints
[docs] def set_progress(self, obj_type, obj_id, progress): hex_id = hashutil.hash_to_hex(obj_id) return"set_progress/{}/{}".format(obj_type, hex_id), data=progress)
[docs] def set_status(self, obj_type, obj_id, status): hex_id = hashutil.hash_to_hex(obj_id) return"set_status/{}/{}".format(obj_type, hex_id), data=status)
# TODO: handle streaming properly
[docs] def put_bundle(self, obj_type, obj_id, bundle): hex_id = hashutil.hash_to_hex(obj_id) return"put_bundle/{}/{}".format(obj_type, hex_id), data=bundle)
[docs] def send_notif(self, obj_type, obj_id): hex_id = hashutil.hash_to_hex(obj_id) return"send_notif/{}/{}".format(obj_type, hex_id), data=None)
# Batch endpoints
[docs] def batch_cook(self, batch): return"batch_cook", data=batch)
[docs] def batch_progress(self, batch_id): return self.get("batch_progress/{}".format(batch_id))