diff --git a/Makefile b/Makefile index 4a1c0626ffd..7273f200952 100644 --- a/Makefile +++ b/Makefile @@ -54,7 +54,7 @@ GIT_REF=$(shell git rev-parse --short HEAD --) REVISION=$(GIT_REF)$(DIRTY) export REVISION -.PHONY: all clean esti lint test gen help +.PHONY: all clean esti lint *-lint test gen help all: build clean: @@ -160,9 +160,9 @@ build-binaries: $(GOBUILD) -o $(LAKEFS_BINARY_NAME) -ldflags $(LD_FLAGS) -v ./cmd/$(LAKEFS_BINARY_NAME) $(GOBUILD) -o $(LAKECTL_BINARY_NAME) -ldflags $(LD_FLAGS) -v ./cmd/$(LAKECTL_BINARY_NAME) -lint: ## Lint code - $(GOCMD) tool golangci-lint run ./... $(GOLANGCI_LINT_FLAGS) - cd $(UI_DIR) && npm run lint +lint: go-lint webui-lint ## Lint code + +lint-full: python-wrapper-lint lint ## Lint all code (slower) esti: ## run esti (system testing) $(GOTEST) -v ./esti --args --system-tests @@ -259,9 +259,15 @@ checks-validator: lint validate-proto validate-ui-format \ validate-api \ validate-wrapper validate-wrapgen-testcode +go-lint: + $(GOCMD) tool golangci-lint run ./... $(GOLANGCI_LINT_FLAGS) + python-wrapper-lint: $(DOCKER) run --user $(UID_GID) --rm -v $(shell pwd):/mnt -e HOME=/tmp/ -w /mnt/clients/python-wrapper $(PYTHON_IMAGE) /bin/bash -c "./lint.sh" +webui-lint: + cd $(UI_DIR) && npm run lint + python-wrapper-gen-docs: @version=$$(python -m pip install -r clients/python-wrapper/requirements.txt && python clients/python-wrapper/_version.py) sphinx-build -b html -W clients/python-wrapper/docs clients/python-wrapper/_site/ diff --git a/clients/python-wrapper/README.md b/clients/python-wrapper/README.md index 33cdd5018ba..29cb5fcbceb 100644 --- a/clients/python-wrapper/README.md +++ b/clients/python-wrapper/README.md @@ -36,12 +36,17 @@ from lakefs.client import Client # If environment variables or .lakectl.yaml file exist repo = lakefs.repository(repository_id="my-repo") +# (The positional argument is simpler!) +repo = lakefs.repository("my-repo") + # Or explicitly initialize and provide a Client object clt = Client(username="", password="", host="") repo = lakefs.Repository(repository_id="my-repo", client=clt) # From this point, proceed using the package according to documentation main_branch = repo.create(storage_namespace="").branch(branch_id="main") + +# Prefer to use "repo" rather than pass in the same repository_id each time. ... ``` diff --git a/clients/python-wrapper/lakefs/_decorators.py b/clients/python-wrapper/lakefs/_decorators.py new file mode 100644 index 00000000000..9ca77cb957e --- /dev/null +++ b/clients/python-wrapper/lakefs/_decorators.py @@ -0,0 +1,77 @@ +""" +Module for renaming kwargs + +Based on https://stackoverflow.com/a/74564394 but does not warn. +""" + +from collections.abc import Callable +from functools import wraps +from typing import Any, ParamSpec, TypeVar, cast + +P = ParamSpec("P") +R = TypeVar("R") + + +def _rename_kwargs(kwarg_map: dict[str, str], func: Callable[P, R]) -> Callable[..., R]: + @wraps(func) + def wrapped(*args: Any, **kwargs: Any) -> R: + new_kwargs = {} + for k, v in kwargs.items(): + new_kwargs[kwarg_map.get(k, k)] = v + return cast(Callable[..., R], func)(*args, **new_kwargs) + return wrapped + + +def renamed_parameters(**kwarg_map: str) -> Callable[[Callable[P, R]], Callable[P, R]]: + """ + Decorator for renaming args + + To rename old1 -> new1 and old2 -> new2, decorate the function definition: + + @renamed_parameters(old1="new1", old2="new2") + def func(new1, new2): + pass + + Old names keep working at runtime, but type checkers see only the new + signature, so they reject calls that use old names. + + If old callers must type-check, declare both signatures with @overload, + mark the old one @deprecated, and decorate the implementation with + renamed_parameters_overloaded instead: + + @overload + def func(new1, new2): ... + @overload + @deprecated("old1, old2 renamed to new1, new2") + def func(*, old1, old2): ... + + @renamed_parameters_overloaded(old1="new1", old2="new2") + def func(new1, new2): + pass + + Make old names keyword-only in the deprecated overload: positional calls + already match the new signature, and mixing a positional new1 with a + keyword old1 fails at runtime. Use typing_extensions.deprecated, which + supports all Python versions. + + Type checkers report calls that match the deprecated overload only when + asked to: mypy needs --enable-error-code=deprecated, pyright needs + reportDeprecated (enabled in strict mode). + """ + def decorator(func: Callable[P, R]) -> Callable[P, R]: + return cast(Callable[P, R], _rename_kwargs(kwarg_map, func)) + return decorator + + +def renamed_parameters_overloaded(**kwarg_map: str) -> Callable[[Callable[P, R]], Callable[..., R]]: + """ + renamed_parameters for implementations of @overload-ed functions. + + Identical at runtime. The decorated function accepts any arguments, so that + type checkers accept it as an implementation of the deprecated overloads that + use old names. Callers are still checked against the overloads. See + renamed_parameters for usage. + """ + def decorator(func: Callable[P, R]) -> Callable[..., R]: + return _rename_kwargs(kwarg_map, func) + return decorator diff --git a/clients/python-wrapper/lakefs/branch.py b/clients/python-wrapper/lakefs/branch.py index 5a377097e5c..b12ac2075c3 100644 --- a/clients/python-wrapper/lakefs/branch.py +++ b/clients/python-wrapper/lakefs/branch.py @@ -7,14 +7,21 @@ import uuid import warnings from contextlib import contextmanager -from typing import Optional, Generator, Iterable, Iterator, Literal, Dict +from typing import Optional, Generator, Iterable, Iterator, Literal, Dict, TYPE_CHECKING, overload + +from typing_extensions import deprecated + +from lakefs._decorators import renamed_parameters_overloaded import lakefs_sdk from lakefs.client import Client +from lakefs.getter import repository_for from lakefs.object import WriteableObject from lakefs.object import StoredObject from lakefs.import_manager import ImportManager from lakefs.reference import Reference, ReferenceType, generate_listing +if TYPE_CHECKING: + from lakefs.repository import Repository from lakefs.models import Change, Commit from lakefs.tag import Tag from lakefs.exceptions import ( @@ -43,7 +50,7 @@ def object(self, path: str) -> WriteableObject: :param path: The object's path """ - return WriteableObject(self.repo_id, self._id, path, client=self._client) + return WriteableObject(self._repo, self._id, path, client=self._client) def uncommitted(self, max_amount: Optional[int] = None, after: Optional[str] = None, prefix: Optional[str] = None, **kwargs) -> Generator[Change]: @@ -60,7 +67,7 @@ def uncommitted(self, max_amount: Optional[int] = None, after: Optional[str] = N """ for diff in generate_listing(self._client.sdk_client.branches_api.diff_branch, - self._repo_id, self._id, max_amount=max_amount, after=after, prefix=prefix, + self.repo_id, self._id, max_amount=max_amount, after=after, prefix=prefix, **kwargs): yield Change(**diff.dict()) @@ -107,7 +114,7 @@ def delete_objects(self, object_paths: str | StoredObject | Iterable[str | Store object_paths = [o.path if isinstance(o, StoredObject) else o for o in object_paths] with api_exception_handler(): return self._client.sdk_client.objects_api.delete_objects( - self._repo_id, + self.repo_id, self._id, lakefs_sdk.PathList(paths=object_paths) ) @@ -127,7 +134,7 @@ def reset_changes(self, path_type: Literal["common_prefix", "object", "reset"] = """ reset_creation = lakefs_sdk.ResetCreation(path=path, type=path_type) - return self._client.sdk_client.branches_api.reset_branch(self._repo_id, self.id, reset_creation, **kwargs) + return self._client.sdk_client.branches_api.reset_branch(self.repo_id, self.id, reset_creation, **kwargs) class Branch(_BaseBranch): @@ -135,8 +142,16 @@ class Branch(_BaseBranch): Class representing a branch in lakeFS. """ - def __init__(self, repository_id: str, branch_id: str, client: Optional[Client] = None): - super().__init__(repository_id, reference_id=branch_id, client=client) + @overload + def __init__(self, repo: Repository | str, branch_id: str, client: Optional[Client] = None) -> None: ... + + @overload + @deprecated("repository_id is renamed to repo") + def __init__(self, *, repository_id: str, branch_id: str, client: Optional[Client] = None) -> None: ... + + @renamed_parameters_overloaded(repository_id="repo") + def __init__(self, repo: Repository | str, branch_id: str, client: Optional[Client] = None): + super().__init__(repo, reference_id=branch_id, client=client) def get_commit(self): """ @@ -162,7 +177,7 @@ def cherry_pick(self, reference: ReferenceType, parent_number: Optional[int] = N cherry_pick_creation = lakefs_sdk.CherryPickCreation(ref=ref, parent_number=parent_number) with api_exception_handler(): res = self._client.sdk_client.branches_api.cherry_pick( - self._repo_id, self._id, cherry_pick_creation, **kwargs) + self.repo_id, self._id, cherry_pick_creation, **kwargs) return Commit(**res.dict()) def create(self, source_reference: ReferenceType, exist_ok: bool = False, **kwargs) -> Branch: @@ -195,7 +210,7 @@ def handle_conflict(e: LakeFSException): reference_id = source_reference if isinstance(source_reference, str) else source_reference.id branch_creation = lakefs_sdk.BranchCreation(name=self._id, source=reference_id, **kwargs) with api_exception_handler(handle_conflict): - self._client.sdk_client.branches_api.create_branch(self._repo_id, branch_creation) + self._client.sdk_client.branches_api.create_branch(self.repo_id, branch_creation) return self @property @@ -209,8 +224,8 @@ def head(self) -> Reference: :raise ServerException: for any other errors """ with api_exception_handler(): - branch = self._client.sdk_client.branches_api.get_branch(self._repo_id, self._id) - return Reference(self._repo_id, branch.commit_id, self._client) + branch = self._client.sdk_client.branches_api.get_branch(self.repo_id, self._id) + return Reference(self._repo, branch.commit_id, self._client) def commit(self, message: str, metadata: Optional[dict] = None, **kwargs) -> Reference: """ @@ -228,8 +243,8 @@ def commit(self, message: str, metadata: Optional[dict] = None, **kwargs) -> Ref commits_creation = lakefs_sdk.CommitCreation(message=message, metadata=metadata, **kwargs) with api_exception_handler(): - c = self._client.sdk_client.commits_api.commit(self._repo_id, self._id, commits_creation) - return Reference(self._repo_id, c.id, self._client) + c = self._client.sdk_client.commits_api.commit(self.repo_id, self._id, commits_creation) + return Reference(self._repo, c.id, self._client) def delete(self, **kwargs) -> None: """ @@ -242,7 +257,7 @@ def delete(self, **kwargs) -> None: :raise ServerException: for any other errors """ with api_exception_handler(): - return self._client.sdk_client.branches_api.delete_branch(self._repo_id, self._id, **kwargs) + return self._client.sdk_client.branches_api.delete_branch(self.repo_id, self._id, **kwargs) def revert(self, reference: Optional[ReferenceType], parent_number: int = 0, *, reference_id: Optional[str] = None, **kwargs) -> Commit: @@ -284,12 +299,12 @@ def revert(self, reference: Optional[ReferenceType], parent_number: int = 0, *, with api_exception_handler(): self._client.sdk_client.branches_api.revert_branch( - self._repo_id, + self.repo_id, self._id, lakefs_sdk.RevertCreation(ref=ref, parent_number=parent_number, allow_empty=False), **kwargs ) - commit = self._client.sdk_client.commits_api.get_commit(self._repo_id, self._id) + commit = self._client.sdk_client.commits_api.get_commit(self.repo_id, self._id) return Commit(**commit.dict()) def import_data(self, commit_message: str = "", metadata: Optional[dict] = None) -> ImportManager: @@ -301,7 +316,7 @@ def import_data(self, commit_message: str = "", metadata: Optional[dict] = None) message is provided, uses the default server commit message for imports. :return: an ImportManager object """ - return ImportManager(self._repo_id, self._id, commit_message, metadata, self._client) + return ImportManager(self.repo_id, self._id, commit_message, metadata, self._client) @contextmanager def transact(self, commit_message: str = "", commit_metadata: Optional[Dict] = None, @@ -342,7 +357,7 @@ def transact(self, commit_message: str = "", commit_metadata: Optional[Dict] = N :param tag: Optional tag name to create after the transaction is successfully completed :return: a Transaction object to perform the operations on """ - with Transaction(self._repo_id, self._id, commit_message, commit_metadata, delete_branch_on_error, + with Transaction(self._repo, self._id, commit_message, commit_metadata, delete_branch_on_error, self._client, tag) as tx: yield tx @@ -352,7 +367,19 @@ class _Transaction(_BaseBranch): def _get_tx_name() -> str: return f"tx-{uuid.uuid4()}" # Don't rely on source branch name as this might exceed valid branch length - def __init__(self, repository_id: str, branch_id: str, commit_message: str = "", + @overload + def __init__(self, repo: Repository | str, branch_id: str, commit_message: str = "", + commit_metadata: Optional[Dict] = None, client: Optional[Client] = None, + tag: Optional[str] = None) -> None: ... + + @overload + @deprecated("repository_id is renamed to repo") + def __init__(self, *, repository_id: str, branch_id: str, commit_message: str = "", + commit_metadata: Optional[Dict] = None, client: Optional[Client] = None, + tag: Optional[str] = None) -> None: ... + + @renamed_parameters_overloaded(repository_id="repo") + def __init__(self, repo: Repository | str, branch_id: str, commit_message: str = "", commit_metadata: Optional[Dict] = None, client: Optional[Client] = None, tag: Optional[str] = None): self._commit_message = commit_message self._commit_metadata = commit_metadata @@ -360,8 +387,8 @@ def __init__(self, repository_id: str, branch_id: str, commit_message: str = "", self._tag = tag tx_name = self._get_tx_name() - self._tx_branch = Branch(repository_id, tx_name, client).create(branch_id, hidden=True) - super().__init__(repository_id, tx_name, client) + self._tx_branch = Branch(repo, tx_name, client).create(branch_id, hidden=True) + super().__init__(repo, tx_name, client) @property def source_id(self) -> str: @@ -424,24 +451,44 @@ class Transaction: operations on the branch which will later be merged back into the source branch. Currently, transaction is supported only as a context manager. """ + _repo: Repository + _commit_message: str + _commit_metadata: Optional[Dict] + _source_branch: str + _tx: Optional[_Transaction] + _tx_branch: Optional[Branch] + _cleanup_branch: bool + _tag: Optional[str] + + @overload + def __init__(self, repo: Repository | str, branch_id: str, commit_message: str = "", + commit_metadata: Optional[Dict] = None, delete_branch_on_error: bool = True, + client: Optional[Client] = None, tag: Optional[str] = None) -> None: ... + + @overload + @deprecated("repository_id is renamed to repo") + def __init__(self, *, repository_id: str, branch_id: str, commit_message: str = "", + commit_metadata: Optional[Dict] = None, delete_branch_on_error: bool = True, + client: Optional[Client] = None, tag: Optional[str] = None) -> None: ... - def __init__(self, repository_id: str, branch_id: str, commit_message: str = "", + @renamed_parameters_overloaded(repository_id="repo") + def __init__(self, repo: Repository | str, branch_id: str, commit_message: str = "", commit_metadata: Optional[Dict] = None, delete_branch_on_error: bool = True, - client: Optional[Client] = None, tag: Optional[str] = None): - self._repo_id = repository_id + client: Optional[Client] = None, tag: Optional[str] = None) -> None: + self._repo = repository_for(repo, client) self._commit_message = commit_message self._commit_metadata = commit_metadata self._source_branch = branch_id self._client = client - self._tx: Optional["_Transaction"] = None - self._tx_branch: Optional["Branch"] = None + self._tx: Optional[_Transaction] = None + self._tx_branch: Optional[Branch] = None self._cleanup_branch = delete_branch_on_error self._tag = tag def __enter__(self): - self._tx = _Transaction(self._repo_id, self._source_branch, self._commit_message, self._commit_metadata, + self._tx = _Transaction(self._repo, self._source_branch, self._commit_message, self._commit_metadata, self._client, self._tag) - self._tx_branch = Branch(self._repo_id, self._tx.id, self._client) + self._tx_branch = Branch(self._repo, self._tx.id, self._client) return self._tx def __exit__(self, typ, value, traceback) -> Literal[False]: @@ -463,6 +510,6 @@ def __exit__(self, typ, value, traceback) -> Literal[False]: raise TransactionException(f"Failed committing transaction {self._tx.id}: {e}") from e if self._tx.tag: - tag = Tag(self._repo_id, self._tx.tag, self._client) + tag = Tag(self._repo, self._tx.tag, self._client) tag.create(merge_commit_id) return False diff --git a/clients/python-wrapper/lakefs/exceptions.py b/clients/python-wrapper/lakefs/exceptions.py index d94a395d697..29da19b6500 100644 --- a/clients/python-wrapper/lakefs/exceptions.py +++ b/clients/python-wrapper/lakefs/exceptions.py @@ -194,3 +194,11 @@ def handle_http_error(resp: HTTPResponse) -> None: exc_class = _STATUS_CODE_TO_EXCEPTION.get(resp.status, ServerException) lakefs_ex = exc_class(resp.status, resp.reason, resp.data, headers_dict) raise lakefs_ex + + +def io_exception_handler(e: LakeFSException): + if isinstance(e, NotFoundException): + return ObjectNotFoundException(e.status_code, e.reason) + if isinstance(e, (NotAuthorizedException, ForbiddenException)): + return PermissionException(e.status_code, e.reason) + return e diff --git a/clients/python-wrapper/lakefs/getter.py b/clients/python-wrapper/lakefs/getter.py new file mode 100644 index 00000000000..8adcc0650a6 --- /dev/null +++ b/clients/python-wrapper/lakefs/getter.py @@ -0,0 +1,27 @@ +""" +Module containing str-or-Object getters + +This can help break import cycles. To convert Repository | str to +Repository in another client module you can + +from typing import TYPE_CHECKING +if TYPE_CHECKING: + from repository import Repository +from getter import repository_for + +repo: Repository = repository_for('abc') +""" + +from __future__ import annotations +from typing import Optional, TYPE_CHECKING +from lakefs.client import Client +# Import all of repository to break circular import. +from lakefs import repository as repository_mod # Rename: repository is also a function. + + +def repository_for(repo: repository_mod.Repository | str, client: Optional[Client]) -> repository_mod.Repository: + return repository_mod.Repository(repo, client) if isinstance(repo, str) else repo + +def repository_id(repo: repository_mod.Repository | str) -> str: + """Return the repository_id of repo, or itself if it is already a repository_id""" + return repo if isinstance(repo, str) else repo.id diff --git a/clients/python-wrapper/lakefs/import_manager.py b/clients/python-wrapper/lakefs/import_manager.py index b18b174b317..295e60d1073 100644 --- a/clients/python-wrapper/lakefs/import_manager.py +++ b/clients/python-wrapper/lakefs/import_manager.py @@ -6,13 +6,21 @@ import asyncio from datetime import timedelta -from typing import Optional, Dict, List +from typing import Optional, Dict, List, TYPE_CHECKING, overload + +from typing_extensions import deprecated + +from lakefs._decorators import renamed_parameters_overloaded import lakefs_sdk from lakefs.models import ImportStatus, _OBJECT, _COMMON_PREFIX from lakefs.client import Client, _BaseLakeFSObject from lakefs.exceptions import ImportManagerException, api_exception_handler +from lakefs.getter import repository_id + +if TYPE_CHECKING: + from lakefs.repository import Repository class ImportManager(_BaseLakeFSObject): @@ -46,9 +54,20 @@ class ImportManager(_BaseLakeFSObject): commit_metadata: Optional[Dict] sources: List[lakefs_sdk.ImportLocation] - def __init__(self, repository_id: str, branch_id: str, commit_message: str = "", + @overload + def __init__(self, repo: Repository | str, branch_id: str, commit_message: str = "", + commit_metadata: Optional[Dict] = None, client: Optional[Client] = None) -> None: ... + + @overload + @deprecated("repository_id is renamed to repo") + def __init__(self, *, repository_id: str, branch_id: str, commit_message: str = "", + commit_metadata: Optional[Dict] = None, client: Optional[Client] = None) -> None: ... + + @renamed_parameters_overloaded(repository_id="repo") + def __init__(self, repo: Repository | str, branch_id: str, commit_message: str = "", commit_metadata: Optional[Dict] = None, client: Optional[Client] = None) -> None: - self._repo_id = repository_id + # Only need an ID so drop the Repository. + self._repo_id = repository_id(repo) self._branch_id = branch_id self.commit_message = commit_message self.commit_metadata = commit_metadata diff --git a/clients/python-wrapper/lakefs/namedtuple.py b/clients/python-wrapper/lakefs/namedtuple.py index b392d4a9c41..65b0cffa8e6 100644 --- a/clients/python-wrapper/lakefs/namedtuple.py +++ b/clients/python-wrapper/lakefs/namedtuple.py @@ -2,6 +2,8 @@ LenientNamedTuple Module """ +from inspect import get_annotations + class LenientNamedTuple: """ @@ -13,7 +15,7 @@ class LenientNamedTuple: unknown: dict = {} def __init__(self, **kwargs): - fields = list(self.__class__.__dict__["__annotations__"].keys()) + fields = list(get_annotations(self.__class__).keys()) for k, v in kwargs.items(): if k in fields: setattr(self, k, v) diff --git a/clients/python-wrapper/lakefs/object.py b/clients/python-wrapper/lakefs/object.py index 8edc450f7e5..523222c035f 100644 --- a/clients/python-wrapper/lakefs/object.py +++ b/clients/python-wrapper/lakefs/object.py @@ -13,25 +13,29 @@ import tempfile import urllib.parse from abc import abstractmethod -from typing import Any, AnyStr, IO, Iterator, List, Literal, Optional, Union, get_args +from typing import Any, AnyStr, IO, Iterator, List, Literal, Optional, Union, get_args, TYPE_CHECKING, overload + +from typing_extensions import deprecated + +from lakefs._decorators import renamed_parameters_overloaded import lakefs_sdk from lakefs_sdk import StagingMetadata -from lakefs.client import Client, _BaseLakeFSObject, SINGLE_STORAGE_ID +from lakefs.client import Client, _BaseLakeFSObject from lakefs.exceptions import ( api_exception_handler, handle_http_error, + io_exception_handler, LakeFSException, NotFoundException, - ObjectNotFoundException, - NotAuthorizedException, - ForbiddenException, - PermissionException, ObjectExistsException, InvalidRangeException, ) +from lakefs.getter import repository_for from lakefs.models import ObjectInfo +if TYPE_CHECKING: + from lakefs.repository import Repository _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 @@ -310,9 +314,9 @@ def _read(self, read_range: str) -> str | bytes: return b'' try: - with api_exception_handler(_io_exception_handler): + with api_exception_handler(io_exception_handler): response = self._client.sdk_client.objects_api.get_object_with_http_info( - self._obj.repo, + self._obj.repo_id, self._obj.ref, self._obj.path, range=read_range, @@ -570,7 +574,7 @@ def _upload_raw(self) -> lakefs_sdk.ObjectStats: headers[_LAKEFS_METADATA_PREFIX + k] = v self._fd.seek(0) - resource_path = urllib.parse.quote(f"/repositories/{self._obj.repo}/branches/{self._obj.ref}/objects", + resource_path = urllib.parse.quote(f"/repositories/{self._obj.repo_id}/branches/{self._obj.ref}/objects", encoding="utf-8") query_params = urllib.parse.urlencode({"path": self._obj.path}, encoding="utf-8") url = self._client.config.host + resource_path + f"?{query_params}" @@ -589,7 +593,7 @@ def _upload_raw(self) -> lakefs_sdk.ObjectStats: def _upload_presign(self) -> lakefs_sdk.ObjectStats: # Extract staging-specific kwargs staging_kwargs = {k: v for k, v in self._kwargs.items() if k in ['_request_timeout', 'async_req']} - staging_location = self._client.sdk_client.staging_api.get_physical_address(self._obj.repo, + staging_location = self._client.sdk_client.staging_api.get_physical_address(self._obj.repo_id, self._obj.ref, self._obj.path, True, @@ -619,7 +623,7 @@ def _upload_presign(self) -> lakefs_sdk.ObjectStats: mtime=None) if_none_match = "*" if self._mode.startswith("x") else None try: - return self._client.sdk_client.staging_api.link_physical_address(self._obj.repo, + return self._client.sdk_client.staging_api.link_physical_address(self._obj.repo_id, self._obj.ref, self._obj.path, staging_metadata=staging_metadata, @@ -668,14 +672,23 @@ class StoredObject(_BaseLakeFSObject): """ Class representing an object in lakeFS. """ - _repo_id: str + _repo: Repository _ref_id: str _path: str _stats: Optional[ObjectInfo] = None - _storage_id: Optional[str] = None - def __init__(self, repository_id: str, reference_id: str, path: str, client: Optional[Client] = None): - self._repo_id = repository_id + @overload + def __init__(self, repo: Repository | str, reference_id: str, path: str, + client: Optional[Client] = None) -> None: ... + + @overload + @deprecated("repository_id is renamed to repo") + def __init__(self, *, repository_id: str, reference_id: str, path: str, + client: Optional[Client] = None) -> None: ... + + @renamed_parameters_overloaded(repository_id="repo") + def __init__(self, repo: Repository | str, reference_id: str, path: str, client: Optional[Client] = None): + self._repo = repository_for(repo, client) self._ref_id = reference_id self._path = path super().__init__(client) @@ -684,14 +697,30 @@ def __str__(self) -> str: return self.path def __repr__(self): - return f'StoredObject(repository="{self.repo}", reference="{self.ref}", path="{self.path}")' + return f'StoredObject(repository="{self.repo_id}", reference="{self.ref}", path="{self.path}")' + + @property + def repository(self) -> Repository: + """ + Returns the object's repository + """ + return self._repo @property def repo(self) -> str: """ Returns the object's repository id + + Prefer repo_id. + """ + return self.repo_id + + @property + def repo_id(self) -> str: + """ + Returns the object's repository id """ - return self._repo_id + return self._repo.id @property def ref(self) -> str: @@ -739,9 +768,9 @@ def stat(self, pre_sign: Optional[bool] = None, **kwargs) -> ObjectInfo: """ # don't cache pre-signed stats if self._stats is None or pre_sign: - with api_exception_handler(_io_exception_handler): + with api_exception_handler(io_exception_handler): stat = self._client.sdk_client.objects_api.stat_object( - self._repo_id, self._ref_id, self._path, presign=pre_sign, **kwargs) + self.repo_id, self._ref_id, self._path, presign=pre_sign, **kwargs) self._stats = ObjectInfo(**stat.dict()) return self._stats @@ -749,11 +778,7 @@ def storage_id(self) -> str: """ Return the storage ID associated with this object """ - if self._storage_id is None: - with api_exception_handler(_io_exception_handler): - repo = self._client.sdk_client.repositories_api.get_repository(self._repo_id) - self._storage_id = repo.storage_id if repo.storage_id else SINGLE_STORAGE_ID - return self._storage_id + return self._repo.storage_id def exists(self, **kwargs) -> bool: """ @@ -765,10 +790,10 @@ def exists(self, **kwargs) -> bool: def exist_handler(e: LakeFSException): if isinstance(e, NotFoundException): return None # exists = False - return _io_exception_handler(e) + return io_exception_handler(e) with api_exception_handler(exist_handler): - self._client.sdk_client.objects_api.head_object(self._repo_id, self._ref_id, self._path, **kwargs) + self._client.sdk_client.objects_api.head_object(self.repo_id, self._ref_id, self._path, **kwargs) exists = True return exists @@ -788,13 +813,13 @@ def copy(self, destination_branch_id: str, destination_path: str, **kwargs) -> W with api_exception_handler(): object_copy_creation = lakefs_sdk.ObjectCopyCreation(src_ref=self._ref_id, src_path=self._path, shallow=False) - self._client.sdk_client.objects_api.copy_object(repository=self._repo_id, + self._client.sdk_client.objects_api.copy_object(repository=self.repo_id, branch=destination_branch_id, dest_path=destination_path, object_copy_creation=object_copy_creation, **kwargs) - return WriteableObject(repository_id=self._repo_id, reference_id=destination_branch_id, path=destination_path, + return WriteableObject(repo=self._repo, reference_id=destination_branch_id, path=destination_path, client=self._client) @@ -805,12 +830,22 @@ class WriteableObject(StoredObject): This Object is instantiated and returned upon invoking writer() on Branch reference type. """ - def __init__(self, repository_id: str, reference_id: str, path: str, + @overload + def __init__(self, repo: Repository | str, reference_id: str, path: str, + client: Optional[Client] = None) -> None: ... + + @overload + @deprecated("repository_id is renamed to repo") + def __init__(self, *, repository_id: str, reference_id: str, path: str, + client: Optional[Client] = None) -> None: ... + + @renamed_parameters_overloaded(repository_id="repo") + def __init__(self, repo: Repository | str, reference_id: str, path: str, client: Optional[Client] = None) -> None: - super().__init__(repository_id, reference_id, path, client=client) + super().__init__(repo, reference_id, path, client=client) def __repr__(self): - return f'WriteableObject(repository="{self.repo}", reference="{self.ref}", path="{self.path}")' + return f'WriteableObject(repository="{self.repo_id}", reference="{self.ref}", path="{self.path}")' def upload(self, data: str | bytes, @@ -855,8 +890,8 @@ def delete(self, **kwargs) -> None: :raise PermissionException: if user is not authorized to perform this operation, or operation is forbidden :raise ServerException: for any other errors """ - with api_exception_handler(_io_exception_handler): - self._client.sdk_client.objects_api.delete_object(self._repo_id, self._ref_id, self._path, **kwargs) + with api_exception_handler(io_exception_handler): + self._client.sdk_client.objects_api.delete_object(self.repo_id, self._ref_id, self._path, **kwargs) self._stats = None def writer(self, @@ -896,11 +931,3 @@ def writer(self, metadata=metadata, client=self._client, **kwargs) - - -def _io_exception_handler(e: LakeFSException): - if isinstance(e, NotFoundException): - return ObjectNotFoundException(e.status_code, e.reason) - if isinstance(e, (NotAuthorizedException, ForbiddenException)): - return PermissionException(e.status_code, e.reason) - return e diff --git a/clients/python-wrapper/lakefs/reference.py b/clients/python-wrapper/lakefs/reference.py index 3cfde6b06f1..c49d319a985 100644 --- a/clients/python-wrapper/lakefs/reference.py +++ b/clients/python-wrapper/lakefs/reference.py @@ -4,13 +4,20 @@ from __future__ import annotations -from typing import Optional, Generator, Union +from typing import Optional, Generator, Union, TYPE_CHECKING, overload + +from typing_extensions import deprecated + +from lakefs._decorators import renamed_parameters_overloaded import lakefs_sdk from lakefs.models import Commit, Change, CommonPrefix, ObjectInfo, _OBJECT from lakefs.client import Client, _BaseLakeFSObject from lakefs.exceptions import api_exception_handler +from lakefs.getter import repository_for +if TYPE_CHECKING: + from lakefs.repository import Repository from lakefs.object import StoredObject @@ -18,14 +25,22 @@ class Reference(_BaseLakeFSObject): """ Class representing a reference in lakeFS. """ - _repo_id: str + _repo: Repository _id: str _commit: Optional[Commit] = None - def __init__(self, repository_id: str, reference_id: str, client: Optional[Client] = None) -> None: + @overload + def __init__(self, repo: Repository | str, reference_id: str, client: Optional[Client] = None) -> None: ... + + @overload + @deprecated("repository_id is renamed to repo") + def __init__(self, *, repository_id: str, reference_id: str, client: Optional[Client] = None) -> None: ... + + @renamed_parameters_overloaded(repository_id="repo") + def __init__(self, repo: Repository | str, reference_id: str, client: Optional[Client] = None) -> None: """Return a reference to a lakeFS commit. - :param repository_id: the repository holding the commit + :param repo: the repository holding the commit, or the id of that repository :param reference_id: a reference expression to the commit Any reference expression can be used as a reference_id, for example: @@ -37,7 +52,7 @@ def __init__(self, repository_id: str, reference_id: str, client: Optional[Clien See https://docs.lakefs.io/understand/model.html#ref-expressions for details. """ - self._repo_id = repository_id + self._repo = repository_for(repo, client) self._id = reference_id super().__init__(client) @@ -46,7 +61,7 @@ def repo_id(self) -> str: """ Return the repository id for this reference """ - return self._repo_id + return self._repo.id @property def id(self) -> str: @@ -76,7 +91,7 @@ def objects(self, """ for res in generate_listing(self._client.sdk_client.objects_api.list_objects, - repository=self._repo_id, + repository=self.repo_id, ref=self._id, max_amount=max_amount, after=after, @@ -96,7 +111,7 @@ def log(self, max_amount: Optional[int] = None, **kwargs) -> Generator[Commit]: :raise NotAuthorizedException: if user is not authorized to perform this operation :raise ServerException: for any other errors """ - for res in generate_listing(self._client.sdk_client.refs_api.log_commits, self._repo_id, self._id, + for res in generate_listing(self._client.sdk_client.refs_api.log_commits, self.repo_id, self._id, max_amount=max_amount, **kwargs): yield Commit(**res.dict()) @@ -110,7 +125,7 @@ def get_commit(self) -> Commit: """ if self._commit is None: with api_exception_handler(): - commit = self._client.sdk_client.commits_api.get_commit(self._repo_id, self._id) + commit = self._client.sdk_client.commits_api.get_commit(self.repo_id, self._id) self._commit = Commit(**commit.dict()) return self._commit @@ -136,7 +151,7 @@ def diff(self, """ other_ref_id = other_ref if isinstance(other_ref, str) else other_ref.id for diff in generate_listing(self._client.sdk_client.refs_api.diff_refs, - repository=self._repo_id, + repository=self.repo_id, left_ref=self._id, right_ref=other_ref_id, after=after, @@ -160,7 +175,7 @@ def merge_into(self, destination_branch: ReferenceType, **kwargs) -> str: branch_id = destination_branch if isinstance(destination_branch, str) else destination_branch.id with api_exception_handler(): merge = lakefs_sdk.Merge(**kwargs) - res = self._client.sdk_client.refs_api.merge_into_branch(self._repo_id, + res = self._client.sdk_client.refs_api.merge_into_branch(self.repo_id, self._id, branch_id, merge=merge) @@ -172,7 +187,7 @@ def object(self, path: str) -> StoredObject: # pylint: disable=C0103 :param path: The object's path """ - return StoredObject(self._repo_id, self._id, path, self._client) + return StoredObject(self._repo, self._id, path, self._client) def __repr__(self): class_name = self.__class__.__name__ diff --git a/clients/python-wrapper/lakefs/repository.py b/clients/python-wrapper/lakefs/repository.py index b566729c179..050d63bfec5 100644 --- a/clients/python-wrapper/lakefs/repository.py +++ b/clients/python-wrapper/lakefs/repository.py @@ -11,8 +11,13 @@ from lakefs.models import RepositoryProperties from lakefs.tag import Tag from lakefs.branch import Branch -from lakefs.client import Client, _BaseLakeFSObject -from lakefs.exceptions import api_exception_handler, ConflictException, LakeFSException +from lakefs.client import Client, _BaseLakeFSObject, SINGLE_STORAGE_ID +from lakefs.exceptions import ( + api_exception_handler, + io_exception_handler, + ConflictException, + LakeFSException, +) from lakefs.reference import Reference, generate_listing @@ -24,6 +29,7 @@ class Repository(_BaseLakeFSObject): """ _id: str _properties: Optional[RepositoryProperties] = None + _storage_id: Optional[str] = None def __init__(self, repository_id: str, client: Optional[Client] = None) -> None: self._id = repository_id @@ -68,6 +74,16 @@ def handle_conflict(e: LakeFSException): self._properties = RepositoryProperties(**repo.dict()) return self + + @property + def storage_id(self) -> str: + if self._storage_id is None: + with api_exception_handler(io_exception_handler): + repo_data = self._client.sdk_client.repositories_api.get_repository(self.id) + self._storage_id = repo_data.storage_id if repo_data.storage_id else SINGLE_STORAGE_ID + return self._storage_id + + def delete(self, **kwargs) -> None: """ Delete repository from lakeFS server @@ -85,28 +101,28 @@ def branch(self, branch_id: str) -> Branch: Return a branch object using the current repository id and client :param branch_id: name of the branch """ - return Branch(self._id, branch_id, self._client) + return Branch(self, branch_id, self._client) def commit(self, commit_id: str) -> Reference: """ Return a reference object using the current repository id and client :param commit_id: id of the commit reference """ - return Reference(self._id, commit_id, self._client) + return Reference(self, commit_id, self._client) def ref(self, ref_id: str) -> Reference: """ Return a reference object using the current repository id and client :param ref_id: branch name, commit id or tag id """ - return Reference(self._id, ref_id, self._client) + return Reference(self, ref_id, self._client) def tag(self, tag_id: str) -> Tag: """ Return a tag object using the current repository id and client :param tag_id: name of the tag """ - return Tag(self._id, tag_id, self._client) + return Tag(self, tag_id, self._client) def branches(self, max_amount: Optional[int] = None, after: Optional[str] = None, prefix: Optional[str] = None, **kwargs) -> Generator[Branch]: @@ -124,7 +140,7 @@ def branches(self, max_amount: Optional[int] = None, for res in generate_listing(self._client.sdk_client.branches_api.list_branches, self._id, max_amount=max_amount, after=after, prefix=prefix, **kwargs): - yield Branch(self._id, res.id, client=self._client) + yield Branch(self, res.id, client=self._client) def tags(self, max_amount: Optional[int] = None, after: Optional[str] = None, prefix: Optional[str] = None, **kwargs) -> Generator[Tag]: @@ -141,7 +157,7 @@ def tags(self, max_amount: Optional[int] = None, """ for res in generate_listing(self._client.sdk_client.tags_api.list_tags, self._id, max_amount=max_amount, after=after, prefix=prefix, **kwargs): - yield Tag(self._id, res.id, client=self._client) + yield Tag(self, res.id, client=self._client) @property def metadata(self) -> dict[str, str]: diff --git a/clients/python-wrapper/lakefs/tag.py b/clients/python-wrapper/lakefs/tag.py index c61a69e55cc..f91e4d14c99 100644 --- a/clients/python-wrapper/lakefs/tag.py +++ b/clients/python-wrapper/lakefs/tag.py @@ -3,13 +3,19 @@ """ from __future__ import annotations -from typing import Optional +from typing import Optional, TYPE_CHECKING, overload + +from typing_extensions import deprecated + +from lakefs._decorators import renamed_parameters_overloaded import lakefs_sdk from lakefs.client import Client from lakefs.exceptions import api_exception_handler, LakeFSException, ConflictException from lakefs.reference import Reference, ReferenceType +if TYPE_CHECKING: + from lakefs.repository import Repository class Tag(Reference): @@ -17,8 +23,16 @@ class Tag(Reference): Class representing a tag in lakeFS. """ - def __init__(self, repository_id: str, tag_id: str, client: Optional[Client] = None): - super().__init__(repository_id, reference_id=tag_id, client=client) + @overload + def __init__(self, repo: Repository | str, tag_id: str, client: Optional[Client] = None) -> None: ... + + @overload + @deprecated("repository_id is renamed to repo") + def __init__(self, *, repository_id: str, tag_id: str, client: Optional[Client] = None) -> None: ... + + @renamed_parameters_overloaded(repository_id="repo") + def __init__(self, repo: Repository | str, tag_id: str, client: Optional[Client] = None): + super().__init__(repo, reference_id=tag_id, client=client) def create(self, source_ref: ReferenceType, exist_ok: Optional[bool] = False, **kwargs) -> Tag: """ @@ -41,7 +55,7 @@ def handle_conflict(e: LakeFSException): return None with api_exception_handler(handle_conflict): - self._client.sdk_client.tags_api.create_tag(self._repo_id, tag_creation, **kwargs) + self._client.sdk_client.tags_api.create_tag(self.repo_id, tag_creation, **kwargs) return self @@ -55,5 +69,5 @@ def delete(self, **kwargs) -> None: :raise ServerException: for any other errors """ with api_exception_handler(): - self._client.sdk_client.tags_api.delete_tag(self._repo_id, self.id, **kwargs) + self._client.sdk_client.tags_api.delete_tag(self.repo_id, self.id, **kwargs) self._commit = None diff --git a/clients/python-wrapper/pyproject.toml b/clients/python-wrapper/pyproject.toml index 878038929cd..f0873e8f1c2 100644 --- a/clients/python-wrapper/pyproject.toml +++ b/clients/python-wrapper/pyproject.toml @@ -16,6 +16,7 @@ authors = [ dependencies = [ "lakefs-sdk>=1.50,< 2", "pyyaml~=6.0.1", + "typing_extensions>=4.8.0,<5", ] classifiers = [ "Intended Audience :: Developers", diff --git a/clients/python-wrapper/tests/integration/test_branch.py b/clients/python-wrapper/tests/integration/test_branch.py index f92da5c0165..7fb2c0d5309 100644 --- a/clients/python-wrapper/tests/integration/test_branch.py +++ b/clients/python-wrapper/tests/integration/test_branch.py @@ -238,8 +238,6 @@ def test_transaction_with_existing_tag(setup_repo): path_and_data = ["a", "b", "bar/a", "bar/b", "bar/c", "c"] test_branch = repo.branch("main") - repo: "lakefs.Repository" - test_branch.object("initial_file").upload("initial content") initial_commit = test_branch.commit("initial commit") repo.tag("v1.0.0").create(initial_commit) diff --git a/clients/python-wrapper/tests/integration/test_object.py b/clients/python-wrapper/tests/integration/test_object.py index 5f1e0fd03fb..7f986dd1446 100644 --- a/clients/python-wrapper/tests/integration/test_object.py +++ b/clients/python-wrapper/tests/integration/test_object.py @@ -27,7 +27,7 @@ def test_object_read_seek(setup_repo, pre_sign): clt, repo = setup_repo data = b"test_data" - obj = WriteableObject(repository_id=repo.properties.id, reference_id="main", path="test_obj", client=clt).upload( + obj = WriteableObject(repo=repo, reference_id="main", path="test_obj", client=clt).upload( data=data, pre_sign=pre_sign) with obj.reader() as fd: @@ -87,7 +87,7 @@ def test_object_read_seek(setup_repo, pre_sign): def test_object_upload_exists(setup_repo, pre_sign): clt, repo = setup_repo data = b"test_data" - obj = WriteableObject(repository_id=repo.properties.id, reference_id="main", path="test_obj", client=clt).upload( + obj = WriteableObject(repo=repo, reference_id="main", path="test_obj", client=clt).upload( data=data) with expect_exception_context(ObjectExistsException): obj.upload(data="some_other_data", mode='xb', pre_sign=pre_sign) @@ -113,7 +113,7 @@ def test_object_upload_read_different_params(setup_repo, w_mode, r_mode, pre_sig # urllib3 encodes TextIOBase to ISO-8859-1, data should contain symbols that can be encoded as such data = b'test \x48\x65\x6c\x6c\x6f\x20\x57\x6f\x72\x6c\x64\x21' - obj = WriteableObject(repository_id=repo.properties.id, reference_id="main", path="test_obj", client=clt).upload( + obj = WriteableObject(repo=repo, reference_id="main", path="test_obj", client=clt).upload( data=data, mode=w_mode, pre_sign=pre_sign) with obj.reader(mode=r_mode) as fd: @@ -127,7 +127,7 @@ def test_object_upload_read_different_params(setup_repo, w_mode, r_mode, pre_sig def test_object_copy(setup_repo): clt, repo = setup_repo data = "test_data" - obj = WriteableObject(repository_id=repo.properties.id, reference_id="main", path="test_obj", client=clt).upload( + obj = WriteableObject(repo=repo, reference_id="main", path="test_obj", client=clt).upload( data=data, metadata={"foo": "bar"}) copy_name = "copy_obj" @@ -233,7 +233,7 @@ def test_read_byte_by_byte(setup_repo): clt, repo = setup_repo data = b'test_data' - obj = WriteableObject(repository_id=repo.properties.id, reference_id="main", path="test_obj", client=clt).upload( + obj = WriteableObject(repo=repo, reference_id="main", path="test_obj", client=clt).upload( data=data, pre_sign=False) res = b"" reader = obj.reader() @@ -353,7 +353,7 @@ def test_read_xml(setup_repo, datafiles): def test_readline_no_newline(setup_repo): clt, repo = setup_repo data = b'test_data' - obj = WriteableObject(repository_id=repo.properties.id, reference_id="main", path="test_obj", client=clt).upload( + obj = WriteableObject(repo=repo, reference_id="main", path="test_obj", client=clt).upload( data=data, pre_sign=False) assert obj.reader().readline() == data @@ -364,7 +364,7 @@ def test_readline_no_newline(setup_repo): def test_readline_partial_line_buffer(setup_repo): clt, repo = setup_repo data = "a" * 15 + "\n" + "b" * 25 + "\n" + "That is all folks! " - obj = WriteableObject(repository_id=repo.properties.id, reference_id="main", path="test_obj", client=clt).upload( + obj = WriteableObject(repo=repo, reference_id="main", path="test_obj", client=clt).upload( data=data, pre_sign=False) with obj.reader(mode="r") as reader: @@ -438,7 +438,7 @@ def test_reader_with_failure(setup_repo): def test_writer_with_failure(setup_repo): clt, repo = setup_repo data = b"test_data" - obj = WriteableObject(repository_id=repo.properties.id, + obj = WriteableObject(repo=repo, reference_id="main", path="test_obj", client=clt).upload(data=data) @@ -487,7 +487,7 @@ def test_writer_gif(setup_repo, datafiles): def test_large_file_object_write(setup_repo, w_mode): clt, repo = setup_repo stored_obj = StoredObject( - repository_id=repo.properties.id, + repo=repo, reference_id="main", path="test_binary_mode", client=clt, diff --git a/clients/python-wrapper/tests/integration/test_sanity.py b/clients/python-wrapper/tests/integration/test_sanity.py index feaefeafa67..f728141d774 100644 --- a/clients/python-wrapper/tests/integration/test_sanity.py +++ b/clients/python-wrapper/tests/integration/test_sanity.py @@ -111,7 +111,7 @@ def test_object_sanity(setup_repo): data = b"test_data" path = "test_obj" metadata = {"foo": "bar"} - obj = lakefs.WriteableObject(repository_id=repo.properties.id, reference_id="main", path=path, client=clt).upload( + obj = lakefs.WriteableObject(repo=repo, reference_id="main", path=path, client=clt).upload( data=data, metadata=metadata) with obj.reader() as fd: assert fd.read() == data diff --git a/clients/python-wrapper/tests/utests/test_object.py b/clients/python-wrapper/tests/utests/test_object.py index 1608f1ae77f..e973266e847 100644 --- a/clients/python-wrapper/tests/utests/test_object.py +++ b/clients/python-wrapper/tests/utests/test_object.py @@ -53,7 +53,7 @@ def readable_object_context(monkey, **kwargs): clt = get_test_client() monkey.setattr(clt, "storage_config_by_id", lambda *args: StorageTestConfig()) read_obj = StoredObject(client=clt, **kwargs) - monkey.setattr(read_obj, "_storage_id", SINGLE_STORAGE_ID) + monkey.setattr(read_obj, "storage_id", lambda *args: SINGLE_STORAGE_ID) yield read_obj @@ -65,7 +65,7 @@ def writeable_object_context(monkey, **kwargs): clt = get_test_client() monkey.setattr(clt, "storage_config_by_id", lambda *args: StorageTestConfig()) obj = WriteableObject(client=clt, **kwargs) - monkey.setattr(obj, "_storage_id", SINGLE_STORAGE_ID) + monkey.setattr(obj, "storage_id", lambda *args: SINGLE_STORAGE_ID) yield obj diff --git a/clients/python-wrapper/tests/utests/test_reference.py b/clients/python-wrapper/tests/utests/test_reference.py index de3980554b6..6603a6fb05f 100644 --- a/clients/python-wrapper/tests/utests/test_reference.py +++ b/clients/python-wrapper/tests/utests/test_reference.py @@ -13,7 +13,7 @@ def get_test_ref(): def test_reference_creation(): ref = get_test_ref() - assert ref._repo_id == "test_repo" + assert ref.repo_id == "test_repo" assert ref.id == "test_reference"