Source code for swh.vault.cache

# Copyright (C) 2016-2023  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 datetime import timedelta
from typing import Optional

from swh.model import hashutil
from swh.model.swhids import CoreSWHID
from swh.objstorage.factory import get_objstorage
from swh.objstorage.interface import ObjId



[docs]
class VaultCache:
    """The Vault cache is an object storage that stores Vault bundles.

    This implementation computes sha1('<bundle_type>:<swhid>') as the
    internal identifiers used in the underlying objstorage.
    """

    def __init__(self, **objstorage):
        self.objstorage = get_objstorage(**objstorage)


[docs]
    def add(self, bundle_type, swhid: CoreSWHID, content) -> None:
        sid = self._get_internal_id(bundle_type, swhid)
        self.objstorage.add(content, sid)



[docs]
    def get(self, bundle_type, swhid: CoreSWHID) -> bytes:
        sid = self._get_internal_id(bundle_type, swhid)
        return self.objstorage.get(sid)



[docs]
    def download_url(
        self,
        bundle_type,
        swhid: CoreSWHID,
        content_disposition: Optional[str] = None,
        expiry: Optional[timedelta] = None,
    ) -> Optional[str]:
        sid = self._get_internal_id(bundle_type, swhid)
        return self.objstorage.download_url(sid, content_disposition, expiry)



[docs]
    def delete(self, bundle_type, swhid: CoreSWHID):
        sid = self._get_internal_id(bundle_type, swhid)
        return self.objstorage.delete(sid)



[docs]
    def is_cached(self, bundle_type, swhid: CoreSWHID) -> bool:
        sid = self._get_internal_id(bundle_type, swhid)
        return sid in self.objstorage


    def _get_internal_id(self, bundle_type, swhid: CoreSWHID) -> ObjId:
        key = "{}:{}".format(bundle_type, swhid).encode()
        return hashutil.MultiHash.from_data(key).digest()