diff --git a/clients/python-wrapper/docs/lakefs.object_manager.rst b/clients/python-wrapper/docs/lakefs.object_manager.rst deleted file mode 100644 index 97ceb3df9e4..00000000000 --- a/clients/python-wrapper/docs/lakefs.object_manager.rst +++ /dev/null @@ -1,7 +0,0 @@ -lakefs.object\_manager module -============================= - -.. automodule:: lakefs.object_manager - :members: - :undoc-members: - :show-inheritance: diff --git a/clients/python-wrapper/docs/lakefs.rst b/clients/python-wrapper/docs/lakefs.rst index df999d43025..b75ad4c681b 100644 --- a/clients/python-wrapper/docs/lakefs.rst +++ b/clients/python-wrapper/docs/lakefs.rst @@ -15,7 +15,6 @@ Submodules lakefs.models lakefs.namedtuple lakefs.object - lakefs.object_manager lakefs.reference lakefs.repository lakefs.tag diff --git a/clients/python-wrapper/lakefs/__init__.py b/clients/python-wrapper/lakefs/__init__.py index 7515110a78e..4810f6bc594 100644 --- a/clients/python-wrapper/lakefs/__init__.py +++ b/clients/python-wrapper/lakefs/__init__.py @@ -4,8 +4,15 @@ from lakefs.repository import Repository from lakefs.reference import Reference -from lakefs.models import Commit, Change, RepositoryProperties +from lakefs.models import ( + Commit, + Change, + ImportStatus, + ServerStorageConfiguration, + ObjectInfo, + CommonPrefix, + RepositoryProperties +) from lakefs.tag import Tag from lakefs.branch import Branch from lakefs.object import StoredObject, WriteableObject, ObjectReader -from lakefs.object_manager import ObjectManager diff --git a/clients/python-wrapper/lakefs/import_manager.py b/clients/python-wrapper/lakefs/import_manager.py index 0ff4490864b..e543cb7fdee 100644 --- a/clients/python-wrapper/lakefs/import_manager.py +++ b/clients/python-wrapper/lakefs/import_manager.py @@ -10,13 +10,10 @@ import lakefs_sdk -from lakefs.models import ImportStatus +from lakefs.models import ImportStatus, _OBJECT, _COMMON_PREFIX from lakefs.client import Client, DEFAULT_CLIENT from lakefs.exceptions import ImportManagerException, api_exception_handler -_PREFIX = "common_prefix" -_OBJECT = "object" - class ImportManager: """ @@ -57,7 +54,8 @@ def prefix(self, object_store_uri: str, destination: str) -> ImportManager: :param destination: The destination prefix relative to the branch :return: The ImportManager instance (self) after update, to allow operations chaining """ - self.sources.append(lakefs_sdk.ImportLocation(type=_PREFIX, path=object_store_uri, destination=destination)) + self.sources.append( + lakefs_sdk.ImportLocation(type=_COMMON_PREFIX, path=object_store_uri, destination=destination)) return self def object(self, object_store_uri: str, destination: str) -> ImportManager: diff --git a/clients/python-wrapper/lakefs/models.py b/clients/python-wrapper/lakefs/models.py index 11e93836415..40a1f06345c 100644 --- a/clients/python-wrapper/lakefs/models.py +++ b/clients/python-wrapper/lakefs/models.py @@ -9,6 +9,9 @@ from lakefs.namedtuple import LenientNamedTuple +_COMMON_PREFIX = "common_prefix" +_OBJECT = "object" + class Commit(LenientNamedTuple): """ @@ -74,12 +77,11 @@ class ServerStorageConfiguration(LenientNamedTuple): default_namespace_prefix: Optional[str] = None -class ObjectStats(LenientNamedTuple): +class ObjectInfo(LenientNamedTuple): """ Represent a lakeFS object's stats """ path: str - path_type: str physical_address: str checksum: str mtime: int @@ -89,6 +91,13 @@ class ObjectStats(LenientNamedTuple): content_type: Optional[str] = None +class CommonPrefix(LenientNamedTuple): + """ + Represents a common prefix in lakeFS + """ + path: str + + class RepositoryProperties(LenientNamedTuple): """ Represent a lakeFS repository's properties diff --git a/clients/python-wrapper/lakefs/object.py b/clients/python-wrapper/lakefs/object.py index a955ffd97aa..910e6990a1b 100644 --- a/clients/python-wrapper/lakefs/object.py +++ b/clients/python-wrapper/lakefs/object.py @@ -30,7 +30,7 @@ PermissionException, ObjectExistsException, ) -from lakefs.models import ObjectStats +from lakefs.models import ObjectInfo _LAKEFS_METADATA_PREFIX = "x-lakefs-meta-" # _BUFFER_SIZE - Writer buffer size. While buffer size not exceed, data will be maintained in memory and file will @@ -302,7 +302,7 @@ class ObjectWriter(LakeFSIOBase): implicitly when using writer as a context. """ _fd: tempfile.SpooledTemporaryFile - _obj_stats: ObjectStats = None + _obj_stats: ObjectInfo = None def __init__(self, obj: StoredObject, @@ -381,7 +381,7 @@ def close(self) -> None: Write the data to the lakeFS server and close open descriptors """ stats = self._upload_presign() if self.pre_sign else self._upload_raw() - self._obj_stats = ObjectStats(**stats.dict()) + self._obj_stats = ObjectInfo(**stats.dict()) self._fd.close() super().close() @@ -502,7 +502,7 @@ class StoredObject: _repo_id: str _ref_id: str _path: str - _stats: Optional[ObjectStats] = None + _stats: Optional[ObjectInfo] = None def __init__(self, repository: str, reference: str, path: str, client: Optional[Client] = DEFAULT_CLIENT): self._client = client @@ -548,14 +548,14 @@ def reader(self, mode: ReadModes = 'rb', pre_sign: Optional[bool] = None) -> Obj """ return ObjectReader(self, mode=mode, pre_sign=pre_sign, client=self._client) - def stat(self) -> ObjectStats: + def stat(self) -> ObjectInfo: """ Return the Stat object representing this object """ if self._stats is None: with api_exception_handler(_io_exception_handler): stat = self._client.sdk_client.objects_api.stat_object(self._repo_id, self._ref_id, self._path) - self._stats = ObjectStats(**stat.dict()) + self._stats = ObjectInfo(**stat.dict()) return self._stats def exists(self) -> bool: diff --git a/clients/python-wrapper/lakefs/object_manager.py b/clients/python-wrapper/lakefs/object_manager.py deleted file mode 100644 index d1eee06c1a7..00000000000 --- a/clients/python-wrapper/lakefs/object_manager.py +++ /dev/null @@ -1,10 +0,0 @@ -""" -Module implementing reference's object management logic -""" - - -class ObjectManager: - """ - Manage objects listing for a given repository - """ - # TODO: Implement diff --git a/clients/python-wrapper/lakefs/reference.py b/clients/python-wrapper/lakefs/reference.py index 36c77fd9ed7..79d8aa504bc 100644 --- a/clients/python-wrapper/lakefs/reference.py +++ b/clients/python-wrapper/lakefs/reference.py @@ -8,11 +8,10 @@ import lakefs_sdk -from lakefs.models import Commit, Change +from lakefs.models import Commit, Change, CommonPrefix, ObjectInfo, _OBJECT from lakefs.client import Client, DEFAULT_CLIENT from lakefs.exceptions import api_exception_handler from lakefs.object import StoredObject -from lakefs.object_manager import ObjectManager class Reference: @@ -43,12 +42,37 @@ def id(self) -> str: """ return self._id - @property - def objects(self) -> ObjectManager: + def objects(self, + max_amount: Optional[int] = None, + after: Optional[str] = None, + prefix: Optional[str] = None, + delimiter: Optional[str] = None, + **kwargs) -> Generator[StoredObject | CommonPrefix]: """ - Returns a ObjectManager object for this reference + Returns an object generator for this reference, the generator can yield either a StoredObject or a CommonPrefix + object depending on the listing parameters provided. + + :param max_amount: Stop showing changes after this amount + :param after: Return items after this value + :param prefix: Return items prefixed with this value + :param delimiter: Group common prefixes by this delimiter + :param kwargs: Additional Keyword Arguments to send to the server + :raises: + NotFoundException if this reference or other_ref does not exist + NotAuthorizedException if user is not authorized to perform this operation + ServerException for any other errors """ - # TODO: Implement + + for res in generate_listing(self._client.sdk_client.objects_api.list_objects, + repository=self._repo_id, + ref=self._id, + max_amount=max_amount, + after=after, + prefix=prefix, + delimiter=delimiter, + **kwargs): + type_class = ObjectInfo if res.path_type == _OBJECT else CommonPrefix + yield type_class(**res.dict()) def log(self, max_amount: Optional[int] = None, **kwargs) -> Generator[Commit]: """ diff --git a/clients/python-wrapper/tests/.pylintrc b/clients/python-wrapper/tests/.pylintrc index 79a51c08326..278290b517f 100644 --- a/clients/python-wrapper/tests/.pylintrc +++ b/clients/python-wrapper/tests/.pylintrc @@ -12,7 +12,6 @@ disable= missing-function-docstring, missing-module-docstring, too-few-public-methods, - pointless-statement, no-member, fixme, diff --git a/clients/python-wrapper/tests/integration/test_branch.py b/clients/python-wrapper/tests/integration/test_branch.py index 8dcbfb65a4a..2fa763e8947 100644 --- a/clients/python-wrapper/tests/integration/test_branch.py +++ b/clients/python-wrapper/tests/integration/test_branch.py @@ -50,9 +50,8 @@ def test_reset_changes(setup_repo): def test_delete_object_changes(setup_repo): _, repo = setup_repo test_branch = repo.branch("main") - path_and_data = ["a", "b", "bar/a", "bar/b", "bar/c", "c", "foo/a", "foo/b", "foo/c", ] - for s in path_and_data: - test_branch.object(s).upload(s) + path_and_data = ["a", "b", "bar/a", "bar/b", "bar/c", "c", "foo/a", "foo/b", "foo/c"] + upload_data(test_branch, path_and_data) test_branch.commit("add some files", {"test_key": "test_value"}) test_branch.delete_objects("foo/a") diff --git a/clients/python-wrapper/tests/integration/test_reference.py b/clients/python-wrapper/tests/integration/test_reference.py index e0a749b0058..82d2244382e 100644 --- a/clients/python-wrapper/tests/integration/test_reference.py +++ b/clients/python-wrapper/tests/integration/test_reference.py @@ -57,3 +57,24 @@ def test_reference_merge_into(setup_branch_with_commits): branch.merge_into(other_branch.id, message="Merge2") assert other_branch.commit_message() == "Merge2" assert list(other_branch.log(max_amount=3))[2].id == commits[0].id + + +def test_reference_objects(setup_repo): + _, repo = setup_repo + test_branch = repo.branch("main") + path_and_data = ["a", "b", "bar/a", "bar/b", "bar/c", "c", "foo/a", "foo/b", "foo/c"] + for s in path_and_data: + test_branch.object(s).upload(s) + + objects = list(test_branch.objects()) + assert len(objects) == len(path_and_data) + for obj in objects: + assert obj.path in path_and_data + + expected = ["a", "b", "bar/", "c", "foo/"] + i = 0 + for obj in test_branch.objects(delimiter='/'): + i += 1 + assert obj.path in expected + + assert i == len(expected) diff --git a/clients/python-wrapper/tests/integration/test_sanity.py b/clients/python-wrapper/tests/integration/test_sanity.py index bfe9efdfa58..a45ffbbe6fa 100644 --- a/clients/python-wrapper/tests/integration/test_sanity.py +++ b/clients/python-wrapper/tests/integration/test_sanity.py @@ -110,7 +110,6 @@ def test_object_sanity(setup_repo): stats = obj.stat() assert stats.path == path == obj.path - assert stats.path_type == "object" assert stats.mtime <= time.time() assert stats.size_bytes == len(data) assert stats.metadata == metadata diff --git a/clients/python-wrapper/tests/utests/test_namedtuple.py b/clients/python-wrapper/tests/utests/test_namedtuple.py index 265187208f2..ecf42fe8804 100644 --- a/clients/python-wrapper/tests/utests/test_namedtuple.py +++ b/clients/python-wrapper/tests/utests/test_namedtuple.py @@ -36,10 +36,10 @@ def test_namedtuple(): assert not nt2.field3 with expect_exception_context(AttributeError): - nt2.field4 + nt2.field4 # pylint: disable=pointless-statement with expect_exception_context(AttributeError): - nt2.field5 + nt2.field5 # pylint: disable=pointless-statement # Verify extra kwargs are in 'unknown' dict assert nt2.unknown['field4'] == "something" diff --git a/clients/python-wrapper/tests/utests/test_reference.py b/clients/python-wrapper/tests/utests/test_reference.py index 8ba925d9b8e..1e36e261964 100644 --- a/clients/python-wrapper/tests/utests/test_reference.py +++ b/clients/python-wrapper/tests/utests/test_reference.py @@ -1,8 +1,8 @@ import lakefs_sdk -from tests.utests.common import get_test_client - +from lakefs import ObjectInfo, CommonPrefix from lakefs.repository import Repository +from tests.utests.common import get_test_client, expect_exception_context def get_test_ref(): @@ -118,3 +118,47 @@ def monkey_diff_refs(*_, **__): idx = 0 max_amount = pages * items_per_page * 2 assert len(list(ref.diff(other_ref="other_ref", max_amount=max_amount))) == pages * items_per_page + + +def test_reference_objects(monkeypatch): + ref = get_test_ref() + with monkeypatch.context(): + def monkey_list_objects(*_, **__): + results = [] + for i in range(10): + if i % 2: + results.append(lakefs_sdk.ObjectStats( + path=f"path-{i}", + path_type="object", + physical_address=f"address-{i}", + checksum=f"{i}", + size_bytes=i, + mtime=i, + )) + else: + results.append(lakefs_sdk.ObjectStats( + path=f"path-{i}", + path_type="common_prefix", + physical_address="?", + checksum="", + mtime=i, + )) + return lakefs_sdk.ObjectStatsList(pagination=lakefs_sdk.Pagination( + has_more=False, + next_offset="", + max_per_page=1, + results=1), + results=results) + + monkeypatch.setattr(ref._client.sdk_client.objects_api, "list_objects", monkey_list_objects) + + for i, item in enumerate(ref.objects()): + if i % 2: + assert isinstance(item, ObjectInfo) + assert item.size_bytes == i + else: + assert isinstance(item, CommonPrefix) + with expect_exception_context(AttributeError): + item.checksum # pylint: disable=pointless-statement + + assert item.path == f"path-{i}"