Skip to content

SDK Models | spectrumx.models

spectrumx.models

Data models for the SpectrumX Data System SDK.

Modules:

Name Description
base
captures

Capture model for SpectrumX.

datasets

Dataset model for SpectrumX.

files

Data models for the SpectrumX Data System SDK.

Classes:

Name Description
SDSModel

Base class for most models in SDS.

Classes

SDSModel

Bases: BaseModel


              flowchart TD
              spectrumx.models.SDSModel[SDSModel]

              

              click spectrumx.models.SDSModel href "" "spectrumx.models.SDSModel"
            

Base class for most models in SDS.

Modules

base

Classes:

Name Description
SDSModel

Base class for most models in SDS.

Classes
SDSModel

Bases: BaseModel


              flowchart TD
              spectrumx.models.base.SDSModel[SDSModel]

              

              click spectrumx.models.base.SDSModel href "" "spectrumx.models.base.SDSModel"
            

Base class for most models in SDS.

captures

Capture model for SpectrumX.

Classes:

Name Description
Capture

A capture in SDS. A collection of spectrum data files that is indexed.

CaptureType

Capture types in SDS.

Classes
Capture

Bases: SDSModel


              flowchart TD
              spectrumx.models.captures.Capture[Capture]
              spectrumx.models.base.SDSModel[SDSModel]

                              spectrumx.models.base.SDSModel --> spectrumx.models.captures.Capture
                


              click spectrumx.models.captures.Capture href "" "spectrumx.models.captures.Capture"
              click spectrumx.models.base.SDSModel href "" "spectrumx.models.base.SDSModel"
            

A capture in SDS. A collection of spectrum data files that is indexed.

CaptureOrigin

Bases: StrEnum


              flowchart TD
              spectrumx.models.captures.CaptureOrigin[CaptureOrigin]

              

              click spectrumx.models.captures.CaptureOrigin href "" "spectrumx.models.captures.CaptureOrigin"
            

Capture origins in SDS.

CaptureType

Bases: StrEnum


              flowchart TD
              spectrumx.models.captures.CaptureType[CaptureType]

              

              click spectrumx.models.captures.CaptureType href "" "spectrumx.models.captures.CaptureType"
            

Capture types in SDS.

datasets

Dataset model for SpectrumX.

Classes:

Name Description
Dataset

A dataset in SDS.

Classes
Dataset

Bases: BaseModel


              flowchart TD
              spectrumx.models.datasets.Dataset[Dataset]

              

              click spectrumx.models.datasets.Dataset href "" "spectrumx.models.datasets.Dataset"
            

A dataset in SDS.

files

Data models for the SpectrumX Data System SDK.

Modules:

Name Description
file

File model for SpectrumX.

permission

Classes:

Name Description
File

A file in SDS.

PermissionRepresentation
Classes
File

Bases: SDSModel


              flowchart TD
              spectrumx.models.files.File[File]
              spectrumx.models.base.SDSModel[SDSModel]

                              spectrumx.models.base.SDSModel --> spectrumx.models.files.File
                


              click spectrumx.models.files.File href "" "spectrumx.models.files.File"
              click spectrumx.models.base.SDSModel href "" "spectrumx.models.base.SDSModel"
            

A file in SDS.

Attributes:

Name Type Description
created_at datetime

The timestamp when the file was created.

directory Annotated[PurePosixPath, Field(default_factory=PurePosixPath)]

The path to the file, relative to the user storage in SDS.

expiration_date datetime | None

The date when the file will be marked for deletion from SDS.

media_type str

The MIME type of the file.

name str

The user-defined name for this file.

permissions UnixPermissionStr

The permissions for the file.

size int

The size of the file in bytes.

sum_blake3 str | None

The BLAKE3 checksum of the file.

updated_at datetime

The timestamp when the file was last updated.

uuid UUID4 | None

The unique identifier of the file in SDS.

# local attributes
is_sample bool

Sample files are not written to disk (used in dry-run mode)

local_path Path | None

The path to the file on the local filesystem (includes name)

Methods:

Name Description
compute_sum_blake3

Calculates the BLAKE3 checksum of the file.

is_same_contents

Checks if the file has the same contents as another local file.

model_post_init

Post-initialization steps.

Attributes
chmod_props property
chmod_props: str

Converts human-readable permissions to chmod properties.

is_local property
is_local: bool

Checks if the file contents are available locally.

path property
path: PurePosixPath

Returns the path to the file, relative to the owner's root on SDS.

