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
add_capture_to_dataset	Add an existing capture to an existing dataset.
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:

add_capture_to_dataset

add_capture_to_dataset(
    dataset_uuid: UUID, capture_uuid: UUID
) -> bool

Add an existing capture to an existing dataset.

Source code in spectrumx/api/datasets.py

def add_capture_to_dataset(self, dataset_uuid: UUID, capture_uuid: UUID) -> bool:
    """Add an existing capture to an existing dataset."""
    if self.verbose:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Attaching capture {capture_uuid} to dataset {dataset_uuid}"
        )
    if self.dry_run:
        log.bind(cat=LogCategory.FILESYSTEM).debug(
            f"Dry run enabled: would attach capture {capture_uuid} "
            f"to dataset {dataset_uuid}"
        )
        return True
    self.gateway.add_capture_to_dataset(
        dataset_uuid=dataset_uuid,
        capture_uuid=capture_uuid,
        verbose=self.verbose,
    )
    return True

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.

Referenced by:

• API Reference Datasets API | <Client>.datasets  datasets Classes  DatasetAPI Methods:  revoke_share_permissions

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 get_files instead, which calls the paginated dataset files manifest endpoint.

Referenced by:

• API Reference Datasets API | <Client>.datasets  datasets Classes  DatasetAPI Methods:  list_artifact_files

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 [`get_files`][spectrumx.api.datasets.DatasetAPI.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.

Referenced by:

• API Reference Datasets API | <Client>.datasets  datasets Classes  DatasetAPI Methods:
  •  get
  •  list_artifact_files

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 get under the files key. For the full downloadable manifest (captures plus artifacts), use 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
    [`get`][spectrumx.api.datasets.DatasetAPI.get] under the ``files`` key.
    For the full downloadable manifest (captures plus artifacts), use
    [`get_files`][spectrumx.api.datasets.DatasetAPI.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 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
    [`DatasetCapture`][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 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
    [`delete`][spectrumx.api.datasets.DatasetAPI.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: