paths

RemotePath

RemotePath is used to model a remote path, it takes inspiration from pathlib and shares some of its interface.

date = Field(None, title='Date') class-attribute instance-attribute

group = Field(None, title='Group') class-attribute instance-attribute

hardlinks = Field(0, title='Hardlinks') class-attribute instance-attribute

name = Field(None, title='Name') class-attribute instance-attribute

perms = Field(None, title='Perms') class-attribute instance-attribute

size = Field(0, title='Size') class-attribute instance-attribute

user = Field(None, title='User') class-attribute instance-attribute

RemotePath is used to model a remote path, it takes inspiration from pathlib and shares some of its interface.

Source code in sfapi_client/_sync/paths.py

def __init__(self, path=None, **kwargs):
    super().__init__(**kwargs)
    self._path = PurePosixPath(path)

    if self.name is None:
        self.name = self._path.name

compute instance-attribute

date = Field(None, title='Date') class-attribute instance-attribute

group = Field(None, title='Group') class-attribute instance-attribute

hardlinks = Field(0, title='Hardlinks') class-attribute instance-attribute

name = self._path.name instance-attribute

parent property

The parent of the path.

Returns:

Type	Description
RemotePath	the parent

parents property

The parents of the path.

Returns:

Type	Description
List[RemotePath]	the parents

parts property

The paths components as a tuple.

Returns:

Type	Description
Tuple[str]	the path components

perms = Field(None, title='Perms') class-attribute instance-attribute

size = Field(0, title='Size') class-attribute instance-attribute

stem property

The final path component, without its suffix.

Returns:

Type	Description
str	the path stem

suffix property

The path extension.

Returns:

Type	Description
str	the path extension

suffixes property

A list of the path extensions.

Returns:

Type	Description
List[str]	the path extensions

user = Field(None, title='User') class-attribute instance-attribute

dict(*args, **kwargs)

Source code in sfapi_client/_sync/paths.py

def dict(self, *args, **kwargs) -> Dict:
    if "exclude" not in kwargs:
        kwargs["exclude"] = {"compute"}
    return super().dict(*args, **kwargs)

download(binary=False)

Download the file contents.

Parameters:

Name	Type	Description	Default
binary		indicate if the file should be treated as binary, defaults to False	False

Raises:

Type	Description
IsADirectoryError	if path points to a directory.
SfApiError

Source code in sfapi_client/_sync/paths.py

def download(self, binary=False) -> IO[AnyStr]:
    """
    Download the file contents.

    :param binary: indicate if the file should be treated as binary, defaults
    to False
    :raises IsADirectoryError: if path points to a directory.
    :raises SfApiError:
    """
    if self.is_dir():
        raise IsADirectoryError(self._path)

    r = self.compute.client.get(
        f"utilities/download/{self.compute.name}/{self._path}?binary={binary}"
    )
    json_response = r.json()
    download_response = FileDownloadResponse.model_validate(json_response)

    if download_response.status == FileDownloadResponseStatus.ERROR:
        raise SfApiError(download_response.error)

    file_data = download_response.file
    if download_response.is_binary:
        binary_file_data = b64decode(file_data)
        return BytesIO(binary_file_data)
    else:
        return StringIO(file_data)

is_dir()

Returns:

Type	Description
bool	Returns True if path is a directory, False otherwise.

Source code in sfapi_client/_sync/paths.py

def is_dir(self) -> bool:
    """
    :return: Returns True if path is a directory, False otherwise.
    """
    if self.perms is None:
        self.update()

    return self.perms[0] == "d"

is_file()

Returns:

Type	Description
bool	Returns True if path is a file, False otherwise.

Source code in sfapi_client/_sync/paths.py

def is_file(self) -> bool:
    """
    :return: Returns True if path is a file, False otherwise.
    """
    return not self.is_dir()

ls()

List the current path

Returns:

Type	Description
List[RemotePath]	the list of child paths

