Datasets API | <Client>.datasets

spectrumx.api.datasets

API functions specific to datasets.

Classes:

Name	Description
DatasetAPI

Classes

DatasetAPI

DatasetAPI(
    *,
    gateway: GatewayClient,
    dry_run: bool = True,
    verbose: bool = False,
)

Methods:

Name	Description
delete	Deletes a dataset from SDS by its UUID.
get	Load dataset metadata, captures, and artifact files from SDS.
get_files	Get files in the dataset as a paginator.
list_artifact_files	Return file rows linked directly to the dataset (artifacts), as JSON dicts.
list_captures	Return capture payloads linked to the dataset (raw JSON objects).
revoke_share_permissions	Revoke all direct share permissions on this dataset (owner-only).

Source code in spectrumx/api/datasets.py

def __init__(
    self,
    *,
    gateway: GatewayClient,
    dry_run: bool = True,
    verbose: bool = False,
) -> None:
    """Initializes the DatasetAPI."""
    self.dry_run = dry_run
    self.gateway = gateway
    self.verbose = verbose

Methods:

delete

delete(dataset_uuid: UUID) -> bool

Deletes a dataset from SDS by its UUID.

Parameters:

Name	Type	Description	Default
dataset_uuid	UUID	The UUID of the dataset to delete.	required

Returns: True if the dataset was deleted successfully, or if in dry run mode.

Source code in spectrumx/api/datasets.py

def delete(
    self,
    dataset_uuid: UUID,
) -> bool:
    """Deletes a dataset from SDS by its UUID.

    Args:
        dataset_uuid: The UUID of the dataset to delete.
    Returns:
        True if the dataset was deleted successfully, or if in dry run mode.
    """
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Deleting dataset with UUID {dataset_uuid}"
        )

    if self.dry_run:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Dry run enabled: would delete dataset {dataset_uuid}"
        )
        return True

    self.gateway.delete_dataset(
        dataset_uuid=dataset_uuid,
    )
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Dataset deleted with UUID {dataset_uuid}"
        )
    return True

get

get(dataset_uuid: UUID) -> Dataset

Load dataset metadata, captures, and artifact files from SDS.

Captures are returned in the same grouped shape as the capture list API (one entry per logical multi-channel capture where applicable). For every file in the dataset (including capture-linked files), use :meth:get_files instead, which calls the paginated dataset files manifest endpoint.

Source code in spectrumx/api/datasets.py

def get(self, dataset_uuid: UUID) -> Dataset:
    """Load dataset metadata, captures, and artifact files from SDS.

    Captures are returned in the same grouped shape as the capture list API
    (one entry per logical multi-channel capture where applicable). For every
    file in the dataset (including capture-linked files), use :meth:`get_files`
    instead, which calls the paginated dataset files manifest endpoint.
    """
    if self.dry_run:
        log_user("Dry run enabled: returning an empty Dataset shell")
        return Dataset(uuid=dataset_uuid)

    raw = self.gateway.get_dataset(
        dataset_uuid=dataset_uuid,
        verbose=self.verbose,
    )
    return Dataset.model_validate_json(raw)

get_files

get_files(
    dataset_uuid: UUID,
    *,
    capture_uuids: Collection[UUID] | None = None,
    top_level_dirs: Collection[PurePosixPath | Path | str]
    | None = None,
    artifacts_only: bool = False,
) -> Paginator[File]

Get files in the dataset as a paginator.

Parameters:

Name	Type	Description	Default
dataset_uuid	UUID	The UUID of the dataset to get files for.	required
capture_uuids	Collection[UUID] | None	If set, passed to the gateway to restrict by capture UUID (OR with top_level_dirs).	None
top_level_dirs	Collection[PurePosixPath | Path | str] | None	If set, passed to the gateway as path prefixes under File.directory (OR with capture_uuids).	None
artifacts_only	bool	If set, only return artifact files (not capture-linked files).	False

Returns: A paginator for the files in the dataset.

Source code in spectrumx/api/datasets.py