Functions
compute_sum_blake3
compute_sum_blake3() -> str | None

Calculates the BLAKE3 checksum of the file.

If the file is currently being downloaded, this method will block until the download is complete or terminated.

Parameters:

Name Type Description Default
block

when True, waits until the file contents are unlocked.

required

Returns: The BLAKE3 checksum of the file, OR None if the file is not available locally.

Source code in spectrumx/models/files/file.py
def compute_sum_blake3(self) -> str | None:
    """Calculates the BLAKE3 checksum of the file.

    If the file is currently being downloaded, this method will
        block until the download is complete or terminated.

    Args:
        block: when True, waits until the file contents are unlocked.
    Returns:
        The BLAKE3 checksum of the file,
            OR None if the file is not available locally.
    """
    # we deliberately don't cache the checksum because the file
    # might be changed externally by a different process or the user.
    with self.contents_lock:
        # content downloads cannot start within this block
        if not self.is_local:
            return None
        # this should not happen anyway, so let's just assert
        assert self.local_path is not None, "Local path is not set"
        return utils.sum_blake3(self.local_path)
is_same_contents
is_same_contents(
    other: File, *, verbose: bool = False
) -> bool

Checks if the file has the same contents as another local file.

Source code in spectrumx/models/files/file.py
def is_same_contents(self, other: "File", *, verbose: bool = False) -> bool:
    """Checks if the file has the same contents as another local file."""
    this_sum = self.compute_sum_blake3()
    other_sum = other.compute_sum_blake3()
    if verbose:
        utils.log_user(f"{this_sum}  {self.local_path}")
        utils.log_user(f"{other_sum}  {other.local_path}")
    return this_sum == other_sum
model_post_init
model_post_init(_) -> None

Post-initialization steps.

Source code in spectrumx/models/files/file.py
def model_post_init(self, _) -> None:
    """Post-initialization steps."""
PermissionRepresentation

Bases: StrEnum


              flowchart TD
              spectrumx.models.files.PermissionRepresentation[PermissionRepresentation]

              

              click spectrumx.models.files.PermissionRepresentation href "" "spectrumx.models.files.PermissionRepresentation"
            
Modules
file

File model for SpectrumX.

Classes:

Name Description
File

A file in SDS.

Classes
File

Bases: SDSModel


              flowchart TD
              spectrumx.models.files.file.File[File]
              spectrumx.models.base.SDSModel[SDSModel]

                              spectrumx.models.base.SDSModel --> spectrumx.models.files.file.File
                


              click spectrumx.models.files.file.File href "" "spectrumx.models.files.file.File"
              click spectrumx.models.base.SDSModel href "" "spectrumx.models.base.SDSModel"
            

A file in SDS.

Attributes:

Name Type Description
created_at datetime

The timestamp when the file was created.

directory Annotated[PurePosixPath, Field(default_factory=PurePosixPath)]

The path to the file, relative to the user storage in SDS.

expiration_date datetime | None

The date when the file will be marked for deletion from SDS.

media_type str

The MIME type of the file.

name str

The user-defined name for this file.

permissions UnixPermissionStr

The permissions for the file.

size int

The size of the file in bytes.

sum_blake3 str | None

The BLAKE3 checksum of the file.

updated_at datetime

The timestamp when the file was last updated.

uuid UUID4 | None

The unique identifier of the file in SDS.

# local attributes
is_sample bool

Sample files are not written to disk (used in dry-run mode)

local_path Path | None

The path to the file on the local filesystem (includes name)

Methods:

Name Description
compute_sum_blake3

Calculates the BLAKE3 checksum of the file.

is_same_contents

Checks if the file has the same contents as another local file.

model_post_init

Post-initialization steps.

Attributes
chmod_props property
chmod_props: str

Converts human-readable permissions to chmod properties.

is_local property
is_local: bool

Checks if the file contents are available locally.

path property
path: PurePosixPath

Returns the path to the file, relative to the owner's root on SDS.

Functions
compute_sum_blake3
compute_sum_blake3() -> str | None

Calculates the BLAKE3 checksum of the file.

If the file is currently being downloaded, this method will block until the download is complete or terminated.

Parameters:

Name Type Description Default
block

when True, waits until the file contents are unlocked.

required

Returns: The BLAKE3 checksum of the file, OR None if the file is not available locally.

