Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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/
Expand Down
5 changes: 5 additions & 0 deletions clients/python-wrapper/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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="<lakefs_access_key_id>", password="<lakefs_secret_access_key>", host="<lakefs_endpoint>")
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="<storage_namespace>").branch(branch_id="main")

# Prefer to use "repo" rather than pass in the same repository_id each time.
...
```

Expand Down
77 changes: 77 additions & 0 deletions clients/python-wrapper/lakefs/_decorators.py
Original file line number Diff line number Diff line change
@@ -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
105 changes: 76 additions & 29 deletions clients/python-wrapper/lakefs/branch.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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]:
Expand All @@ -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())

Expand Down Expand Up @@ -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)
)
Expand All @@ -127,16 +134,24 @@ 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):
"""
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):
"""
Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand All @@ -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:
"""
Expand All @@ -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:
"""
Expand All @@ -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:
Expand Down Expand Up @@ -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:
Expand All @@ -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)
Comment thread
skshetry marked this conversation as resolved.
Resolved

@contextmanager
def transact(self, commit_message: str = "", commit_metadata: Optional[Dict] = None,
Expand Down Expand Up @@ -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

Expand All @@ -352,16 +367,28 @@ 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
self._source_branch = branch_id
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:
Expand Down Expand Up @@ -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]:
Expand All @@ -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
8 changes: 8 additions & 0 deletions clients/python-wrapper/lakefs/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Loading
Loading