def get_files(
    self,
    dataset_uuid: UUID,
    *,
    capture_uuids: Collection[UUID] | None = None,
    top_level_dirs: Collection[PurePosixPath | Path | str] | None = None,
    artifacts_only: bool = False,
) -> Paginator[File]:
    """Get files in the dataset as a paginator.

    Args:
        dataset_uuid: The UUID of the dataset to get files for.
        capture_uuids: If set, passed to the gateway to restrict by capture UUID (OR
            with ``top_level_dirs``).
        top_level_dirs: If set, passed to the gateway as path prefixes under
            ``File.directory`` (OR with ``capture_uuids``).
        artifacts_only: If set, only return artifact files (not capture-linked
            files).
    Returns:
        A paginator for the files in the dataset.
    """
    if self.dry_run:
        log_user("Dry run enabled: files will be simulated")

    if artifacts_only and capture_uuids:
        log.bind(cat=LogCategory.FILESYSTEM).warning(
            "Capture UUIDs are not allowed when artifacts_only is True."
        )
        capture_uuids = None

    if artifacts_only and top_level_dirs:
        log.bind(cat=LogCategory.FILESYSTEM).info(
            "Top level directories included will ONLY return artifact files."
        )

    list_kwargs: dict[str, Any] = {"dataset_uuid": dataset_uuid}
    if capture_uuids is not None:
        list_kwargs["capture_uuids"] = tuple(capture_uuids)
    if top_level_dirs is not None:
        list_kwargs["top_level_dirs"] = tuple(str(p) for p in top_level_dirs)
    if artifacts_only:
        list_kwargs["artifacts_only"] = True

    pagination: Paginator[File] = Paginator(
        Entry=File,
        gateway=self.gateway,
        list_method=self.gateway.get_dataset_files,
        list_kwargs=list_kwargs,
        dry_run=self.dry_run,
        verbose=self.verbose,
    )

    return pagination

list_artifact_files

list_artifact_files(
    dataset_uuid: UUID,
) -> list[dict[str, Any]]

Return file rows linked directly to the dataset (artifacts), as JSON dicts.

These are the same objects embedded on :meth:get under the files key. For the full downloadable manifest (captures plus artifacts), use :meth:get_files.

Source code in spectrumx/api/datasets.py

def list_artifact_files(self, dataset_uuid: UUID) -> list[dict[str, Any]]:
    """Return file rows linked directly to the dataset (artifacts), as JSON dicts.

    These are the same objects embedded on :meth:`get` under the ``files`` key.
    For the full downloadable manifest (captures plus artifacts), use
    :meth:`get_files`.
    """
    if self.dry_run:
        log_user("Dry run enabled: returning an empty artifact file list")
        return []

    raw = self.gateway.get_dataset(
        dataset_uuid=dataset_uuid,
        verbose=self.verbose,
    )
    data = json.loads(raw)
    files = data.get("files")
    return list(files) if isinstance(files, list) else []

list_captures

list_captures(dataset_uuid: UUID) -> list[dict[str, Any]]

Return capture payloads linked to the dataset (raw JSON objects).

Use this when you need composite capture fields (for example channels) without coercing through :class:~spectrumx.models.datasets.DatasetCapture.

Source code in spectrumx/api/datasets.py

def list_captures(self, dataset_uuid: UUID) -> list[dict[str, Any]]:
    """Return capture payloads linked to the dataset (raw JSON objects).

    Use this when you need composite capture fields (for example ``channels``)
    without coercing through :class:`~spectrumx.models.datasets.DatasetCapture`.
    """
    if self.dry_run:
        log_user("Dry run enabled: returning an empty capture list")
        return []

    raw = self.gateway.get_dataset(
        dataset_uuid=dataset_uuid,
        verbose=self.verbose,
    )
    data = json.loads(raw)
    captures = data.get("captures")
    return list(captures) if isinstance(captures, list) else []

revoke_share_permissions

revoke_share_permissions(dataset_uuid: UUID) -> bool

Revoke all direct share permissions on this dataset (owner-only).

Use this (or the web portal) before :meth:delete when the dataset is shared.

Source code in spectrumx/api/datasets.py

def revoke_share_permissions(self, dataset_uuid: UUID) -> bool:
    """Revoke all direct share permissions on this dataset (owner-only).

    Use this (or the web portal) before :meth:`delete` when the dataset is shared.
    """
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Revoking share permissions for dataset {dataset_uuid}"
        )
    if self.dry_run:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            "Dry run enabled: would revoke dataset share permissions"
        )
        return True
    self.gateway.revoke_dataset_share_permissions(dataset_uuid=dataset_uuid)
    return True

Functions: