Imports

imports

Service class for handling data imports into Kaleidoscope workspace.

This module provides the ImportsService class, which facilitates pushing data records into the Kaleidoscope system. It supports flexible data import operations, allowing organization by experiments, programs, and data sources.

Classes:

Name	Description
ImportsService	Service for handling data imports into the Kaleidoscope workspace, providing methods to push records organized by source, experiment, program, record views, and set names.

Example

key_fields = ["id", "timestamp"]
records = [
    {"id": "001", "timestamp": "2024-01-01", "value": 42.5, "status": "active"},
    {"id": "002", "timestamp": "2024-01-02", "value": 38.7, "status": "pending"}
]
# Push data to a specific source and experiment
import_id = client.imports.push_data(
    key_field_names=key_fields,
    data=records,
    source_id="data_source_123",
    operation_id="exp_456",
    set_name="january_batch"
)

ImportResult

Bases: BaseModel

Results summary from a completed import.

ImportRecord

Bases: BaseModel

Represents an import record returned by the imports API.

ImportsService

ImportsService(client: KaleidoscopeClient)

Service class for handling data imports into Kaleidoscope workspace.

This service provides functionality to push data records into the workspace, with support for organizing data by sources, experiments, programs, and record views.

Methods:

Name	Description
push_data	Import data into the workspace using field names.
push_data_by_field_id	Import data into the workspace using field UUIDs instead of field names.
get_imports	Retrieve imports in the workspace.
get_import	Retrieve a specific import by ID.

Source code in kalbio/imports.py

def __init__(self, client: KaleidoscopeClient):
    self._client = client

push_data

push_data(key_field_names: list[str], data: list[dict[str, Any]], source_id: str | None = None, operation_id: str | None = None, program_id: str | None = None, record_view_id: str | None = None, add_fields_to_record_view_ids: list[str] | None = None, set_name: str | None = None, record_view_ids: list[str] | None = None) -> str | None

Import data into the workspace using field names.

Sends a list of records, each represented as a dictionary of field names and values, to the Kaleidoscope workspace.

Parameters:

Name	Type	Description	Default
key_field_names	list[str]	List of field names that serve as keys for the records.	required
data	list[dict[str, Any]]	List of records to import, each as a dictionary mapping field names to values.	required
source_id	str	Identifier for the data source. If provided, data is imported under this source.	None
operation_id	str	Identifier for the operation/experiment. If provided, data is imported into this specific operation.	None
program_id	str	Identifier for the program. If provided, data is imported under this program.	None
record_view_id	str	UUID of the record view on the operation to import the data into. Use Activity.record_views to discover the available views on an operation. Only meaningful when operation_id is also provided.	None
add_fields_to_record_view_ids	list[str]	UUIDs of additional record views that any newly-created fields from this import should be added to. Distinct from record_view_id: these views are not the target of the import, they just get the new fields added to them.	None
set_name	str	Name of the set to which the imported data belongs.	None
record_view_ids	list[str]	Deprecated. The server only accepts a singular record_view_id for operation-targeted imports plus add_fields_to_record_view_ids for adding fields to extra views. Values passed here are ignored. Switch to record_view_id and/or add_fields_to_record_view_ids.	None

Returns:

Name	Type	Description
str	str | None	The import ID for the created import, or None if the request failed.

Example

# Import into a specific record view on an operation
activity = client.activities.get_activity_by_id("op_uuid")
view = activity.record_views[0]
client.imports.push_data(
    key_field_names=["id"],
    data=[{"id": "S-001", "yield": 42.5}],
    operation_id=activity.id,
    record_view_id=view.id,
)

Source code in kalbio/imports.py

def push_data(
    self,
    key_field_names: list[str],
    data: list[dict[str, Any]],
    source_id: Optional[str] = None,
    operation_id: Optional[str] = None,
    program_id: Optional[str] = None,
    record_view_id: Optional[str] = None,
    add_fields_to_record_view_ids: Optional[list[str]] = None,
    set_name: Optional[str] = None,
    record_view_ids: Optional[list[str]] = None,
) -> Optional[str]:
    """Import data into the workspace using field names.

    Sends a list of records, each represented as a dictionary of field names and values,
    to the Kaleidoscope workspace.

    Args:
        key_field_names (list[str]): List of field names that serve as keys for the records.
        data (list[dict[str, Any]]): List of records to import, each as a dictionary mapping field names to values.
        source_id (str, optional): Identifier for the data source. If provided, data is imported under this source.
        operation_id (str, optional): Identifier for the operation/experiment. If provided, data is imported into this specific operation.
        program_id (str, optional): Identifier for the program. If provided, data is imported under this program.
        record_view_id (str, optional): UUID of the record view on the operation
            to import the data into. Use `Activity.record_views` to discover the
            available views on an operation. Only meaningful when `operation_id`
            is also provided.
        add_fields_to_record_view_ids (list[str], optional): UUIDs of additional
            record views that any newly-created fields from this import should be
            added to. Distinct from `record_view_id`: these views are not the
            target of the import, they just get the new fields added to them.
        set_name (str, optional): Name of the set to which the imported data belongs.
        record_view_ids (list[str], optional): Deprecated. The server only accepts
            a singular `record_view_id` for operation-targeted imports plus
            `add_fields_to_record_view_ids` for adding fields to extra views.
            Values passed here are ignored. Switch to `record_view_id` and/or
            `add_fields_to_record_view_ids`.

    Returns:
        str: The import ID for the created import, or None if the request failed.

    Example:
        ```python
        # Import into a specific record view on an operation
        activity = client.activities.get_activity_by_id("op_uuid")
        view = activity.record_views[0]
        client.imports.push_data(
            key_field_names=["id"],
            data=[{"id": "S-001", "yield": 42.5}],
            operation_id=activity.id,
            record_view_id=view.id,
        )
        ```
    """
    if record_view_ids is not None:
        warnings.warn(
            "`record_view_ids` (plural) is deprecated and was silently ignored "
            "by the server. Use `record_view_id` (singular) to target a specific "
            "record view on the operation, and/or `add_fields_to_record_view_ids` "
            "to add new fields to additional views.",
            DeprecationWarning,
            stacklevel=2,
        )

    payload: dict[str, Any] = {
        "key_field_names": key_field_names,
        "data": data,
    }

    if program_id:
        payload["program_id"] = program_id
    if operation_id:
        payload["operation_id"] = operation_id
    if record_view_id:
        payload["record_view_id"] = record_view_id
    if add_fields_to_record_view_ids:
        payload["add_fields_to_record_view_ids"] = add_fields_to_record_view_ids
    if set_name:
        payload["set_name"] = set_name

    url = "/push/imports"
    if source_id:
        url = url + f"/{source_id}"

    resp = self._client._post(url, payload)
    if resp and "import_id" in resp:
        return resp["import_id"]
    return None

