SDS Files API | <Client>.sds_files

spectrumx.api.sds_files

API functions specific to files.

Classes:

Name	Description
FileUploadMode	Modes for uploading files to SDS.

Functions:

Name	Description
delete_file	Deletes a file from SDS by its UUID.
download_file	Downloads a file from SDS: metadata and contents (unless skip_contents=True).
get_file	Get a file instance by its ID. Only metadata is downloaded from SDS.
list_files	Lists files in a given SDS path.
upload_file	Uploads a file to SDS.

Classes

FileUploadMode

Bases: Enum

              flowchart TD
              spectrumx.api.sds_files.FileUploadMode[FileUploadMode]

              

              click spectrumx.api.sds_files.FileUploadMode href "" "spectrumx.api.sds_files.FileUploadMode"
            

Modes for uploading files to SDS.

Functions

delete_file

delete_file(client: Client, file_uuid: UUID4 | str) -> bool

Deletes a file from SDS by its UUID.

Parameters:

Name	Type	Description	Default
client	Client	The client instance.	required
file_uuid	UUID4 | str	The UUID of the file to delete.	required

Returns: True if the file was deleted successfully, or if in dry run mode (simulating success). Raises: SDSError: If the file couldn't be deleted.

Source code in spectrumx/api/sds_files.py

def delete_file(client: Client, file_uuid: UUID4 | str) -> bool:
    """Deletes a file from SDS by its UUID.

    Args:
        client: The client instance.
        file_uuid: The UUID of the file to delete.
    Returns:
        True if the file was deleted successfully,
        or if in dry run mode (simulating success).
    Raises:
        SDSError: If the file couldn't be deleted.
    """
    uuid_to_delete: UUID4 = (
        uuid.UUID(file_uuid) if isinstance(file_uuid, str) else file_uuid
    )

    if client.dry_run:
        log_user(f"Dry run enabled: would delete file with UUID {uuid_to_delete.hex}")
        return True

    checksum: str | None = None
    if not client.dry_run:
        try:
            file_instance = get_file(client=client, file_uuid=uuid_to_delete)
            checksum = file_instance.sum_blake3
        except (ValueError, SDSError) as err:
            log_user_warning(
                f"Could not fetch file info before deletion: {err}. "
                "Persisted uploads may not be cleaned up."
            )

    if client._gateway.delete_file_by_id(uuid=uuid_to_delete.hex):
        if checksum:
            __cleanup_persisted_uploads_for_deleted_file(checksum=checksum)
        return True

    return False

download_file

download_file(
    *,
    client: Client,
    file_instance: File | None = None,
    file_uuid: UUID4 | str | None = None,
    to_local_path: Path | str | None = None,
    skip_contents: bool = False,
    warn_missing_path: bool = True,
) -> File

Downloads a file from SDS: metadata and contents (unless skip_contents=True).

Note this function always overwrites the local path of the file instance.

Source code in spectrumx/api/sds_files.py

def download_file(
    *,
    client: Client,
    file_instance: File | None = None,
    file_uuid: UUID4 | str | None = None,
    to_local_path: Path | str | None = None,
    skip_contents: bool = False,
    warn_missing_path: bool = True,
) -> File:
    """Downloads a file from SDS: metadata and contents (unless skip_contents=True).

    Note this function always overwrites the local path of the file instance.
    """
    # prepare file instance
    if isinstance(file_instance, File):
        file_instance, valid_uuid, valid_local_path_or_none = (
            __extract_download_info_from_file_instance(
                file_instance=file_instance,
                to_local_path=to_local_path,
                warn_missing_path=warn_missing_path,
            )
        )
    # or fetch file info from SDS creating a new instance
    else:
        file_instance, valid_uuid, valid_local_path_or_none = (
            __pre_fetch_file_for_download(
                client=client,
                file_uuid=file_uuid,
                to_local_path=to_local_path,
                warn_missing_path=warn_missing_path,
            )
        )

    __download_file_contents_if_applicable(
        client=client,
        file_instance=file_instance,
        valid_uuid=valid_uuid,
        valid_local_path_or_none=valid_local_path_or_none,
        skip_contents=skip_contents,
    )
    return file_instance

get_file

get_file(client: Client, file_uuid: UUID4 | str) -> File

Get a file instance by its ID. Only metadata is downloaded from SDS.

Note this does not download the file contents from the server. File instances still need to have their contents downloaded to create a local copy - see download_file().

Parameters:

