Records

records

Records module for managing Kaleidoscope record operations.

This module provides classes and services for interacting with records in the Kaleidoscope system. It includes functionality for filtering, sorting, managing record values, handling file attachments, and searching records.

Classes:

Name	Description
FilterRuleTypeEnum	Enumeration of available filter rule types for record filtering
ViewFieldFilter	TypedDict for view-based field filter configuration
ViewFieldSort	TypedDict for view-based field sort configuration
FieldFilter	TypedDict for entity-based field filter configuration
FieldSort	TypedDict for entity-based field sort configuration
RecordValue	Model representing a single value within a record field
Record	Model representing a complete record with all its fields and values
RecordsService	Service class providing record-related API operations

The module uses Pydantic models for data validation and serialization, and integrates with the KaleidoscopeClient for API communication.

Example

    # Get a record by ID
    record = client.records.get_record_by_id("record_uuid")

    # Add a value to a record field
    record.add_value(
        field_id="field_uuid",
        content="Experiment result",
        activity_id="activity_uuid"
    )

    # Get a field value
    value = record.get_value_content(field_id="field_uuid")

    # Update a field
    record.update_field(
        field_id="field_uuid",
        value="Updated value",
        activity_id="activity_uuid"
    )

    # Get activities associated with a record
    activities = record.get_activities()

FilterRuleTypeEnum

Bases: str, Enum

Enumeration of filter rule types for record filtering operations.

This enum defines all available filter rule types that can be applied to record properties. Filter rules are categorized into several groups:

• Existence checks: IS_SET, IS_EMPTY
• Equality checks: IS_EQUAL, IS_NOT_EQUAL, IS_ANY_OF_TEXT
• String operations: INCLUDES, DOES_NOT_INCLUDE, STARTS_WITH, ENDS_WITH
• Membership checks: IS_IN, IS_NOT_IN
• Set operations: VALUE_IS_SUBSET_OF_PROPS, VALUE_IS_SUPERSET_OF_PROPS, VALUE_HAS_OVERLAP_WITH_PROPS, VALUE_HAS_NO_OVERLAP_WITH_PROPS, VALUE_HAS_SAME_ELEMENTS_AS_PROPS
• Numeric comparisons: IS_LESS_THAN, IS_LESS_THAN_EQUAL, IS_GREATER_THAN, IS_GREATER_THAN_EQUAL
• Absolute date comparisons: IS_BEFORE, IS_AFTER, IS_BETWEEN
• Relative date comparisons:
  • Day-based: IS_BEFORE_RELATIVE_DAY, IS_AFTER_RELATIVE_DAY, IS_BETWEEN_RELATIVE_DAY
  • Week-based: IS_BEFORE_RELATIVE_WEEK, IS_AFTER_RELATIVE_WEEK, IS_BETWEEN_RELATIVE_WEEK, IS_LAST_WEEK, IS_THIS_WEEK, IS_NEXT_WEEK
  • Month-based: IS_BEFORE_RELATIVE_MONTH, IS_AFTER_RELATIVE_MONTH, IS_BETWEEN_RELATIVE_MONTH, IS_THIS_MONTH, IS_NEXT_MONTH
• Update tracking: IS_LAST_UPDATED_AFTER

Each enum value corresponds to a string representation used in filter configurations.

ViewFieldFilter

Bases: TypedDict

TypedDict for view-based field filter configuration.

Attributes:

Name	Type	Description
key_field_id	str | None	The ID of the key field to filter by.
view_field_id	str | None	The ID of the view field to filter by.
filter_type	FilterRuleTypeEnum	The type of filter rule to apply.
filter_prop	Any	The property value to filter against.

Example

from kalbio.records import FilterRuleTypeEnum, ViewFieldFilter

filter_config: ViewFieldFilter = {
    "key_field_id": "field_uuid",
    "view_field_id": None,
    "filter_type": FilterRuleTypeEnum.IS_EQUAL,
    "filter_prop": "S",
}

ViewFieldSort

Bases: TypedDict

TypedDict for view-based field sort configuration.

Attributes:

Name	Type	Description
key_field_id	str | None	The ID of the key field to sort by.
view_field_id	str | None	The ID of the view field to sort by.
descending	bool	Whether to sort in descending order.

Example

from kalbio.records import ViewFieldSort

sort_config: ViewFieldSort = {
    "key_field_id": "field_uuid",
    "view_field_id": None,
    "descending": True,
}

FieldFilter

Bases: TypedDict

TypedDict for entity-based field filter configuration.

Attributes:

Name	Type	Description
field_id	str | None	The ID of the field to filter by.
filter_type	FilterRuleTypeEnum	The type of filter rule to apply.
filter_prop	Any	The property value to filter against.

Example

from kalbio.records import FieldFilter, FilterRuleTypeEnum

field_filter: FieldFilter = {
    "field_id": "field_uuid",
    "filter_type": FilterRuleTypeEnum.STARTS_WITH,
    "filter_prop": "EXP-",
}

FieldSort

Bases: TypedDict

TypedDict for entity-based field sort configuration.

Attributes:

Name	Type	Description
field_id	str | None	The ID of the field to sort by.
descending	bool	Whether to sort in descending order.

Example

from kalbio.records import FieldSort

sort_config: FieldSort = {
    "field_id": "field_uuid",
    "descending": False,
}

RecordValue

Bases: _KaleidoscopeBaseModel

Represents a single value entry in a record within the Kaleidoscope system.

A RecordValue stores the actual content of a record along with metadata about when it was created and its relationships to parent records and operations.

Attributes:

Name	Type	Description
id	str	UUID of the record value
content	Any	The actual data value stored in this record. Can be of any type.
created_at	datetime | None	Timestamp indicating when this value was created. Defaults to None.
record_id	str | None	Identifier of the parent record this value belongs to. Defaults to None.
operation_id	str | None	Identifier of the operation that created or modified this value. Defaults to None.

Example

from datetime import datetime
from kalbio.records import RecordValue

value = RecordValue(
    id="value_uuid",
    content="Completed",
    created_at=datetime.utcnow(),
    record_id="record_uuid",
    operation_id="activity_uuid",
)

Record

Bases: _KaleidoscopeBaseModel

Represents a record in the Kaleidoscope system.

A Record is a core data structure that contains values organized by fields, can be associated with experiments, and may have sub-records. Records are identified by a unique ID and belong to an entity slice.

Attributes:

Name	Type	Description
id	str	UUID of the record.
created_at	datetime	The timestamp when the record was created.
entity_slice_id	str	The ID of the entity slice this record belongs to.
identifier_ids	list[str]	A list of identifier IDs associated with this record.
record_identifier	str	Human-readable identifier string for the record.
record_values	dict[str, list[RecordValue]]	A dictionary mapping field IDs to lists of record values.
initial_operation_id	str | None	The ID of the initial operation that created this record, if applicable.
sub_record_ids	list[str]	A list of IDs for sub-records associated with this record.

Example

from kalbio.client import KaleidoscopeClient

client = KaleidoscopeClient()
record = client.records.get_record_by_id("record_uuid")
latest_value = record.get_value_content(field_id="field_uuid")
print(record.record_identifier, latest_value)

Methods:

Name	Description
get_activities	Retrieves a list of activities associated with this record.
add_value	Adds a value to a specified field for a given activity.
get_value_content	Retrieves the content of a record value for a specified field.
get_activity_data	Retrieves activity data for a specific activity ID.
update_field	Updates a specific field of the record with the given value.
update_field_file	Update a record value with a file.
get_values	Retrieve all values associated with this record.
refetch	Refreshes all the data of the current record instance.

get_activities

get_activities() -> list['Activity']

Retrieves a list of activities associated with this record.

Returns:

Type	Description
list['Activity']	A list of activities related to this record.

Note

If an exception occurs during the API request, it logs the error and returns an empty list.

Example

activities = record.get_activities()
for activity in activities:
    print(activity.id)

Source code in kalbio/records.py

def get_activities(self) -> List["Activity"]:
    """Retrieves a list of activities associated with this record.

    Returns:
        A list of activities related to this record.

    Note:
        If an exception occurs during the API request, it logs the error and returns an empty list.

    Example:
        ```python
        activities = record.get_activities()
        for activity in activities:
            print(activity.id)
        ```
    """
    return self._client.activities.get_activities_with_record(self.id)

add_value

add_value(field_id: EntityFieldIdentifier, content: Any, activity_id: ActivityIdentifier | None = None) -> None

Adds a value to a specified field for a given activity.

Parameters:

Name	Type	Description	Default
field_id	EntityFieldIdentifier	Identifier of the field to which the value will be added. Any type of EntityFieldIdentifier will be accepted and resolved.	required
content	Any	The value/content to be saved for the field.	required
activity_id	ActivityIdentifier | None	The identifier of the activity. Defaults to None. Any type of EntityFieldIdentifier will be accepted and resolved.	None

Example

record.add_value(
    field_id="field_uuid",
    content="Experiment result",
    activity_id="activity_uuid",
)

Source code in kalbio/records.py

def add_value(
    self,
    field_id: EntityFieldIdentifier,
    content: Any,
    activity_id: Optional[ActivityIdentifier] = None,
) -> None:
    """Adds a value to a specified field for a given activity.

    Args:
        field_id: Identifier of the field to which the value will be added.

            Any type of EntityFieldIdentifier will be accepted and resolved.
        content: The value/content to be saved for the field.
        activity_id: The identifier of the activity. Defaults to None.

            Any type of EntityFieldIdentifier will be accepted and resolved.

    Example:
        ```python
        record.add_value(
            field_id="field_uuid",
            content="Experiment result",
            activity_id="activity_uuid",
        )
        ```
    """
    try:
        self._client._post(
            "/records/" + self.id + "/values",
            {
                "content": content,
                "field_id": self._client.entity_fields._resolve_data_field_id(
                    field_id
                ),
                "operation_id": (
                    self._client.activities._resolve_activity_id(activity_id)
                ),
            },
        )
        self.refetch()
        return
    except Exception as e:
        _logger.error(f"Error adding this value: {e}")
        return

get_value_content

get_value_content(field_id: EntityFieldIdentifier, activity_id: ActivityIdentifier | None = None, include_sub_record_values: bool | None = False, sub_record_id: 'RecordIdentifier' | None = None) -> Any | None

Retrieves the content of a record value for a specified field.

Optionally filtered by activity, sub-record, and inclusion of sub-record values.

Parameters:

Name	Type	Description	Default
field_id	EntityFieldIdentifier	The ID of the field to retrieve the value for.	required
activity_id	ActivityIdentifier | None	The ID of the activity to filter values by. Defaults to None.	None
include_sub_record_values	bool | None	Whether to include values from sub-records. Defaults to False.	False
sub_record_id	'RecordIdentifier' | None	The ID of a specific sub-record to filter values by. Defaults to None.	None

Returns:

Type	Description
Any | None	The content of the most recent matching record value, or None if no value is found.

Example

latest_content = record.get_value_content(
    field_id="field_uuid",
    activity_id="activity_uuid",
    include_sub_record_values=True,
)
print(latest_content)

Source code in kalbio/records.py

def get_value_content(
    self,
    field_id: EntityFieldIdentifier,
    activity_id: Optional[ActivityIdentifier] = None,
    include_sub_record_values: Optional[bool] = False,
    sub_record_id: Optional["RecordIdentifier"] = None,
) -> Any | None:
    """Retrieves the content of a record value for a specified field.

    Optionally filtered by activity, sub-record, and inclusion of sub-record values.

    Args:
        field_id: The ID of the field to retrieve the value for.
        activity_id: The ID of the activity to filter values by. Defaults to None.
        include_sub_record_values: Whether to include values from sub-records. Defaults to False.
        sub_record_id: The ID of a specific sub-record to filter values by. Defaults to None.

    Returns:
        The content of the most recent matching record value, or None if no value is found.

    Example:
        ```python
        latest_content = record.get_value_content(
            field_id="field_uuid",
            activity_id="activity_uuid",
            include_sub_record_values=True,
        )
        print(latest_content)
        ```
    """
    field_uuid = self._client.entity_fields._resolve_data_field_id(field_id)
    activity_uuid = self._client.activities._resolve_activity_id(activity_id)
    sub_record_uuid = self._client.records._resolve_to_record_id(sub_record_id)

    if not field_uuid:
        return None

    values = self.record_values.get(field_uuid)
    if not values:
        return None

    # include key values in the activity data (record_id = None)
    if activity_uuid is not None:
        values = [
            value
            for value in values
            if (value.operation_id == activity_uuid) or value.record_id is None
        ]

    if not include_sub_record_values:
        # key values have None for the record_id
        values = [
            value
            for value in values
            if value.record_id == self.id or value.record_id is None
        ]

    if sub_record_uuid:
        values = [value for value in values if value.record_id == sub_record_uuid]

    sorted_values: List[RecordValue] = sorted(
        values,
        key=lambda x: x.created_at if x.created_at else datetime.min,
        reverse=True,
    )
    value = next(iter(sorted_values), None)
    return value.content if value else None

get_activity_data

get_activity_data(activity_id: ActivityIdentifier) -> dict

Retrieves activity data for a specific activity ID.

Parameters:

Name	Type	Description	Default
activity_id	ActivityIdentifier	The identifier of the activity. Any type of ActivityIdentifier will be accepted and resolved.	required

Returns:

Type	Description
dict	A dictionary mapping field IDs to their corresponding values for the given activity.
dict	Only fields with non-None values are included.

Example

activity_data = record.get_activity_data(activity_id="activity_uuid")
print(activity_data.get("field_uuid"))

Source code in kalbio/records.py

def get_activity_data(self, activity_id: ActivityIdentifier) -> dict:
    """Retrieves activity data for a specific activity ID.

    Args:
        activity_id: The identifier of the activity.

            Any type of ActivityIdentifier will be accepted and resolved.

    Returns:
        A dictionary mapping field IDs to their corresponding values for the given activity.
        Only fields with non-None values are included.

    Example:
        ```python
        activity_data = record.get_activity_data(activity_id="activity_uuid")
        print(activity_data.get("field_uuid"))
        ```
    """
    activity_uuid = self._client.activities._resolve_activity_id(activity_id)

    data = {}
    for field_id in self.record_values.keys():
        result = self.get_value_content(field_id, activity_uuid)
        if result is not None:
            data[field_id] = result

    return data

update_field

update_field(field_id: EntityFieldIdentifier, value: Any, activity_id: ActivityIdentifier | None) -> RecordValue | None

Updates a specific field of the record with the given value.

Parameters:

Name	Type	Description	Default
field_id	EntityFieldIdentifier	The ID of the field to update.	required
value	Any	The new value to set for the field.	required
activity_id	ActivityIdentifier | None	The ID of the activity associated with the update, or None if not an activity value	required

Returns:

Type	Description
RecordValue | None	The updated record value if the operation is successful, otherwise None.

Example

updated_value = record.update_field(
    field_id="field_uuid",
    value="Updated value",
    activity_id="activity_uuid",
)
print(updated_value.content if updated_value else None)

Source code in kalbio/records.py

def update_field(
    self,
    field_id: EntityFieldIdentifier,
    value: Any,
    activity_id: ActivityIdentifier | None,
) -> RecordValue | None:
    """Updates a specific field of the record with the given value.

    Args:
        field_id: The ID of the field to update.
        value: The new value to set for the field.
        activity_id: The ID of the activity associated with the update, or None if not an activity value

    Returns:
        The updated record value if the operation is successful, otherwise None.

    Example:
        ```python
        updated_value = record.update_field(
            field_id="field_uuid",
            value="Updated value",
            activity_id="activity_uuid",
        )
        print(updated_value.content if updated_value else None)
        ```
    """
    try:
        field_uuid = self._client.entity_fields._resolve_data_field_id(field_id)
        activity_uuid = self._client.activities._resolve_activity_id(activity_id)

        body = {
            "field_id": field_uuid,
            "content": value,
            "operation_id": activity_uuid,
        }

        resp = self._client._post("/records/" + self.id + "/values", body)
        self.refetch()

        if resp is None or len(resp) == 0:
            return None

        return RecordValue.model_validate(resp.get("resource"))
    except Exception as e:
        _logger.error(f"Error updating the field: {e}")
        return None

update_field_file

update_field_file(field_id: EntityFieldIdentifier, file_name: str, file_data: BinaryIO, file_type: str, activity_id: ActivityIdentifier | None = None) -> RecordValue | None

Update a record value with a file.

Parameters:

Name	Type	Description	Default
field_id	EntityFieldIdentifier	The ID of the field to update.	required
file_name	str	The name of the file to upload.	required
file_data	BinaryIO	The binary data of the file.	required
file_type	str	The MIME type of the file.	required
activity_id	ActivityIdentifier | None	The ID of the activity, if applicable. Defaults to None.	None

Returns:

Type	Description
RecordValue | None	The updated record value if the operation is successful, otherwise None.

Example

with open("report.pdf", "rb") as file_data:
    uploaded_value = record.update_field_file(
        field_id="file_field_uuid",
        file_name="report.pdf",
        file_data=file_data,
        file_type="application/pdf",
    )

Source code in kalbio/records.py

def update_field_file(
    self,
    field_id: EntityFieldIdentifier,
    file_name: str,
    file_data: BinaryIO,
    file_type: str,
    activity_id: Optional[ActivityIdentifier] = None,
) -> RecordValue | None:
    """Update a record value with a file.

    Args:
        field_id: The ID of the field to update.
        file_name: The name of the file to upload.
        file_data: The binary data of the file.
        file_type: The MIME type of the file.
        activity_id: The ID of the activity, if applicable. Defaults to None.

    Returns:
        The updated record value if the operation is successful, otherwise None.

    Example:
        ```python
        with open("report.pdf", "rb") as file_data:
            uploaded_value = record.update_field_file(
                field_id="file_field_uuid",
                file_name="report.pdf",
                file_data=file_data,
                file_type="application/pdf",
            )
        ```
    """
    try:
        field_uuid = self._client.entity_fields._resolve_data_field_id(field_id)
        activity_uuid = self._client.activities._resolve_activity_id(activity_id)

        body = {
            "field_id": field_uuid,
        }

        if activity_uuid:
            body["operation_id"] = activity_uuid

        resp = self._client._post_file(
            "/records/" + self.id + "/values/file",
            (file_name, file_data, file_type),
            body,
        )
        self.refetch()

        if resp is None or len(resp) == 0:
            return None

        return RecordValue.model_validate(resp.get("resource"))
    except Exception as e:
        _logger.error(f"Error uploading file to field: {e}")
        return None

get_values

get_values() -> list[RecordValue]

Retrieve all values associated with this record.

Makes a GET request to fetch the values for the current record using its ID. If the request is successful, returns the list of record values. If the response is None or an error occurs during the request, returns an empty list.

Returns:

Type	Description
list[RecordValue]	A list of RecordValue objects associated with this record. Returns an empty list if no values exist.

Note

If an exception occurs during the API request, it logs the error and returns an empty list.

Example

values = record.get_values()
print([value.content for value in values])

Source code in kalbio/records.py

def get_values(self) -> List[RecordValue]:
    """Retrieve all values associated with this record.

    Makes a GET request to fetch the values for the current record using its ID.
    If the request is successful, returns the list of record values. If the response
    is None or an error occurs during the request, returns an empty list.

    Returns:
        A list of RecordValue objects associated with this record. Returns an empty list if no values exist.

    Note:
        If an exception occurs during the API request, it logs the error and returns an empty list.

    Example:
        ```python
        values = record.get_values()
        print([value.content for value in values])
        ```
    """
    try:
        resp = self._client._get("/records/" + self.id + "/values")
        if resp is None:
            return []
        return TypeAdapter(List[RecordValue]).validate_python(resp)
    except Exception as e:
        _logger.error(f"Error fetching values for this record: {e}")
        return []

refetch

refetch()

Refreshes all the data of the current record instance.

The record is also removed from all local caches of its associated client.

Automatically called by mutating methods of this record, but can also be called manually.

Example

record.refetch()
refreshed_value = record.get_value_content(field_id="field_uuid")

Source code in kalbio/records.py

def refetch(self):
    """Refreshes all the data of the current record instance.

    The record is also removed from all local caches of its associated client.

    Automatically called by mutating methods of this record, but can also be called manually.

    Example:
        ```python
        record.refetch()
        refreshed_value = record.get_value_content(field_id="field_uuid")
        ```
    """

    self._client.records._clear_record_from_caches(self)

    new = self._client.records.get_record_by_id(self.id)
    for k, v in new.__dict__.items():
        setattr(self, k, v)

RecordIdentifier module-attribute

RecordIdentifier: TypeAlias = Record | str | dict[EntityFieldIdentifier, str]

Identifier type for Record

Record can be identified by:

• object instance of a Record
• uuid
• key field dictionary
  • a dict that maps EntityFieldIdentifiers to strs

SearchRecordsQuery

Bases: TypedDict

TypedDict for search records query parameters.

Attributes:

Name	Type	Description
record_set_id	str | None	The ID of the record set to search within.
program_id	str | None	The ID of the program associated with the records.
entity_slice_id	str | None	The ID of the entity slice to filter records.
operation_id	str | None	The ID of the operation to filter records.
identifier_ids	list[str] | None	List of identifier IDs to filter records.
record_set_filters	list[str] | None	List of filters to apply on record sets.
view_field_filters	list[ViewFieldFilter] | None	List of filters to apply on view fields.
view_field_sorts	list[ViewFieldSort] | None	List of sorting criteria for view fields.
entity_field_filters	list[FieldFilter] | None	List of filters to apply on entity fields.
entity_field_sorts	list[FieldSort] | None	List of sorting criteria for entity fields.
search_text	str | None	Text string to search for within records.
limit	int | None	Maximum number of records to return in the search results.

Example

from kalbio.records import SearchRecordsQuery, FilterRuleTypeEnum

query: SearchRecordsQuery = {
    "entity_slice_id": "entity_uuid",
    "search_text": "treatment",
    "entity_field_filters": [
        {
            "field_id": "status_field_uuid",
            "filter_type": FilterRuleTypeEnum.IS_EQUAL,
            "filter_prop": "Completed",
        }
    ],
    "limit": 25,
}

RecordsService

RecordsService(client: KaleidoscopeClient)

Service class for managing records in Kaleidoscope.

This service provides methods for creating, retrieving, and searching records, as well as managing record values and file uploads. It acts as an interface between the KaleidoscopeClient and Record objects.

Example

# Get a record by ID
record = client.records.get_record_by_id("record_uuid")
# Get multiple records (preserves order)
records = client.records.get_records_by_ids(["id1", "id2"])
# Search by text
matches = client.records.search_records(search_text="experiment-a")

Methods:

Name	Description
get_record_by_id	Retrieves a record by its identifier.
get_records_by_ids	Retrieves records corresponding to the provided list of record IDs.
get_or_create_record	Retrieves an existing record matching the provided key-value pairs, or creates a new one if none exists.
search_records	Searches for records using the provided query parameters.
create_record_value_file	Creates a record value for a file and uploads it to the specified record.
clear_record_caches	Clears all caches for Record objects.

Source code in kalbio/records.py

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

get_record_by_id

get_record_by_id(record_id: RecordIdentifier) -> Record | None

Retrieves a record by its identifier.

Parameters:

Name	Type	Description	Default
record_id	RecordIdentifier	The identifier of the record to retrieve. Any type of RecordIdentifier will be accepted and resolved	required

Returns:

Type	Description
Record | None	The record object if found, otherwise None.

Example

record = client.records.get_record_by_id("record_uuid")
print(record.record_identifier if record else "missing")

Source code in kalbio/records.py

def get_record_by_id(self, record_id: RecordIdentifier) -> Record | None:
    """Retrieves a record by its identifier.

    Args:
        record_id: The identifier of the record to retrieve.
            Any type of RecordIdentifier will be accepted and resolved

    Returns:
        The record object if found, otherwise None.

    Example:
        ```python
        record = client.records.get_record_by_id("record_uuid")
        print(record.record_identifier if record else "missing")
        ```
    """
    if isinstance(record_id, Record):
        return record_id

    if isinstance(record_id, str):
        return self._get_record_by_uuid(record_id)
    else:
        return self._get_record_by_key_values(record_id)

get_records_by_ids

get_records_by_ids(record_ids: list[RecordIdentifier], batch_size: int = 250) -> list[Record]

Retrieves records corresponding to the provided list of record IDs.

Parameters:

Name	Type	Description	Default
record_ids	list[RecordIdentifier]	A list of record IDs to retrieve.	required
batch_size	int	How many records retrieved with every API call. Defaults to 250.	250

Returns:

Type	Description
list[Record]	A list of Record objects corresponding to the provided IDs.

Example

records = client.records.get_records_by_ids([
    "record_uuid_1",
    "record_uuid_2",
])
print(len(records))

Source code in kalbio/records.py

def get_records_by_ids(
    self, record_ids: List[RecordIdentifier], batch_size: int = 250
) -> List[Record]:
    """Retrieves records corresponding to the provided list of record IDs.

    Args:
        record_ids: A list of record IDs to retrieve.
        batch_size: How many records retrieved with every API call. Defaults to 250.

    Returns:
        A list of Record objects corresponding to the provided IDs.

    Example:
        ```python
        records = client.records.get_records_by_ids([
            "record_uuid_1",
            "record_uuid_2",
        ])
        print(len(records))
        ```
    """
    try:
        all_records = []

        for batch in itertools.batched(record_ids, batch_size):
            all_records.extend(self._get_records_in_order(list(batch)))

        return [uuid for uuid in all_records if uuid]
    except Exception as e:
        _logger.error(f"Error fetching records {record_ids}: {e}")
        return []

get_or_create_record

get_or_create_record(key_values: dict[EntityFieldIdentifier, str]) -> Record | None

Retrieves an existing record matching the provided key-value pairs, or creates a new one if none exists.

Parameters:

Name	Type	Description	Default
key_values	dict[EntityFieldIdentifier, str]	A dictionary containing key-value pairs to identify or create the record.	required

Returns:

Type	Description
Record | None	The retrieved or newly created Record object if successful or None, if no record is found or created

Example

record = client.records.get_or_create_record({"FIELD_ID": "KEY-123"})
print(record.record_identifier if record else "not created")

Source code in kalbio/records.py

def get_or_create_record(
    self, key_values: dict[EntityFieldIdentifier, str]
) -> Record | None:
    """Retrieves an existing record matching the provided key-value pairs, or creates a new one if none exists.

    Args:
        key_values: A dictionary containing key-value pairs to identify or create the record.

    Returns:
        The retrieved or newly created Record object if successful or None, if no record is found or created

    Example:
        ```python
        record = client.records.get_or_create_record({"FIELD_ID": "KEY-123"})
        print(record.record_identifier if record else "not created")
        ```
    """
    try:
        resolved_values = self._resolve_key_values(key_values)
    except ValueError as e:
        _logger.error(f"Invalid key fields: {e}")
        return None

    record_key = frozenset(resolved_values.items())

    if record_key in self._records_key_field_map:
        return self._records_key_field_map[record_key]

    try:
        resp = self._client._post(
            "/records",
            {"key_field_to_value": resolved_values},
        )
        if resp is None or len(resp) == 0:
            return None

        return self._create_record(resp)
    except Exception as e:
        _logger.error(f"Error creating record {key_values}: {e}")
        return None

search_records

search_records(**params: Unpack[SearchRecordsQuery]) -> list[str]

Searches for records using the provided query parameters.

Parameters:

Name	Type	Description	Default
**params	Unpack[SearchRecordsQuery]	Keyword arguments representing search criteria. Non-string values will be JSON-encoded before being sent.	{}

Returns:

Type	Description
list[str]	A list of record identifiers matching the search criteria. Returns an empty list if the response is empty.

Note

If an exception occurs during the API request, it logs the error and returns an empty list.

Example

record_ids = client.records.search_records(search_text="cell line")

Source code in kalbio/records.py