push_data_by_field_id

push_data_by_field_id(key_field_ids: list[str], data: list[dict[str, Any]], operation_id: str | None = None, program_id: str | None = None, record_view_ids: list[str] | None = None, set_name: str | None = None) -> str | None

Import data into the workspace using field UUIDs instead of field names.

Sends a list of records where keys are field UUIDs rather than field names.

Parameters:

Name	Type	Description	Default
key_field_ids	list[str]	List of field UUIDs that serve as keys for the records.	required
data	list[dict[str, Any]]	List of records to import, each as a dictionary mapping field UUIDs to values.	required
operation_id	str	Identifier for the experiment.	None
program_id	str	Identifier for the program.	None
record_view_ids	list[str]	List of record view IDs to associate with the imported data.	None
set_name	str	Name of the set to which the imported data belongs.	None

Returns:

Name	Type	Description
str	str | None	The import ID for the created import, or None if the request failed.

Source code in kalbio/imports.py

def push_data_by_field_id(
    self,
    key_field_ids: list[str],
    data: list[dict[str, Any]],
    operation_id: Optional[str] = None,
    program_id: Optional[str] = None,
    record_view_ids: Optional[list[str]] = None,
    set_name: Optional[str] = None,
) -> Optional[str]:
    """Import data into the workspace using field UUIDs instead of field names.

    Sends a list of records where keys are field UUIDs rather than field names.

    Args:
        key_field_ids (list[str]): List of field UUIDs that serve as keys for the records.
        data (list[dict[str, Any]]): List of records to import, each as a dictionary mapping field UUIDs to values.
        operation_id (str, optional): Identifier for the experiment.
        program_id (str, optional): Identifier for the program.
        record_view_ids (list[str], optional): List of record view IDs to associate with the imported data.
        set_name (str, optional): Name of the set to which the imported data belongs.

    Returns:
        str: The import ID for the created import, or None if the request failed.
    """
    if record_view_ids is None:
        record_view_ids = []

    payload = {
        "key_field_ids": key_field_ids,
        "data": data,
        "record_view_ids": record_view_ids,
    }

    if program_id:
        payload["program_id"] = program_id
    if operation_id:
        payload["operation_id"] = operation_id
    if set_name:
        payload["set_name"] = set_name

    resp = self._client._post("/push/imports/by-field-id", payload)
    if resp and "import_id" in resp:
        return resp["import_id"]
    return None

get_imports

get_imports(is_complete: bool | None = None, page: int | None = None, page_size: int | None = None) -> list[ImportRecord]

Retrieve imports in the workspace.

Parameters:

Name	Type	Description	Default
is_complete	bool	Filter by completion status.	None
page	int	Page number for pagination.	None
page_size	int	Number of items per page.	None

Returns:

Type	Description
list[ImportRecord]	List[ImportRecord]: A list of import records.

Source code in kalbio/imports.py

def get_imports(
    self,
    is_complete: Optional[bool] = None,
    page: Optional[int] = None,
    page_size: Optional[int] = None,
) -> List[ImportRecord]:
    """Retrieve imports in the workspace.

    Args:
        is_complete (bool, optional): Filter by completion status.
        page (int, optional): Page number for pagination.
        page_size (int, optional): Number of items per page.

    Returns:
        List[ImportRecord]: A list of import records.
    """
    params = {}
    if is_complete is not None:
        params["is_complete"] = 1 if is_complete else 0
    if page is not None:
        params["page"] = page
    if page_size is not None:
        params["page_size"] = page_size

    resp = self._client._get("/imports", params if params else None)
    if resp is None:
        return []
    return [ImportRecord(**record) for record in resp]

get_import

get_import(import_id: str) -> ImportRecord | None

Retrieve a specific import by ID.

Parameters:

Name	Type	Description	Default
import_id	str	The UUID of the import to retrieve.	required

Returns:

Name	Type	Description
ImportRecord	ImportRecord | None	The import record, or None if not found.

Source code in kalbio/imports.py

def get_import(self, import_id: str) -> Optional[ImportRecord]:
    """Retrieve a specific import by ID.

    Args:
        import_id (str): The UUID of the import to retrieve.

    Returns:
        ImportRecord: The import record, or None if not found.
    """
    resp = self._client._get(f"/imports/{import_id}")
    if resp is None:
        return None
    return ImportRecord(**resp)