Name	Type	Description	Default
file_uuid	UUID4 | str	The UUID of the file to retrieve.	required

Returns: The file instance, or a sample file if in dry run mode.

Source code in spectrumx/api/sds_files.py

def get_file(client: Client, file_uuid: UUID4 | str) -> File:
    """Get a file instance by its ID. Only metadata is downloaded from SDS.

    Note this does not download the file contents from the server. File
        instances still need to have their contents downloaded to create
        a local copy - see `download_file()`.

    Args:
        file_uuid: The UUID of the file to retrieve.
    Returns:
        The file instance, or a sample file if in dry run mode.
    """

    uuid_to_set: UUID4 = (
        uuid.UUID(file_uuid) if isinstance(file_uuid, str) else file_uuid
    )

    if not client.dry_run:
        file_bytes = client._gateway.get_file_by_id(uuid=uuid_to_set.hex)
        return File.model_validate_json(file_bytes)

    log_user("Dry run enabled: a sample file is being returned instead")
    return files.generate_sample_file(uuid_to_set)

list_files

list_files(
    *,
    client: Client,
    sds_path: PurePosixPath | Path | str,
    verbose: bool = False,
) -> Paginator[File]

Lists files in a given SDS path.

Parameters:

Name	Type	Description	Default
sds_path	PurePosixPath | Path | str	The virtual directory on SDS to list files from.	required

Returns: A paginator for the files in the given SDS path.

Source code in spectrumx/api/sds_files.py

def list_files(
    *,
    client: Client,
    sds_path: PurePosixPath | Path | str,
    verbose: bool = False,
) -> Paginator[File]:
    """Lists files in a given SDS path.

    Args:
        sds_path: The virtual directory on SDS to list files from.
    Returns:
        A paginator for the files in the given SDS path.
    """
    sds_path = PurePosixPath(sds_path)
    if client.dry_run:
        log_user("Dry run enabled: files will be simulated")
    pagination: Paginator[File] = Paginator(
        gateway=client._gateway,
        Entry=File,
        list_method=client._gateway.list_files,
        list_kwargs={"sds_path": sds_path},
        dry_run=client.dry_run,
        verbose=verbose,
    )

    return pagination

upload_file

upload_file(
    *,
    client: Client,
    local_file: File | Path | str,
    sds_path: PurePosixPath | Path | str = "/",
) -> File

Uploads a file to SDS.

If the file instance passed already has a directory set, sds_path will be prepended. E.g. given: sds_path = 'baz'; and local_file.directory = 'foo/bar/', then: The file will be uploaded to baz/foo/bar/ (under the user root in SDS) and the returned file instance will have its directory attribute updated to match this new path.

Parameters:

Name	Type	Description	Default
local_file	File | Path | str	The local file to upload.	required
sds_path	PurePosixPath | Path | str	The virtual directory on SDS to upload the file to.	'/'

Returns: The file instance with updated attributes, or a sample when in dry run.

Source code in spectrumx/api/sds_files.py

def upload_file(
    *,
    client: Client,
    local_file: File | Path | str,
    sds_path: PurePosixPath | Path | str = "/",
) -> File:
    """Uploads a file to SDS.

    If the file instance passed already has a directory set, `sds_path` will
    be prepended. E.g. given:
        `sds_path = 'baz'`; and
        `local_file.directory = 'foo/bar/'`, then:
    The file will be uploaded to `baz/foo/bar/` (under the user root in SDS) and
        the returned file instance will have its `directory` attribute updated
        to match this new path.

    Args:
        local_file:     The local file to upload.
        sds_path:       The virtual directory on SDS to upload the file to.
    Returns:
        The file instance with updated attributes, or a sample when in dry run.
    """
    # validate inputs
    if not isinstance(local_file, (File, Path, str)):
        msg = f"file_path must be a Path, str, or File instance, not {type(local_file)}"
        raise TypeError(msg)
    local_file = Path(local_file) if isinstance(local_file, str) else local_file
    sds_path = PurePosixPath(sds_path)

    # construct the file instance if needed
    if isinstance(local_file, File):
        file_instance = local_file.model_copy()
        if file_instance.directory:
            composed_sds_path = PurePosixPath(f"{sds_path}/{file_instance.directory}")
            file_instance.directory = composed_sds_path
    else:
        file_instance = files.construct_file(local_file, sds_path=sds_path)
    del local_file

    return __upload_file_mux(client=client, file_instance=file_instance)