Source code for swh.vault.in_memory_backend

# Copyright (C) 2017-2021  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 swh.model.swhids import CoreSWHID

from .cache import VaultCache


[docs]class InMemoryVaultBackend: """Stub vault backend, for use in the CLI.""" def __init__(self): self._cache = VaultCache(cls="memory")
[docs] def fetch(self, bundle_type: str, swhid: CoreSWHID) -> Optional[bytes]: return self._cache.get(bundle_type, swhid)
[docs] def cook( self, bundle_type: str, swhid: CoreSWHID, email: Optional[str] = None ) -> Dict[str, Any]: raise NotImplementedError("InMemoryVaultBackend.cook()")
[docs] def progress(self, bundle_type: str, swhid: CoreSWHID): raise NotImplementedError("InMemoryVaultBackend.progress()")
# Cookers endpoints
[docs] def set_progress(self, bundle_type: str, swhid: CoreSWHID, progress: str) -> None: pass
[docs] def set_status(self, bundle_type: str, swhid: CoreSWHID, status: str) -> None: pass
[docs] def put_bundle(self, bundle_type: str, swhid: CoreSWHID, bundle) -> bool: self._cache.add(bundle_type, swhid, bundle) return True
[docs] def send_notif(self, bundle_type: str, swhid: CoreSWHID): pass
# Batch endpoints
[docs] def batch_cook(self, batch: List[Tuple[str, str]]) -> int: raise NotImplementedError("InMemoryVaultBackend.batch_cook()")
[docs] def batch_progress(self, batch_id: int) -> Dict[str, Any]: pass