Source code in sfapi_client/_sync/paths.py

def ls(self) -> List["RemotePath"]:
    """
    List the current path

    :return: the list of child paths
    """
    return self._ls(self.compute, str(self._path))

open(mode)

Open the file at this path.

Parameters:

Name	Type	Description	Default
mode	str	The mode to open the file. Valid options are 'rwb'. raises: IsDirectoryError: If the path is not a file.	required

Source code in sfapi_client/_sync/paths.py

@contextmanager
def open(self, mode: str) -> IO[AnyStr]:
    """
    Open the file at this path.

    :param mode: The mode to open the file. Valid options are 'rwb'.

    raises: IsDirectoryError: If the path is not a file.
    """
    try:
        if self.is_dir():
            raise IsADirectoryError()
    except SfApiError as ex:
        # Its a valid use case to add a open a new file to an exiting directory.
        # In this case the is_dir() will raise a SfApiError with
        # "No such file or directory" So we check for that and then see if the
        # parent directory exists, if it does we can just continue.
        if not _is_no_such(ex):
            raise

        # Check if the parent is a directory ( as in we are creating a new file ),
        # if not re raise the original exception
        if not self.parent.is_dir():
            raise

    valid_modes_chars = set("rwb")
    mode_chars = set(mode)

    # If we have duplicate raise exception
    if len(mode_chars) != len(mode):
        raise ValueError(f"invalid mode: '{mode}'")

    # check mode chars
    if not mode_chars.issubset(valid_modes_chars):
        raise ValueError(f"invalid mode: '{mode}'")

    # we don't support read/write
    if "r" in mode_chars and "w" in mode_chars:
        raise ValueError(f"invalid mode: '{mode}', 'rw' not supported.")

    if "r" in mode_chars:
        binary = "b" in mode_chars
        yield self.download(binary)
    else:
        tmp = None
        try:
            tmp = tempfile.NamedTemporaryFile(mode, delete=False)
            yield tmp
            tmp.close()
            # Now upload the changes, we have to reopen the file to
            # ensure binary mode
            with open(tmp.name, "rb") as fp:
                self.upload(fp)
        finally:
            if tmp is not None:
                tmp.close()
                Path(tmp.name).unlink()

update()

Update the path in the latest information from the resource.

Source code in sfapi_client/_sync/paths.py

def update(self):
    """
    Update the path in the latest information from the resource.
    """
    # Here we pass filter_dots=False so that we with get . if this is a
    # directory
    file_state = self._ls(self.compute, str(self._path), filter_dots=False)
    if len(file_state) == 0:
        raise FileNotFoundError(self._path)

    # We update the name as it could be . from a directory listing and in that
    # case we don't want to update the name
    new_state = file_state[0]
    new_state.name = self.name

    self._update(new_state)

upload(file)

Source code in sfapi_client/_sync/paths.py

def upload(self, file: BytesIO) -> "RemotePath":
    try:
        if self.is_dir():
            upload_path = f"{str(self._path)}/{file.filename}"
        else:
            upload_path = str(self._path)
    except SfApiError as ex:
        # Its a valid use case to add a upload a new file to an exiting directory.
        # In this case the is_dir() will raise a SfApiError with
        # "No such file or directory" So we check for that and then see if the
        # parent directory exists, if it does we can just continue.
        if not _is_no_such(ex):
            raise

        # Check if the parent is a directory ( as in we are creating a new file ),
        # if not re raise the original exception
        if not self.parent.is_dir():
            raise
        else:
            upload_path = str(self._path)

    url = f"utilities/upload/{self.compute.name}/{upload_path}"
    files = {"file": file}

    r = self.compute.client.put(url, files=files)

    json_response = r.json()
    upload_response = UploadResponse.model_validate(json_response)
    if upload_response.status == UploadResponseStatus.ERROR:
        raise SfApiError(upload_response.error)

    remote_path = RemotePath(path=upload_path, compute=self.compute)

    return remote_path