Source code in spectrumx/models/files/file.py
def compute_sum_blake3(self) -> str | None:
    """Calculates the BLAKE3 checksum of the file.

    If the file is currently being downloaded, this method will
        block until the download is complete or terminated.

    Args:
        block: when True, waits until the file contents are unlocked.
    Returns:
        The BLAKE3 checksum of the file,
            OR None if the file is not available locally.
    """
    # we deliberately don't cache the checksum because the file
    # might be changed externally by a different process or the user.
    with self.contents_lock:
        # content downloads cannot start within this block
        if not self.is_local:
            return None
        # this should not happen anyway, so let's just assert
        assert self.local_path is not None, "Local path is not set"
        return utils.sum_blake3(self.local_path)
is_same_contents
is_same_contents(
    other: File, *, verbose: bool = False
) -> bool

Checks if the file has the same contents as another local file.

Source code in spectrumx/models/files/file.py
def is_same_contents(self, other: "File", *, verbose: bool = False) -> bool:
    """Checks if the file has the same contents as another local file."""
    this_sum = self.compute_sum_blake3()
    other_sum = other.compute_sum_blake3()
    if verbose:
        utils.log_user(f"{this_sum}  {self.local_path}")
        utils.log_user(f"{other_sum}  {other.local_path}")
    return this_sum == other_sum
model_post_init
model_post_init(_) -> None

Post-initialization steps.

Source code in spectrumx/models/files/file.py
def model_post_init(self, _) -> None:
    """Post-initialization steps."""
Modules
permission

Classes:

Name Description
PermissionRepresentation
Classes
PermissionRepresentation

Bases: StrEnum


              flowchart TD
              spectrumx.models.files.permission.PermissionRepresentation[PermissionRepresentation]

              

              click spectrumx.models.files.permission.PermissionRepresentation href "" "spectrumx.models.files.permission.PermissionRepresentation"
            
Functions
octal_to_unix_perm_string
octal_to_unix_perm_string(value: int) -> str

Convert an octal Unix file permission to a string representation.

:param value: An int representing the file permission in octal format (e.g., 0o755) :return: A 9-character string representing Unix permissions (e.g., 'rwxr-xr-x') :raises ValueError: If the input is not a valid octal Unix permission.

Source code in spectrumx/models/files/permission.py
def octal_to_unix_perm_string(value: int) -> str:
    """
    Convert an octal Unix file permission to a string representation.

    :param value: An int representing the file permission in octal format (e.g., 0o755)
    :return: A 9-character string representing Unix permissions (e.g., 'rwxr-xr-x')
    :raises ValueError: If the input is not a valid octal Unix permission.
    """
    perm_oct = TypeAdapter(PermissionOct).validate_python(value)

    mapping = {
        0: "---",
        1: "--x",
        2: "-w-",
        3: "-wx",
        4: "r--",
        5: "r-x",
        6: "rw-",
        7: "rwx",
    }

    return "".join(mapping[(perm_oct >> (3 * i)) & 0o7] for i in reversed(range(3)))
serialize_unix_permission
serialize_unix_permission(
    value: Any, info: SerializationInfo
)

Serialize Unix permission based on context (octal or string).

Source code in spectrumx/models/files/permission.py
def serialize_unix_permission(value: Any, info: SerializationInfo):
    """Serialize Unix permission based on context (`octal` or `string`)."""

    # If we have no context return the string representation
    if info.context is None:
        return value

    # Get the mode from the context and default to string
    mode = info.context.get("mode", PermissionRepresentation.STRING)
    return PermissionRepresentation(mode).convert(value)
unix_perm_string_to_octal
unix_perm_string_to_octal(value: str) -> int | ValueError

Convert a Unix permission string (e.g., 'rwxr-xr-x') to an octal integer.

:param value: A 9-character Unix permission string (e.g., 'rw-r--r--') :return: An integer representing the octal value of the permission (e.g., 0o755) :raises ValueError: If the input string is not a valid Unix permission string.

Source code in spectrumx/models/files/permission.py
def unix_perm_string_to_octal(value: str) -> int | ValueError:
    """
    Convert a Unix permission string (e.g., 'rwxr-xr-x') to an octal integer.

    :param value: A 9-character Unix permission string (e.g., 'rw-r--r--')
    :return: An integer representing the octal value of the permission (e.g., 0o755)
    :raises ValueError: If the input string is not a valid Unix permission string.
    """
    perm_str = PermissionStrAdapter.validate_python(value)

    one_hot = "".join(["0" if flag == "-" else "1" for flag in perm_str])
    return int(one_hot, base=2)