def search_records(self, **params: Unpack[SearchRecordsQuery]) -> list[str]:
    """Searches for records using the provided query parameters.

    Args:
        **params: Keyword arguments representing search criteria. Non-string values will be JSON-encoded before being sent.

    Returns:
        A list of record identifiers matching the search criteria. Returns an empty list if the response is empty.

    Note:
        If an exception occurs during the API request, it logs the error and returns an empty list.

    Example:
        ```python
        record_ids = client.records.search_records(search_text="cell line")
        ```
    """
    try:
        client_params = {
            key: (value if isinstance(value, str) else json.dumps(value))
            for key, value in params.items()
        }
        resp = self._client._get("/records/search", client_params)
        if resp is None:
            return []

        return resp
    except Exception as e:
        _logger.error(f"Error searching records {params}: {e}")
        return []

create_record_value_file

create_record_value_file(record_id: RecordIdentifier, field_id: str, file_name: str, file_data: BinaryIO, file_type: str, activity_id: str | None = None) -> RecordValue | None

Creates a record value for a file and uploads it to the specified record.

Parameters:

Name	Type	Description	Default
record_id	RecordIdentifier	The identifier of the record to which the file value will be added. Any type of RecordIdentifier will be accepted and resolved.	required
field_id	str	The identifier of the field associated with the file value.	required
file_name	str	The name of the file to be uploaded.	required
file_data	BinaryIO	A binary stream representing the file data.	required
file_type	str	The MIME type of the file.	required
activity_id	str | None	An optional activity identifier.	None

Returns:

Type	Description
RecordValue | None	The created RecordValue object if successful, otherwise None.

Example

with open("results.csv", "rb") as file_data:
    value = client.records.create_record_value_file(
        record_id="record_uuid",
        field_id="file_field_uuid",
        file_name="results.csv",
        file_data=file_data,
        file_type="text/csv",
    )

Source code in kalbio/records.py

def create_record_value_file(
    self,
    record_id: RecordIdentifier,
    field_id: str,
    file_name: str,
    file_data: BinaryIO,
    file_type: str,
    activity_id: Optional[str] = None,
) -> RecordValue | None:
    """Creates a record value for a file and uploads it to the specified record.

    Args:
        record_id: The identifier of the record to which the file value will be added.

            Any type of RecordIdentifier will be accepted and resolved.
        field_id: The identifier of the field associated with the file value.
        file_name: The name of the file to be uploaded.
        file_data: A binary stream representing the file data.
        file_type: The MIME type of the file.
        activity_id: An optional activity identifier.

    Returns:
        The created RecordValue object if successful, otherwise None.

    Example:
        ```python
        with open("results.csv", "rb") as file_data:
            value = client.records.create_record_value_file(
                record_id="record_uuid",
                field_id="file_field_uuid",
                file_name="results.csv",
                file_data=file_data,
                file_type="text/csv",
            )
        ```
    """
    record_uuid = self._resolve_to_record_id(record_id)
    if record_uuid is None:
        return None

    try:
        body = {
            "field_id": field_id,
        }

        if activity_id:
            body["operation_id"] = activity_id

        resp = self._client._post_file(
            "/records/" + record_uuid + "/values/file",
            (file_name, file_data, file_type),
            body,
        )

        if resp is None or len(resp) == 0:
            return None

        return RecordValue.model_validate(resp.get("resource"))
    except Exception as e:
        _logger.error(f"Error uploading file to record field: {e}")
        return None

clear_record_caches

clear_record_caches()

Clears all caches for Record objects.

Call whenever caches may be stale.

Note that all methods of Record automaticaly update the caches. This is to be called if you would like your program to refetch the latest data from the API.

Example

client.records.clear_record_caches()

Source code in kalbio/records.py

def clear_record_caches(self):
    """Clears all caches for Record objects.

    Call whenever caches may be stale.

    Note that all methods of Record automaticaly update the caches.
    This is to be called if you would like your program to refetch the latest data from the API.

    Example:
        ```python
        client.records.clear_record_caches()
        ```
    """
    self._records_uuid_map.clear()
    self._records_key_field_map.clear()
    self._client.activities.get_activities_with_record.cache_clear()