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.
detach_file_from_datasets	Detach a file from every dataset it is linked to (owner-only).
download_file	Downloads a file from SDS: metadata and contents (unless skip_contents=True).
file_list_time_query_param	Format a datetime for Gateway file list temporal query params (ISO 8601, UTC).
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

detach_file_from_datasets

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

Detach a file from every dataset it is linked to (owner-only).

Clears the file's deprecated dataset FK and datasets M2M on the gateway.

Source code in spectrumx/api/sds_files.py

def detach_file_from_datasets(
    client: Client,
    file_uuid: UUID4 | str,
) -> bool:
    """Detach a file from every dataset it is linked to (owner-only).

    Clears the file's deprecated ``dataset`` FK and ``datasets`` M2M on the gateway.
    """
    uuid_to_detach: UUID4 = (
        uuid.UUID(file_uuid) if isinstance(file_uuid, str) else file_uuid
    )

    if client.dry_run:
        log_user(
            f"Dry run enabled: would detach file {uuid_to_detach.hex} from datasets"
        )
        return True

    client._gateway.detach_file_from_datasets(file_uuid=uuid_to_detach)
    return True

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,
    progress_callback: Callable[[int], None] | None = None,
) -> 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,
    progress_callback: collections.abc.Callable[[int], None] | None = None,
) -> 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,
        progress_callback=progress_callback,
    )
    return file_instance

file_list_time_query_param

file_list_time_query_param(value: datetime) -> str

Format a datetime for Gateway file list temporal query params (ISO 8601, UTC).

Source code in spectrumx/api/sds_files.py

def file_list_time_query_param(value: datetime) -> str:
    """Format a datetime for Gateway file list temporal query params (ISO 8601, UTC)."""
    value = value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
    return value.isoformat()

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,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    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
start_time	datetime | None	When set, lower bound for Digital RF data files (UTC-aligned ISO sent to the API). Must be used together with end_time.	None
end_time	datetime | None	Upper bound for Digital RF data files, paired with start_time.	None

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,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    verbose: bool = False,
) -> Paginator[File]:
    """Lists files in a given SDS path.

    Args:
        sds_path: The virtual directory on SDS to list files from.
        start_time: When set, lower bound for Digital RF data files (UTC-aligned ISO
            sent to the API). Must be used together with ``end_time``.
        end_time: Upper bound for Digital RF data files, paired with ``start_time``.
    Returns:
        A paginator for the files in the given SDS path.
    """
    if (start_time is None) ^ (end_time is None):
        msg = "start_time and end_time must both be set or both omitted."
        raise ValueError(msg)
    sds_path = PurePosixPath(sds_path)
    start_q: str | None = file_list_time_query_param(start_time) if start_time else None
    end_q: str | None = (
        file_list_time_query_param(end_time) if end_time is not None else None
    )
    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,
            "start_time": start_q,
            "end_time": end_q,
        },
        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 = "/",
    progress_callback: Callable[[int], None] | None = None,
) -> 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 = "/",
    progress_callback: collections.abc.Callable[[int], None] | None = None,
) -> 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, progress_callback=progress_callback
    )