Changeset View
Changeset View
Standalone View
Standalone View
swh/vault/in_memory_backend.py
| Show All 13 Lines | |||||
| class InMemoryVaultBackend: | class InMemoryVaultBackend: | ||||
| """Stub vault backend, for use in the CLI.""" | """Stub vault backend, for use in the CLI.""" | ||||
| def __init__(self): | def __init__(self): | ||||
| self._cache = VaultCache(cls="memory") | self._cache = VaultCache(cls="memory") | ||||
| def fetch(self, obj_type: str, obj_id: ObjectId) -> Optional[bytes]: | def fetch(self, obj_type: str, obj_id: ObjectId) -> Optional[bytes]: | ||||
| return self._cache.get(obj_type, obj_id) | return self._cache.get(obj_type, hash_to_bytes(obj_id)) | ||||
| def cook( | def cook( | ||||
| self, obj_type: str, obj_id: ObjectId, email: Optional[str] = None | self, obj_type: str, obj_id: ObjectId, email: Optional[str] = None | ||||
| ) -> Dict[str, Any]: | ) -> Dict[str, Any]: | ||||
| raise NotImplementedError("InMemoryVaultBackend.cook()") | raise NotImplementedError("InMemoryVaultBackend.cook()") | ||||
| def progress(self, obj_type: str, obj_id: ObjectId): | def progress(self, obj_type: str, obj_id: ObjectId): | ||||
| raise NotImplementedError("InMemoryVaultBackend.progress()") | raise NotImplementedError("InMemoryVaultBackend.progress()") | ||||
| Show All 23 Lines | |||||