Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/objstorage.py
# Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 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 Dict, Generator, Iterable | from typing import Dict, Iterable, Optional | ||||
from swh.storage.interface import Sha1 | |||||
from swh.model.model import Content, MissingData | from swh.model.model import Content, MissingData | ||||
from swh.objstorage import get_objstorage | from swh.objstorage import get_objstorage | ||||
from swh.objstorage.exc import ObjNotFoundError | from swh.objstorage.exc import ObjNotFoundError | ||||
from .exc import StorageArgumentException | from .exc import StorageArgumentException | ||||
class ObjStorage: | class ObjStorage: | ||||
"""Objstorage collaborator in charge of adding objects to | """Objstorage collaborator in charge of adding objects to | ||||
the objstorage. | the objstorage. | ||||
""" | """ | ||||
def __init__(self, objstorage_config: Dict): | def __init__(self, objstorage_config: Dict): | ||||
self.objstorage = get_objstorage(**objstorage_config) | self.objstorage = get_objstorage(**objstorage_config) | ||||
def __getattr__(self, key): | def __getattr__(self, key): | ||||
if key == "objstorage": | if key == "objstorage": | ||||
raise AttributeError(key) | raise AttributeError(key) | ||||
return getattr(self.objstorage, key) | return getattr(self.objstorage, key) | ||||
def content_get(self, contents: Iterable[bytes]) -> Generator: | def content_get(self, obj_id: Sha1) -> Optional[bytes]: | ||||
"""Retrieve content data from the objstorage | """Retrieve data associated to the content from the objstorage | ||||
Args: | Args: | ||||
contents: List of contents to retrieve data from | content: content identitier | ||||
Returns: | |||||
associated content's data if any, None otherwise. | |||||
""" | """ | ||||
for obj_id in contents: | |||||
try: | try: | ||||
data = self.objstorage.get(obj_id) | data = self.objstorage.get(obj_id) | ||||
except ObjNotFoundError: | except ObjNotFoundError: | ||||
yield None | data = None | ||||
continue | |||||
yield {"sha1": obj_id, "data": data} | return data | ||||
def content_add(self, contents: Iterable[Content]) -> Dict: | def content_add(self, contents: Iterable[Content]) -> Dict: | ||||
"""Add contents to the objstorage. | """Add contents to the objstorage. | ||||
Args: | Args: | ||||
contents: List of contents to add1 | contents: List of contents to add1 | ||||
Returns: | Returns: | ||||
Show All 13 Lines |