Activities

activities

Activities Module for Kaleidoscope API Client.

This module provides functionality for managing activities (tasks, experiments, projects, stages, milestones, and design cycles) within the Kaleidoscope platform. It includes models for activities, activity definitions, and properties, as well as service classes for performing CRUD operations and managing activity workflows.

The module manages:

• Activity creation, updates, and status transitions
• Activity definitions
• Properties
• Records of activities
• User and group assignments
• Labels of activities
• Related programs
• Parent-child activity relationships
• Activity dependencies and scheduling

Classes and types

ActivityStatusEnum: Enumeration of possible activity statuses used across activity workflows. ActivityType: Type alias for supported activity categories (task, experiment, project, stage, milestone, cycle). Property: Model representing a property (field) attached to entities, with update and file upload helpers. ActivityDefinition: Template/definition for activities (templates for programs, users, groups, labels, and properties). Activity: Core activity model (task/experiment/project) with cached relations, record accessors, and update helpers. ActivitiesService: Service class exposing CRUD and retrieval operations for activities and activity definitions. ActivityIdentifier: Identifier union for activities (instance, title, or UUID). DefinitionIdentifier: Identifier union for activity definitions (instance, title, or UUID).

Example

# Create a new activity
activity = client.activities.create_activity(
    title="Synthesis Experiment",
    activity_type="experiment",
    program_ids=["program-uuid", ...],
    assigned_user_ids=["user-uuid", ...]
)

# Update activity status
activity.update(status=ActivityStatusEnum.IN_PROGRESS)

# Add records to activity
activity.add_records(["record-uuid"])

# Get activity data
record_data = activity.get_record_data()

Note

This module uses Pydantic for data validation and serialization. All datetime objects are timezone-aware and follow ISO 8601 format.

ActivityStatusEnum

Bases: str, Enum

Enumeration of possible activity status values.

This enum defines all possible states that an activity can be in during its lifecycle, including general workflow states, review states, and domain-specific states for design, synthesis, testing, and compound selection processes.

Attributes:

Name	Type	Description
REQUESTED	str	Activity has been requested but not yet started.
TODO	str	Activity is queued to be worked on.
IN_PROGRESS	str	Activity is currently being worked on.
NEEDS_REVIEW	str	Activity requires review.
BLOCKED	str	Activity is blocked by dependencies or issues.
PAUSED	str	Activity has been temporarily paused.
CANCELLED	str	Activity has been cancelled.
IN_REVIEW	str	Activity is currently under review.
LOCKED	str	Activity is locked from modifications.
TO_REVIEW	str	Activity is ready to be reviewed.
UPLOAD_COMPLETE	str	Upload process for the activity is complete.
NEW	str	Newly created activity.
IN_DESIGN	str	Activity is in the design phase.
READY_FOR_MAKE	str	Activity is ready for manufacturing/creation.
IN_SYNTHESIS	str	Activity is in the synthesis phase.
IN_TEST	str	Activity is in the testing phase.
IN_ANALYSIS	str	Activity is in the analysis phase.
PARKED	str	Activity has been parked for later consideration.
COMPLETE	str	Activity has been completed.
IDEATION	str	Activity is in the ideation phase.
TWO_D_SELECTION	str	Activity is in 2D selection phase.
COMPUTATION	str	Activity is in the computation phase.
COMPOUND_SELECTION	str	Activity is in the compound selection phase.
SELECTED	str	Activity or compound has been selected.
QUEUE_FOR_SYNTHESIS	str	Activity is queued for synthesis.
DATA_REVIEW	str	Activity is in the data review phase.
DONE	str	Activity is done.

Example

from kalbio.activities import ActivityStatusEnum

status = ActivityStatusEnum.IN_PROGRESS
print(status.value)

CompletedBehaviorEnum

Bases: str, Enum

How completed activities are surfaced in ActivitiesService.search_activities.

Attributes:

Name	Type	Description
SHOW_ALL		Include every matching activity regardless of completion status.
HIDE_COMPLETED_TREES		Hide activities whose entire ancestor chain is complete.
HIDE_ALL		Hide every completed activity.

ActivityType module-attribute

ActivityType: TypeAlias = Literal['task'] | Literal['experiment'] | Literal['project'] | Literal['stage'] | Literal['milestone'] | Literal['cycle']

Type alias representing the valid types of activities in the system.

This type defines the allowed string values for the activity_type field in Activity and ActivityDefinition models.

ACTIVITY_TYPE_TO_LABEL module-attribute

ACTIVITY_TYPE_TO_LABEL: dict[ActivityType, str] = {'task': 'Task', 'experiment': 'Experiment', 'project': 'Project', 'stage': 'Stage', 'milestone': 'Milestone', 'cycle': 'Design cycle'}

Dictionary mapping activity type keys to their human-readable labels.

This mapping is used to convert the internal activity_type identifiers into display-friendly strings for UI and reporting purposes.

QueueContentLayoutMapping

Bases: TypedDict

A view-to-content-layouts mapping used by set_queuing_behavior.

Attributes:

Name	Type	Description
view_id	str	UUID of the record view.
content_layout_ids	list[str]	UUIDs of the content layouts (placements of the view within an activity) that should receive queued records for this view.

ContentLayoutComponentTypeEnum

Bases: str, Enum

Component types for items in an activity definition's content layout.

ContentLayoutItem

Bases: _KaleidoscopeBaseModel

One component placement within an activity definition's content layout.

Each item represents a placement of a component (a note section, sheet, record view, etc.) within an activity. The id is what server-side "advanced settings" calls a content_layout_id. Exactly one of the typed identifier fields (note_section_id, sheet_id, platemap_id, record_view_id, drc_config_id) is populated based on component_type.

Attributes:

Name	Type	Description
id	str	UUID of this content layout item (the content_layout_id).
component_type	ContentLayoutComponentTypeEnum	Which kind of component this placement holds.
position_index	int	Display order within the activity layout.
note_section_id	str | None	UUID of the note section, if component_type is note_section.
sheet_id	str | None	UUID of the sheet, if component_type is sheet.
platemap_id	str | None	UUID of the platemap, if component_type is platemap.
record_view_id	str | None	UUID of the record view, if component_type is result_table or lookup_table.
drc_config_id	str | None	UUID of the DRC config, if component_type is drc_chart.

Property

Bases: _KaleidoscopeBaseModel

Represents a property in the Kaleidoscope system.

A Property is a data field associated with an entity that contains a value of a specific type. It includes metadata about when and by whom it was created/updated, and provides methods to update its content.

Attributes:

Name	Type	Description
id	str	UUID of the property.
property_field_id	str	UUID to the property field that defines this property's schema.
content	Any	The actual value/content stored in this property.
created_at	datetime	Timestamp when the property was created.
last_updated_by	str	UUID of the user who last updated this property.
created_by	str	UUID of the user who created this property.
property_name	str	Human-readable name of the property.
field_type	DataFieldTypeEnum	The data type of this property's content.

Example

from kalbio.activities import Property

prop = Property(
    id="prop_uuid",
    property_field_id="field_uuid",
    content="In progress",
    created_at=datetime.utcnow(),
    last_updated_by="user_uuid",
    created_by="user_uuid",
    property_name="Status",
    field_type=DataFieldTypeEnum.TEXT,
)
print(prop.property_name, prop.content)

Methods:

Name	Description
update_property	Update the property with a new value.
update_property_file	Update a property by uploading a file.

update_property

update_property(property_value: Any) -> None

Update the property with a new value.

Parameters:

Name	Type	Description	Default
property_value	Any	The new value to set for the property.	required

Raises:

Type	Description
KalbioAPIError	If the API request fails.

Example

prop.update_property("Reviewed")

Source code in kalbio/activities.py

def update_property(self, property_value: Any) -> None:
    """Update the property with a new value.

    Args:
        property_value: The new value to set for the property.

    Raises:
        KalbioAPIError: If the API request fails.

    Example:
        ```python
        prop.update_property("Reviewed")
        ```
    """
    resp = self._client._put("/properties/" + self.id, {"content": property_value})
    if resp:
        for key, value in resp.items():
            if hasattr(self, key):
                setattr(self, key, value)

update_property_file

update_property_file(file_name: str, file_data: BinaryIO, file_type: str) -> dict | None

Update a property by uploading a file.

Parameters:

Name	Type	Description	Default
file_name	str	The name of the file to be updated.	required
file_data	BinaryIO	The binary data of the file to be updated.	required
file_type	str	The MIME type of the file to be updated.	required

Returns:

Type	Description
dict | None	A dict of response JSON data (contains reference to the uploaded file), or None if the server returned an empty body.

Raises:

Type	Description
KalbioAPIError	If the API request fails.

Example

with open("report.pdf", "rb") as file_data:
    upload_info = prop.update_property_file(
        file_name="report.pdf",
        file_data=file_data,
        file_type="application/pdf",
    )

Source code in kalbio/activities.py

def update_property_file(
    self,
    file_name: str,
    file_data: BinaryIO,
    file_type: str,
) -> dict | None:
    """Update a property by uploading a file.

    Args:
        file_name: The name of the file to be updated.
        file_data: The binary data of the file to be updated.
        file_type: The MIME type of the file to be updated.

    Returns:
        A dict of response JSON data (contains reference to the
            uploaded file), or None if the server returned an empty body.

    Raises:
        KalbioAPIError: If the API request fails.

    Example:
        ```python
        with open("report.pdf", "rb") as file_data:
            upload_info = prop.update_property_file(
                file_name="report.pdf",
                file_data=file_data,
                file_type="application/pdf",
            )
        ```
    """
    resp = self._client._post_file(
        "/properties/" + self.id + "/file",
        (file_name, file_data, file_type),
    )
    if resp is None or len(resp) == 0:
        return None

    return resp

ActivityDefinition

Bases: _KaleidoscopeBaseModel

Represents the definition of an activity in the Kaleidoscope system.

An ActivityDefinition contains a template for the metadata about a task or activity, including associated programs, users, groups, labels, and properties.

Attributes:

Name	Type	Description
id	str	UUID of the Activity Definition.
program_ids	list[str]	List of program UUIDs associated with this activity.
title	str	The title of the activity.
activity_type	ActivityType	The type/category of the activity.
status	ActivityStatusEnum | None	The current status of the activity. Defaults to None if not specified.
assigned_user_ids	list[str]	List of user IDs assigned to this activity.
assigned_group_ids	list[str]	List of group IDs assigned to this activity.
label_ids	list[str]	List of label identifiers associated with this activity.
properties	list[Property]	List of properties that define additional characteristics of the activity.
external_id	str | None	The id of the activity definition if it was imported from an external source
registration_property_field_id	str | None	UUID of the file property field whose uploads trigger external registration. Only set on operation-type definitions (experiment/cycle); None otherwise.
registration_record_view_id	str | None	UUID of the record view (table) whose records are sent for registration.
registration_content_layout_id	str | None	UUID of the content layout corresponding to registration_record_view_id.
registration_status_field_id	str | None	UUID of the status field used to track registration progress.
registration_result_record_view_id	str | None	UUID of an optional results table for registration outputs.
registration_result_content_layout_id	str | None	UUID of the content layout corresponding to registration_result_record_view_id.
view_ids_to_add_to_when_record_attached	list[str]	UUIDs of record views that records are automatically added to when queued by an activity of this type.
queue_content_layout_ids	list[str]	UUIDs of the content layouts within the activity that receive queued records.
content_layout	list[ContentLayoutItem]	Component placements (note sections, sheets, record views, etc.) inside this definition. Each item's id is the content_layout_id used by set_queuing_behavior and set_registration_settings.

Example

definition = client.activities.get_definition_by_id("definition_uuid")
if definition:
    print(definition.title, definition.activity_type)

Methods:

Name	Description
update	Update this activity definition.
set_registration_settings	Configure external registration settings on this activity definition.
set_queuing_behavior	Configure queuing behavior on this activity definition.
content_layout_ids_for_view	Find content_layout item UUIDs that map to a given record view.
queue_to_views	Auto-add records to these views when queued by activities of this type.
configure_registration	Configure external registration on this activity definition.

record_views property

record_views: list[RecordView]

Regular record views attached to activities of this definition.

Returns the live record views (tables) whose operation_definition_ids contain this definition's id. Use this to discover view UUIDs for advanced settings without needing an existing activity instance.

Templates linked to this definition are NOT included — use templates instead.

Returns:

Type	Description
list[RecordView]	List of regular RecordView objects associated with this
list[RecordView]	definition. Empty if no views are attached.

Example

for view in defn.record_views:
    print(view.id, view.view_name)

templates property

templates: list['ResultTableTemplate']

Result table templates linked to this activity definition.

Returns templates whose operation_definition_ids contain this definition's id.

Returns:

Type	Description
list['ResultTableTemplate']	List of ResultTableTemplate objects linked to this definition.
list['ResultTableTemplate']	Empty if none are linked.

Example

for template in defn.templates:
    print(template.id, template.template_name or template.view_name)

activities cached property

activities: list[Activity]

Get the activities for this activity definition.

Returns:

Type	Description
list[Activity]	The activities associated with this activity definition.

Note

This is a cached property.

Example

definition = client.activities.get_definition_by_id("definition_uuid")
related = definition.activities if definition else []

update

update(**kwargs: Any) -> None

Update this activity definition.

Only the kwargs you pass are sent to the server — omitted fields are left unchanged.

For advanced/registration settings prefer the typed helpers set_registration_settings and set_queuing_behavior, which guard against accidentally clearing nullable fields.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Fields to update. Commonly used fields include: title (str): New title. is_archived (bool): Archive/unarchive. status (str): New status. description (Any): Description JSON. external_id (str): External identifier. propagate_to_instances (bool): Whether to apply the change to existing activity instances created from this definition.	{}

Example

defn.update(title="Renamed assay", propagate_to_instances=True)

Source code in kalbio/activities.py

def update(self, **kwargs: Any) -> None:
    """Update this activity definition.

    Only the kwargs you pass are sent to the server — omitted fields
    are left unchanged.

    For advanced/registration settings prefer the typed helpers
    `set_registration_settings` and `set_queuing_behavior`, which
    guard against accidentally clearing nullable fields.

    Args:
        **kwargs: Fields to update. Commonly used fields include:

            * ``title`` (str): New title.
            * ``is_archived`` (bool): Archive/unarchive.
            * ``status`` (str): New status.
            * ``description`` (Any): Description JSON.
            * ``external_id`` (str): External identifier.
            * ``propagate_to_instances`` (bool): Whether to apply
              the change to existing activity instances created from
              this definition.

    Example:
        ```python
        defn.update(title="Renamed assay", propagate_to_instances=True)
        ```
    """
    if not kwargs:
        return None
    try:
        resp = self._client._put(f"/activity_definitions/{self.id}", kwargs)
        if resp:
            # Validate the response through the model so nested fields
            # (e.g. content_layout) come back as typed objects, not dicts.
            updated = self.__class__.model_validate(resp)
            updated._set_client(self._client)
            for field_name in self.__class__.model_fields.keys():
                setattr(self, field_name, getattr(updated, field_name))
        self._client.activities._clear_definition_caches()
    except Exception as e:
        _logger.error(f"Error updating activity definition {self.id}: {e}")
        return None

set_registration_settings

set_registration_settings(*, property_field_id: str | None | _UnsetType = UNSET, record_view_id: str | None | _UnsetType = UNSET, content_layout_id: str | None | _UnsetType = UNSET, status_field_id: str | None | _UnsetType = UNSET, result_record_view_id: str | None | _UnsetType = UNSET, result_content_layout_id: str | None | _UnsetType = UNSET) -> None

Configure external registration settings on this activity definition.

Registration sends a registration_submitted webhook to a configured external service when team members register files from activities of this type.

Only the arguments you pass are sent to the server. To clear a nullable field, pass None explicitly. Omitted arguments default to UNSET and are left unchanged.

Changes take effect immediately for all activities of this type — registration settings live on the definition itself and are read by instances at runtime.

Parameters:

Name	Type	Description	Default
property_field_id	str | None | _UnsetType	UUID of the file property field whose uploads should trigger registration. Pass None to clear.	UNSET
record_view_id	str | None | _UnsetType	UUID of the record view (table) whose records will be sent for registration. Pass None to clear.	UNSET
content_layout_id	str | None | _UnsetType	UUID of the content layout corresponding to record_view_id. Pass None to clear.	UNSET
status_field_id	str | None | _UnsetType	UUID of the status field used to track registration progress. Pass None to clear.	UNSET
result_record_view_id	str | None | _UnsetType	UUID of an optional results table for registration outputs. Pass None to clear.	UNSET
result_content_layout_id	str | None | _UnsetType	UUID of the content layout corresponding to result_record_view_id. Pass None to clear.	UNSET

Example

defn = client.activities.get_activity_definition_by_external_id(
    "my-assay-v2"
)
defn.set_registration_settings(
    property_field_id="file-field-uuid",
    record_view_id="view-uuid",
    content_layout_id="layout-uuid",
    status_field_id="status-uuid",
)

Source code in kalbio/activities.py

def set_registration_settings(
    self,
    *,
    property_field_id: Union[Optional[str], _UnsetType] = UNSET,
    record_view_id: Union[Optional[str], _UnsetType] = UNSET,
    content_layout_id: Union[Optional[str], _UnsetType] = UNSET,
    status_field_id: Union[Optional[str], _UnsetType] = UNSET,
    result_record_view_id: Union[Optional[str], _UnsetType] = UNSET,
    result_content_layout_id: Union[Optional[str], _UnsetType] = UNSET,
) -> None:
    """Configure external registration settings on this activity definition.

    Registration sends a ``registration_submitted`` webhook to a
    configured external service when team members register files
    from activities of this type.

    Only the arguments you pass are sent to the server. To clear a
    nullable field, pass ``None`` explicitly. Omitted arguments
    default to ``UNSET`` and are left unchanged.

    Changes take effect immediately for all activities of this type —
    registration settings live on the definition itself and are read
    by instances at runtime.

    Args:
        property_field_id: UUID of the file property field whose
            uploads should trigger registration. Pass None to clear.
        record_view_id: UUID of the record view (table) whose
            records will be sent for registration. Pass None to clear.
        content_layout_id: UUID of the content layout corresponding
            to ``record_view_id``. Pass None to clear.
        status_field_id: UUID of the status field used to track
            registration progress. Pass None to clear.
        result_record_view_id: UUID of an optional results table
            for registration outputs. Pass None to clear.
        result_content_layout_id: UUID of the content layout
            corresponding to ``result_record_view_id``. Pass None to clear.

    Example:
        ```python
        defn = client.activities.get_activity_definition_by_external_id(
            "my-assay-v2"
        )
        defn.set_registration_settings(
            property_field_id="file-field-uuid",
            record_view_id="view-uuid",
            content_layout_id="layout-uuid",
            status_field_id="status-uuid",
        )
        ```
    """
    body: dict = {}
    if not isinstance(property_field_id, _UnsetType):
        body["registration_property_field_id"] = property_field_id
    if not isinstance(record_view_id, _UnsetType):
        body["registration_record_view_id"] = record_view_id
    if not isinstance(content_layout_id, _UnsetType):
        body["registration_content_layout_id"] = content_layout_id
    if not isinstance(status_field_id, _UnsetType):
        body["registration_status_field_id"] = status_field_id
    if not isinstance(result_record_view_id, _UnsetType):
        body["registration_result_record_view_id"] = result_record_view_id
    if not isinstance(result_content_layout_id, _UnsetType):
        body["registration_result_content_layout_id"] = result_content_layout_id

    if not body:
        return None

    self.update(**body)

set_queuing_behavior

set_queuing_behavior(*, add_view_ids: list[str] | _UnsetType = UNSET, remove_view_ids: list[str] | _UnsetType = UNSET, queue_content_layout_ids: list[QueueContentLayoutMapping] | _UnsetType = UNSET) -> None

Configure queuing behavior on this activity definition.

Controls which tables records are automatically added to when they are queued by an activity of this type.

Only the arguments you pass are sent to the server. Omitted arguments default to UNSET and are left unchanged.

Changes take effect immediately for all activities of this type.

Parameters:

Name	Type	Description	Default
add_view_ids	list[str] | _UnsetType	Record view UUIDs to start adding records to when records are attached to activities of this type.	UNSET
remove_view_ids	list[str] | _UnsetType	Record view UUIDs to stop adding records to.	UNSET
queue_content_layout_ids	list[QueueContentLayoutMapping] | _UnsetType	For each view, the content layouts within the activity that should receive queued records. Each entry is a dict of the form {"view_id": str, "content_layout_ids": List[str]}. Passing this replaces the existing mapping wholesale.	UNSET

Example

defn.set_queuing_behavior(
    add_view_ids=["view-uuid-1"],
    queue_content_layout_ids=[
        {
            "view_id": "view-uuid-1",
            "content_layout_ids": ["layout-uuid"],
        }
    ],
)

Source code in kalbio/activities.py

def set_queuing_behavior(
    self,
    *,
    add_view_ids: Union[List[str], _UnsetType] = UNSET,
    remove_view_ids: Union[List[str], _UnsetType] = UNSET,
    queue_content_layout_ids: Union[
        List[QueueContentLayoutMapping], _UnsetType
    ] = UNSET,
) -> None:
    """Configure queuing behavior on this activity definition.

    Controls which tables records are automatically added to when
    they are queued by an activity of this type.

    Only the arguments you pass are sent to the server. Omitted
    arguments default to ``UNSET`` and are left unchanged.

    Changes take effect immediately for all activities of this type.

    Args:
        add_view_ids: Record view UUIDs to start adding records to
            when records are attached to activities of this type.
        remove_view_ids: Record view UUIDs to stop adding records to.
        queue_content_layout_ids: For each view, the content layouts
            within the activity that should receive queued records.
            Each entry is a dict of the form
            ``{"view_id": str, "content_layout_ids": List[str]}``.
            Passing this replaces the existing mapping wholesale.

    Example:
        ```python
        defn.set_queuing_behavior(
            add_view_ids=["view-uuid-1"],
            queue_content_layout_ids=[
                {
                    "view_id": "view-uuid-1",
                    "content_layout_ids": ["layout-uuid"],
                }
            ],
        )
        ```
    """
    body: dict = {}
    if not isinstance(add_view_ids, _UnsetType):
        body["add_view_ids_to_add_to_when_record_attached"] = add_view_ids
    if not isinstance(remove_view_ids, _UnsetType):
        body["remove_view_ids_to_add_to_when_record_attached"] = remove_view_ids
    if not isinstance(queue_content_layout_ids, _UnsetType):
        body["set_queue_content_layout_ids"] = queue_content_layout_ids

    if not body:
        return None

    self.update(**body)

content_layout_ids_for_view

content_layout_ids_for_view(view_id: str) -> list[str]

Find content_layout item UUIDs that map to a given record view.

Use this to discover the content_layout_ids needed by set_queuing_behavior and set_registration_settings.

Parameters:

Name	Type	Description	Default
view_id	str	UUID of the record view to look up.	required

Returns:

Type	Description
list[str]	UUIDs of content layout items in this definition whose
list[str]	record_view_id matches view_id. Empty if no items match.

Example

layout_ids = defn.content_layout_ids_for_view(view.id)
defn.set_queuing_behavior(
    queue_content_layout_ids=[
        {"view_id": view.id, "content_layout_ids": layout_ids}
    ],
)

Source code in kalbio/activities.py

def content_layout_ids_for_view(self, view_id: str) -> List[str]:
    """Find content_layout item UUIDs that map to a given record view.

    Use this to discover the `content_layout_ids` needed by
    `set_queuing_behavior` and `set_registration_settings`.

    Args:
        view_id: UUID of the record view to look up.

    Returns:
        UUIDs of content layout items in this definition whose
        ``record_view_id`` matches ``view_id``. Empty if no items match.

    Example:
        ```python
        layout_ids = defn.content_layout_ids_for_view(view.id)
        defn.set_queuing_behavior(
            queue_content_layout_ids=[
                {"view_id": view.id, "content_layout_ids": layout_ids}
            ],
        )
        ```
    """
    return [
        item.id for item in self.content_layout if item.record_view_id == view_id
    ]

queue_to_views

queue_to_views(views: list[RecordView | ContentLayoutItem]) -> None

Auto-add records to these views when queued by activities of this type.

Sugar layer over set_queuing_behavior. For each entry, derives the underlying content_layout_id from this definition's content_layout.

Changes take effect immediately for all activities of this type.

Parameters:

Name	Type	Description	Default
views	list[RecordView | ContentLayoutItem]	Views to queue records into. Each entry can be either: A RecordView — auto-derives its placement in this definition. Raises ValueError if the view has zero or multiple placements. A ContentLayoutItem — targets that specific placement. Use this for views that are placed more than once.	required

Example

# Common case: just pass views
defn.queue_to_views([results_view, qc_view])

# Multi-placement case: pick the specific placement
top_placement = next(
    item for item in defn.content_layout
    if item.record_view_id == results_view.id
    and item.position_index == 0
)
defn.queue_to_views([top_placement])

Source code in kalbio/activities.py

def queue_to_views(
    self,
    views: List[Union[RecordView, ContentLayoutItem]],
) -> None:
    """Auto-add records to these views when queued by activities of this type.

    Sugar layer over `set_queuing_behavior`. For each entry, derives
    the underlying `content_layout_id` from this definition's
    `content_layout`.

    Changes take effect immediately for all activities of this type.

    Args:
        views: Views to queue records into. Each entry can be either:

            * A ``RecordView`` — auto-derives its placement in this
              definition. Raises ``ValueError`` if the view has zero or
              multiple placements.
            * A ``ContentLayoutItem`` — targets that specific placement.
              Use this for views that are placed more than once.

    Example:
        ```python
        # Common case: just pass views
        defn.queue_to_views([results_view, qc_view])

        # Multi-placement case: pick the specific placement
        top_placement = next(
            item for item in defn.content_layout
            if item.record_view_id == results_view.id
            and item.position_index == 0
        )
        defn.queue_to_views([top_placement])
        ```
    """
    layouts_by_view: Dict[str, List[str]] = {}
    for target in views:
        view_id, layout_id = self._resolve_placement(target)
        layouts_by_view.setdefault(view_id, []).append(layout_id)

    self.set_queuing_behavior(
        add_view_ids=list(layouts_by_view.keys()),
        queue_content_layout_ids=[
            {"view_id": vid, "content_layout_ids": lids}
            for vid, lids in layouts_by_view.items()
        ],
    )

configure_registration

configure_registration(*, view: RecordView | ContentLayoutItem, file_property_field: Property | str, status_field: DataField | str | None | _UnsetType = UNSET, result_view: RecordView | ContentLayoutItem | None | _UnsetType = UNSET) -> None

Configure external registration on this activity definition.

Sugar layer over set_registration_settings. Derives content_layout_id (and result_content_layout_id) from the passed view(s).

Changes take effect immediately for all activities of this type.

Parameters:

Name	Type	Description	Default
view	RecordView | ContentLayoutItem	The record view whose records are sent for registration. Either a RecordView (auto-derives layout) or a ContentLayoutItem (specific placement). Required.	required
file_property_field	Property | str	The file property field whose uploads trigger registration. Either a Property (uses its property_field_id) or a property-field UUID string. Required.	required
status_field	DataField | str | None | _UnsetType	Optional status field used to track registration progress. Pass a DataField, a UUID string, None to clear, or leave unset to keep the current value.	UNSET
result_view	RecordView | ContentLayoutItem | None | _UnsetType	Optional results table for registration outputs. Same view/placement semantics as view. Pass None to clear, or leave unset to keep the current value.	UNSET

Example

defn.configure_registration(
    view=results_view,
    file_property_field=file_property,
    status_field=status_data_field,
)

Source code in kalbio/activities.py

def configure_registration(
    self,
    *,
    view: Union[RecordView, ContentLayoutItem],
    file_property_field: Union[Property, str],
    status_field: Union[DataField, str, None, _UnsetType] = UNSET,
    result_view: Union[RecordView, ContentLayoutItem, None, _UnsetType] = UNSET,
) -> None:
    """Configure external registration on this activity definition.

    Sugar layer over `set_registration_settings`. Derives
    `content_layout_id` (and `result_content_layout_id`) from the
    passed view(s).

    Changes take effect immediately for all activities of this type.

    Args:
        view: The record view whose records are sent for registration.
            Either a ``RecordView`` (auto-derives layout) or a
            ``ContentLayoutItem`` (specific placement). Required.
        file_property_field: The file property field whose uploads
            trigger registration. Either a ``Property`` (uses its
            ``property_field_id``) or a property-field UUID string.
            Required.
        status_field: Optional status field used to track registration
            progress. Pass a ``DataField``, a UUID string, ``None`` to
            clear, or leave unset to keep the current value.
        result_view: Optional results table for registration outputs.
            Same view/placement semantics as ``view``. Pass ``None`` to
            clear, or leave unset to keep the current value.

    Example:
        ```python
        defn.configure_registration(
            view=results_view,
            file_property_field=file_property,
            status_field=status_data_field,
        )
        ```
    """
    view_id, content_layout_id = self._resolve_placement(view)

    pf_id = (
        file_property_field.property_field_id
        if isinstance(file_property_field, Property)
        else file_property_field
    )

    body_kwargs: dict = {
        "property_field_id": pf_id,
        "record_view_id": view_id,
        "content_layout_id": content_layout_id,
    }

    if not isinstance(status_field, _UnsetType):
        if status_field is None or isinstance(status_field, str):
            body_kwargs["status_field_id"] = status_field
        else:
            body_kwargs["status_field_id"] = status_field.id

    if not isinstance(result_view, _UnsetType):
        if result_view is None:
            body_kwargs["result_record_view_id"] = None
            body_kwargs["result_content_layout_id"] = None
        else:
            r_view_id, r_layout_id = self._resolve_placement(result_view)
            body_kwargs["result_record_view_id"] = r_view_id
            body_kwargs["result_content_layout_id"] = r_layout_id

    self.set_registration_settings(**body_kwargs)

Comment

Bases: _KaleidoscopeBaseModel

A user comment on a Kaleidoscope resource (e.g. an activity).

Attributes:

Name	Type	Description
id	str	UUID of the comment.
workspace_id	str	UUID of the workspace this comment belongs to.
created_by	str	UUID of the user who authored the comment.
content	Any	Tiptap-compatible rich-text JSON document for the comment body.
parent_comment_id	str | None	UUID of the parent comment if this is a reply, else None.
mentioned_user_ids	list[str] | None	UUIDs of users @-mentioned in the comment, if any.
resource_type	str	Type of the resource the comment is attached to (e.g. "task").
resource_id	str	UUID of the resource the comment is attached to.
created_at	datetime	Timestamp the comment was created.
updated_at	datetime	Timestamp the comment was last updated.

ActivityEvent

Bases: _KaleidoscopeBaseModel

An entry in the audit/event log for an activity.

Returned by Activity.get_events. Each event captures one user action or system-generated change to the activity (status change, assignment, record attachment, etc.).

Attributes:

Name	Type	Description
id	str	UUID of the event.
event_type	str	String identifier of the event type.
event_type_version	int	Version number of the event type schema.
event_attrs	dict	Event-specific attributes (shape varies by event_type).
event_user_id	str	UUID of the user that triggered the event.
created_at	datetime	Timestamp the event was recorded.
resource_id	str | None	UUID of the resource the event applies to.
resource_type	str | None	Type of the resource (e.g. "task").
workspace_id	str | None	UUID of the workspace.
parent_bulk_event_id	str | None	UUID of the parent bulk event if this event was part of a bulk operation.
is_bulk	bool	Whether this event was part of a bulk operation.
request_id	str | None	Request identifier the event was emitted under, if any.
session_id	str | None	Session identifier the event was emitted under, if any.
log	str | None	Human-readable summary of the event, if available.

Activity

Bases: _KaleidoscopeBaseModel

Represents an activity (e.g. task or experiment) within the Kaleidoscope system.

An Activity is a unit of work that can be assigned to users or groups, have dependencies, and contain associated records and properties. Activities can be organized hierarchically with parent-child relationships and linked to programs.

Attributes:

Name	Type	Description
id	str	Unique identifier for the model instance.
created_at	datetime	The timestamp when the activity was created.
parent_id	str | None	The ID of the parent activity, if this is a child activity.
child_ids	list[str]	List of child activity IDs.
definition_id	str | None	The ID of the activity definition template.
program_ids	list[str]	List of program IDs this activity belongs to.
activity_type	ActivityType	The type/category of the activity.
title	str	The title of the activity.
description	Any	Detailed description of the activity.
status	ActivityStatusEnum	Current status of the activity.
assigned_user_ids	list[str]	List of user IDs assigned to this activity.
assigned_group_ids	list[str]	List of group IDs assigned to this activity.
due_date	datetime | None	The deadline for completing the activity.
start_date	datetime | None	The scheduled start date for the activity.
duration	int | None	Expected duration of the activity.
completed_at_date	datetime | None	The timestamp when the activity was completed.
dependencies	list[str]	List of activity IDs that this activity depends on.
label_ids	list[str]	List of label IDs associated with this activity.
is_draft	bool	Whether the activity is in draft status.
properties	list[Property]	List of custom properties associated with the activity.
external_id	str | None	The id of the activity if it was imported from an external source
all_record_ids	list[str]	All record IDs associated with the activity across operations.
data_table_record_mapping	dict[str, list[str]]	Per-view record ordering for this operation, keyed by record view id. The authoritative source for "which records are on this view".

Example

activity = client.activities.get_activity_by_id("activity_uuid")
if activity:
    print(activity.title, activity.status)
    first_record = activity.records[0] if activity.records else None

Methods:

Name	Description
records_on_view	Records currently on a specific view of this operation.
get_record	Retrieves the record with the given identifier if it is in the operation.
has_record	Retrieve whether a record with the given identifier is in the operation
update	Update the activity with the provided keyword arguments.
add_records	Add a list of record IDs to the activity.
get_record_data	Retrieve data from all this activity's associated records.
get_events	Retrieve the audit/event log entries for this activity.
get_comments	Retrieve comments posted on this activity.
refetch	Refreshes all the data of the current activity instance.

activity_definition cached property

activity_definition: ActivityDefinition | None

Get the activity definition for this activity.

Returns:

Type	Description
ActivityDefinition | None	The activity definition associated with this activity. If the activity has no definition, returns None.

Note

This is a cached property.

Example

definition = activity.activity_definition
print(definition.title if definition else "No template")

assigned_users cached property

assigned_users: list[WorkspaceUser]

Get the assigned users for this activity.

Returns:

Type	Description
list[WorkspaceUser]	The users assigned to this activity.

Note

This is a cached property.

assigned_groups cached property

assigned_groups: list[WorkspaceGroup]

Get the assigned groups for this activity.

Returns:

Type	Description
list[WorkspaceGroup]	The groups assigned to this activity.

Note

This is a cached property.

labels cached property

labels: list[Label]

Get the labels for this activity.

Returns:

Type	Description
list[Label]	The labels associated with this activity.

Note

This is a cached property.

Example

label_names = [label.name for label in activity.labels]

programs cached property

programs: list[Program]

Retrieve the programs associated with this activity.

Returns:

Type	Description
list[Program]	A list of Program instances fetched by their IDs.

Note

This is a cached property.

Example

program_titles = [program.title for program in activity.programs]

child_activities cached property

child_activities: list[Activity]

Retrieve the child activities associated with this activity.

Returns:

Type	Description
list[Activity]	A list of Activity objects representing the child activities.

Note

This is a cached property.

records property

records: list['Record']

Retrieve the records associated with this activity.

Returns:

Type	Description
list['Record']	A list of Record objects corresponding to the activity.

Note

This is a cached property.

record_views property

record_views: list[RecordView]

Regular record views attached to this operation.

Returns the live record views (tables) associated with this operation — i.e. the views you can target when writing values via Record.add_value(record_view_id=...) or importing via client.imports.push_data(operation_id=..., record_view_id=...).

Templates linked to this activity's definition are NOT included — access them via activity.activity_definition.templates.

Returns:

Type	Description
list[RecordView]	List of regular RecordView objects whose operation_ids contains
list[RecordView]	this activity's id. Returns an empty list if this activity is not
list[RecordView]	an operation or has no attached views.

Example

for view in activity.record_views:
    print(view.id, view.view_name)

records_on_view

records_on_view(view_id: str) -> list['Record']

Records currently on a specific view of this operation.

Reads data_table_record_mapping[view_id]. Records that fail to resolve are dropped.

Parameters:

Name	Type	Description	Default
view_id	str	The UUID of the record view.	required

Returns:

Type	Description
list['Record']	Records on the view, in the order the server stores them.

Example

for record in activity.records_on_view(view.id):
    print(record.record_identifier)

Source code in kalbio/activities.py

def records_on_view(self, view_id: str) -> List["Record"]:
    """Records currently on a specific view of this operation.

    Reads `data_table_record_mapping[view_id]`.
    Records that fail to resolve are dropped.

    Args:
        view_id: The UUID of the record view.

    Returns:
        Records on the view, in the order the server stores them.

    Example:
        ```python
        for record in activity.records_on_view(view.id):
            print(record.record_identifier)
        ```
    """
    record_ids = self.data_table_record_mapping.get(view_id, [])
    if not record_ids:
        return []
    return [r for r in self._client.records.get_records_by_ids(record_ids) if r]

get_record

get_record(identifier: RecordIdentifier) -> Record | None

Retrieves the record with the given identifier if it is in the operation.

Parameters:

Name	Type	Description	Default
identifier	RecordIdentifier	An identifier for a Record. This method will accept and resolve any type of RecordIdentifier.	required

Returns:

Type	Description
Record | None	The record if it is in the operation, otherwise None

Example

record = activity.get_record("record_uuid")

Source code in kalbio/activities.py

def get_record(self, identifier: RecordIdentifier) -> Record | None:
    """Retrieves the record with the given identifier if it is in the operation.

    Args:
        identifier: An identifier for a Record.

            This method will accept and resolve any type of RecordIdentifier.

    Returns:
        The record if it is in the operation, otherwise None

    Example:
        ```python
        record = activity.get_record("record_uuid")
        ```
    """
    idx = self._client.records._resolve_to_record_id(identifier)

    if idx is None:
        return None

    return next(
        (r for r in self.records if r.id == idx),
        None,
    )

has_record

has_record(identifier: RecordIdentifier) -> bool

Retrieve whether a record with the given identifier is in the operation

Parameters:

Name	Type	Description	Default
identifier	RecordIdentifier	An identifier for a Record. This method will accept and resolve any type of RecordIdentifier.	required

Returns:

Type	Description
bool	Whether the record is in the operation

Example

has_link = activity.has_record("record_uuid")

Source code in kalbio/activities.py

def has_record(self, identifier: RecordIdentifier) -> bool:
    """Retrieve whether a record with the given identifier is in the operation

    Args:
        identifier: An identifier for a Record.

            This method will accept and resolve any type of RecordIdentifier.

    Returns:
        Whether the record is in the operation

    Example:
        ```python
        has_link = activity.has_record("record_uuid")
        ```
    """
    return self.get_record(identifier) is not None

update

update(**kwargs: Any) -> None

Update the activity with the provided keyword arguments.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Arbitrary keyword arguments representing fields to update for the activity.	{}

Raises:

Type	Description
KalbioAPIError	If the API request fails.

Note

After calling update(), cached properties may be stale. Re-fetch the activity if needed.

Example

activity.update(status=ActivityStatusEnum.IN_PROGRESS)

Source code in kalbio/activities.py

def update(self, **kwargs: Any) -> None:
    """Update the activity with the provided keyword arguments.

    Args:
        **kwargs: Arbitrary keyword arguments representing fields to update
            for the activity.

    Raises:
        KalbioAPIError: If the API request fails.

    Note:
        After calling update(), cached properties may be stale. Re-fetch the activity if needed.

    Example:
        ```python
        activity.update(status=ActivityStatusEnum.IN_PROGRESS)
        ```
    """
    resp = self._client._put("/activities/" + self.id, kwargs)
    if resp:
        for key, value in resp.items():
            if hasattr(self, key):
                setattr(self, key, value)

add_records

add_records(record_ids: list[str]) -> None

Add a list of record IDs to the activity.

Parameters:

Name	Type	Description	Default
record_ids	list[str]	A list of record IDs to be added to the activity.	required

Raises:

Type	Description
KalbioAPIError	If the API request fails.

Example

activity.add_records(["record_uuid_1", "record_uuid_2"])

Source code in kalbio/activities.py

def add_records(self, record_ids: List[str]) -> None:
    """Add a list of record IDs to the activity.

    Args:
        record_ids: A list of record IDs to be added to the activity.

    Raises:
        KalbioAPIError: If the API request fails.

    Example:
        ```python
        activity.add_records(["record_uuid_1", "record_uuid_2"])
        ```
    """
    self._client._put(
        "/operations/" + self.id + "/records", {"record_ids": record_ids}
    )

get_record_data

get_record_data() -> list[dict]

Retrieve data from all this activity's associated records.

Returns:

Type	Description
list[dict]	A list containing the activity data for each record, obtained by calling get_activity_data with the current activity's UUID.

Example

data = activity.get_record_data()

Source code in kalbio/activities.py

def get_record_data(self) -> List[dict]:
    """Retrieve data from all this activity's associated records.

    Returns:
        A list containing the activity data for each record,
            obtained by calling get_activity_data with the current activity's UUID.

    Example:
        ```python
        data = activity.get_record_data()
        ```
    """
    data = []
    for record in self.records:
        data.append(record.get_activity_data(self.id))
    return data

get_events

get_events() -> list[ActivityEvent]

Retrieve the audit/event log entries for this activity.

Returns:

Type	Description
list[ActivityEvent]	A list of ActivityEvent records for this activity, in the order
list[ActivityEvent]	the server provides them (typically newest first).

Example

for event in activity.get_events():
    print(event.event_type, event.created_at, event.log)

Source code in kalbio/activities.py

def get_events(self) -> List[ActivityEvent]:
    """Retrieve the audit/event log entries for this activity.

    Returns:
        A list of ActivityEvent records for this activity, in the order
        the server provides them (typically newest first).

    Example:
        ```python
        for event in activity.get_events():
            print(event.event_type, event.created_at, event.log)
        ```
    """
    resp = self._client._get(f"/activities/{self.id}/events")
    if resp is None:
        return []
    return [ActivityEvent.model_validate(e) for e in resp]

get_comments

get_comments() -> list[Comment]

Retrieve comments posted on this activity.

Returns:

Type	Description
list[Comment]	A list of Comment objects attached to this activity.

Example

for comment in activity.get_comments():
    print(comment.created_by, comment.content)

Source code in kalbio/activities.py

def get_comments(self) -> List[Comment]:
    """Retrieve comments posted on this activity.

    Returns:
        A list of Comment objects attached to this activity.

    Example:
        ```python
        for comment in activity.get_comments():
            print(comment.created_by, comment.content)
        ```
    """
    resp = self._client._get(f"/activities/{self.id}/comments")
    if resp is None:
        return []
    comments = [Comment.model_validate(c) for c in resp]
    for c in comments:
        c._set_client(self._client)
    return comments

refetch

refetch()

Refreshes all the data of the current activity instance.

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

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

Example

activity.refetch()
up_to_date_records = activity.records

Source code in kalbio/activities.py

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

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

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

    Example:
        ```python
        activity.refetch()
        up_to_date_records = activity.records
        ```
    """
    self._client.activities._clear_activity_caches()

    new = self._client.activities.get_activity_by_id(self.id)

    if new is None:
        _logger.error(f"Unable to refresh Activity({self.id})")
        return None

    for k, v in new.__dict__.items():
        setattr(self, k, v)

ActivityIdentifier module-attribute

ActivityIdentifier: TypeAlias = Activity | str

Identifier class for Activity

Activities are able to be identified by:

• an object instance of an Activity
• title
• UUID

DefinitionIdentifier module-attribute

DefinitionIdentifier: TypeAlias = ActivityDefinition | str

Identifier class for ActivityDefinition

ActivityDefinitions are able to be identified by:

• an object instance of an ActivityDefinition
• title
• UUID

ActivitiesService

ActivitiesService(client: KaleidoscopeClient)

Service class for managing activities in the Kaleidoscope platform.

This service provides methods to create, retrieve, and manage activities (tasks/experiments) and their definitions within a Kaleidoscope workspace. It handles activity lifecycle operations including creation, retrieval by ID or associated records, and batch operations.

Note

Some methods use LRU caching to improve performance. Cache is cleared on errors.

Methods:

Name	Description
get_activities	Retrieve all activities in the workspace, including experiments.
get_activity_by_type	Retrieve all activities of a certain type in the workspace.
get_activity_by_id	Retrieve an activity by its identifier.
get_activities_by_ids	Fetch multiple activities by their identifiers.
get_activity_by_external_id	Retrieve an activity by its external identifier.
create_activity	Create a new activity.
get_activities_with_record	Retrieve all activities that contain a specific record.
search_activities	Search activities with server-side filtering.
get_definitions	Retrieve all available activity definitions.
get_definition_by_id	Retrieve an activity definition by ID (UUID or name)
get_definitions_by_ids	Retrieve activity definitions by their identifiers
get_activity_definition_by_external_id	Retrieve an activity definition by its external identifier.

Source code in kalbio/activities.py

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

get_activities cached

get_activities() -> list[Activity]

Retrieve all activities in the workspace, including experiments.

Returns:

Type	Description
list[Activity]	A list of Activity objects representing the activities in the workspace.

Note

This method caches its results. If an exception occurs, logs the error, clears the cache, and returns an empty list.

Example

activities = client.activities.get_activities()

Source code in kalbio/activities.py

@lru_cache
def get_activities(self) -> List[Activity]:
    """Retrieve all activities in the workspace, including experiments.

    Returns:
        A list of Activity objects representing the activities
            in the workspace.

    Note:
        This method caches its results. If an exception occurs, logs the error,
        clears the cache, and returns an empty list.

    Example:
        ```python
        activities = client.activities.get_activities()
        ```
    """
    resp = self._client._get("/activities")
    return self._create_activity_list(resp)

get_activity_by_type

get_activity_by_type(activity_type: ActivityType) -> list[Activity]

Retrieve all activities of a certain type in the workspace.

Parameters:

Name	Type	Description	Default
activity_type	ActivityType	The type of Activity to retrieve.	required

Returns:

Type	Description
list[Activity]	A list of Activity objects with the type of activity_type

Example

experiments = client.activities.get_activity_by_type("experiment")
tasks = client.activities.get_activity_by_type("task")

Source code in kalbio/activities.py

def get_activity_by_type(self, activity_type: ActivityType) -> List[Activity]:
    """Retrieve all activities of a certain type in the workspace.

    Args:
        activity_type: The type of `Activity` to retrieve.

    Returns:
        A list of Activity objects with the type of `activity_type`

    Example:
        ```python
        experiments = client.activities.get_activity_by_type("experiment")
        tasks = client.activities.get_activity_by_type("task")
        ```
    """

    return [
        act for act in self.get_activities() if act.activity_type == activity_type
    ]

get_activity_by_id

get_activity_by_id(activity_id: ActivityIdentifier) -> Activity | None

Retrieve an activity by its identifier.

Parameters:

Name	Type	Description	Default
activity_id	ActivityIdentifier	An identifier of the activity to retrieve. This method will accept and resolve any type of ActivityIdentifier.	required

Returns:

Type	Description
Activity | None	The Activity object if found, otherwise None.

Example

activity = client.activities.get_activity_by_id("activity_uuid")

Source code in kalbio/activities.py

def get_activity_by_id(self, activity_id: ActivityIdentifier) -> Activity | None:
    """Retrieve an activity by its identifier.

    Args:
        activity_id: An identifier of the activity to retrieve.

            This method will accept and resolve any type of ActivityIdentifier.

    Returns:
        The Activity object if found, otherwise None.

    Example:
        ```python
        activity = client.activities.get_activity_by_id("activity_uuid")
        ```
    """
    id_to_activity = self._get_activity_id_map()
    identifier = self._resolve_activity_id(activity_id)

    if identifier is None:
        return None

    return id_to_activity.get(identifier, None)

get_activities_by_ids

get_activities_by_ids(ids: list[ActivityIdentifier]) -> list[Activity]

Fetch multiple activities by their identifiers.

Parameters:

Name	Type	Description	Default
ids	list[ActivityIdentifier]	A list of activity identifier strings to fetch. This method will accept and resolve any type of ActivityIdentifier inside the ids.	required

Returns:

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

Note

ids that are invalid and return None are not included in the returned list of Activities

Example

selected = client.activities.get_activities_by_ids([
    "activity_uuid_1",
    "activity_uuid_2",
])

Source code in kalbio/activities.py

def get_activities_by_ids(self, ids: List[ActivityIdentifier]) -> List[Activity]:
    """Fetch multiple activities by their identifiers.

    Args:
        ids: A list of activity identifier strings to fetch.

            This method will accept and resolve any type of ActivityIdentifier inside the `ids`.

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

    Note:
        ids that are invalid and return None are not included in the returned list of Activities

    Example:
        ```python
        selected = client.activities.get_activities_by_ids([
            "activity_uuid_1",
            "activity_uuid_2",
        ])
        ```
    """
    activities = []

    for activity_id in ids:
        res = self.get_activity_by_id(activity_id)
        if res:
            activities.append(res)

    return activities

get_activity_by_external_id

get_activity_by_external_id(external_id: str) -> Activity | None

Retrieve an activity by its external identifier.

Parameters:

Name	Type	Description	Default
external_id	str	The external identifier of the activity to retrieve.	required

Returns:

Type	Description
Activity | None	The Activity object if found, otherwise None.

Example

ext_activity = client.activities.get_activity_by_external_id("jira-123")

Source code in kalbio/activities.py

def get_activity_by_external_id(self, external_id: str) -> Activity | None:
    """Retrieve an activity by its external identifier.

    Args:
        external_id: The external identifier of the activity to retrieve.

    Returns:
        The Activity object if found, otherwise None.

    Example:
        ```python
        ext_activity = client.activities.get_activity_by_external_id("jira-123")
        ```
    """
    activities = self.get_activities()
    return next(
        (a for a in activities if a.external_id == external_id),
        None,
    )

create_activity

create_activity(title: str, activity_type: ActivityType, program_ids: list[str] | None = None, activity_definition_id: DefinitionIdentifier | None = None, assigned_user_ids: list[str] | None = None, start_date: datetime | None = None, duration: int | None = None) -> Activity

Create a new activity.

Parameters:

Name	Type	Description	Default
title	str	The title/name of the activity.	required
activity_type	ActivityType	The type of activity (e.g. task, experiment, etc.).	required
program_ids	list[str] | None	List of program IDs to associate with the activity. Defaults to None.	None
activity_definition_id	DefinitionIdentifier | None	Identifier for an activity definition to create the activity with. Defaults to None. The identifier will resolve any type of DefinitionIdentifier.	None
assigned_user_ids	list[str] | None	List of user IDs to assign to the activity. Defaults to None.	None
start_date	datetime | None	Start date for the activity. Defaults to None.	None
duration	int | None	Duration in days for the activity. Defaults to None.	None

Returns:

Type	Description
Activity	The newly created activity instance.

Raises:

Type	Description
KalbioAPIError	If the API request fails.

Example

new_activity = client.activities.create_activity(
    title="Synthesis",
    activity_type="experiment",
    program_ids=["program_uuid"],
)

Source code in kalbio/activities.py

def create_activity(
    self,
    title: str,
    activity_type: ActivityType,
    program_ids: Optional[list[str]] = None,
    activity_definition_id: Optional[DefinitionIdentifier] = None,
    assigned_user_ids: Optional[List[str]] = None,
    start_date: Optional[datetime] = None,
    duration: Optional[int] = None,
) -> Activity:
    """Create a new activity.

    Args:
        title: The title/name of the activity.
        activity_type: The type of activity (e.g. task, experiment, etc.).
        program_ids: List of program IDs to associate with
            the activity. Defaults to None.
        activity_definition_id: Identifier for an activity definition to create the activity with.
            Defaults to None.

            The identifier will resolve any type of DefinitionIdentifier.
        assigned_user_ids: List of user IDs to assign to
            the activity. Defaults to None.
        start_date: Start date for the activity. Defaults to None.
        duration: Duration in days for the activity. Defaults to None.

    Returns:
        The newly created activity instance.

    Raises:
        KalbioAPIError: If the API request fails.

    Example:
        ```python
        new_activity = client.activities.create_activity(
            title="Synthesis",
            activity_type="experiment",
            program_ids=["program_uuid"],
        )
        ```
    """
    self._clear_activity_caches()

    payload = {
        "title": title,
        "activity_type": activity_type,
        "definition_id": self._resolve_definition_id(activity_definition_id),
        "program_ids": program_ids if program_ids else [],
        "assigned_user_ids": assigned_user_ids if assigned_user_ids else [],
        "start_date": start_date.isoformat() if start_date else None,
        "duration": duration,
    }
    resp = self._client._post("/activities", payload)
    return self._create_activity(resp[0])

get_activities_with_record

get_activities_with_record(record_id: RecordIdentifier) -> list[Activity]

Retrieve all activities that contain a specific record.

Parameters:

Name	Type	Description	Default
record_id	RecordIdentifier	Identifier for the record. Any type of RecordIdentifier will be accepted.	required

Returns:

Type	Description
list[Activity]	Activities that include the specified record.

Note

If an exception occurs, logs the error and returns an empty list.

Example

activities = client.activities.get_activities_with_record("record_uuid")

Source code in kalbio/activities.py

@cachetools.func.ttl_cache(maxsize=128, ttl=10)
def get_activities_with_record(self, record_id: RecordIdentifier) -> List[Activity]:
    """Retrieve all activities that contain a specific record.

    Args:
        record_id: Identifier for the record.

            Any type of RecordIdentifier will be accepted.

    Returns:
        Activities that include the specified record.

    Note:
        If an exception occurs, logs the error and returns an empty list.

    Example:
        ```python
        activities = client.activities.get_activities_with_record("record_uuid")
        ```
    """
    record_uuid = self._client.records._resolve_to_record_id(record_id)
    if record_uuid is None:
        return []

    resp = self._client._get("/records/" + record_uuid + "/operations")
    if resp is None:
        return []
    return self._create_activity_list(resp)

search_activities

search_activities(*, search_text: str | None = None, activity_types: list[ActivityType] | None = None, definition_ids: list[str] | None = None, record_ids: list[str] | None = None, statuses: list[ActivityStatusEnum] | None = None, label_ids: list[str] | None = None, assigned_user_ids: list[str] | None = None, created_by: str | None = None, parent_id: str | None | _UnsetType = UNSET, limit: int | None = None, completed_behavior: CompletedBehaviorEnum | None = None) -> list[Activity]

Search activities with server-side filtering.

Returns the matching activities, in the server's default order, capped at limit (server max: 500).

For list filters (statuses, label_ids, assigned_user_ids) the values are OR-combined: a hit on any value qualifies an activity.

Parameters:

Name	Type	Description	Default
search_text	str | None	Free-text query matched against activity title	None
activity_types	list[ActivityType] | None	Restrict to these activity types (e.g. ["task", "experiment"]).	None
definition_ids	list[str] | None	Restrict to activities created from these definition UUIDs.	None
record_ids	list[str] | None	Restrict to activities containing any of these record UUIDs.	None
statuses	list[ActivityStatusEnum] | None	Restrict to activities currently in any of these statuses.	None
label_ids	list[str] | None	Restrict to activities tagged with any of these label UUIDs.	None
assigned_user_ids	list[str] | None	Restrict to activities assigned to any of these user UUIDs.	None
created_by	str | None	Restrict to activities created by this user UUID.	None
parent_id	str | None | _UnsetType	Parent-filter behavior — UNSET (default) for no filter, None to return root activities only, or a UUID for direct children of that activity.	UNSET
limit	int | None	Maximum number of results to return (server caps at 500).	None
completed_behavior	CompletedBehaviorEnum | None	How completed activities are surfaced; defaults server-side to SHOW_ALL.	None

Returns:

Type	Description
list[Activity]	Matching activities, possibly empty.

Raises:

Type	Description
KalbioAPIError	If the API request fails.

Example

# Find in-progress experiments assigned to a specific user
results = client.activities.search_activities(
    activity_types=["experiment"],
    statuses=[ActivityStatusEnum.IN_PROGRESS],
    assigned_user_ids=["user-uuid"],
    limit=50,
)

# Find only root activities
roots = client.activities.search_activities(parent_id=None)

Source code in kalbio/activities.py

def search_activities(
    self,
    *,
    search_text: Optional[str] = None,
    activity_types: Optional[List[ActivityType]] = None,
    definition_ids: Optional[List[str]] = None,
    record_ids: Optional[List[str]] = None,
    statuses: Optional[List[ActivityStatusEnum]] = None,
    label_ids: Optional[List[str]] = None,
    assigned_user_ids: Optional[List[str]] = None,
    created_by: Optional[str] = None,
    parent_id: Union[Optional[str], _UnsetType] = UNSET,
    limit: Optional[int] = None,
    completed_behavior: Optional[CompletedBehaviorEnum] = None,
) -> List[Activity]:
    """Search activities with server-side filtering.

    Returns the matching activities, in the server's default order, capped
    at `limit` (server max: 500).

    For list filters (`statuses`, `label_ids`, `assigned_user_ids`) the values
    are OR-combined: a hit on any value qualifies an activity.

    Args:
        search_text: Free-text query matched against activity title
        activity_types: Restrict to these activity types (e.g. `["task", "experiment"]`).
        definition_ids: Restrict to activities created from these definition UUIDs.
        record_ids: Restrict to activities containing any of these record UUIDs.
        statuses: Restrict to activities currently in any of these statuses.
        label_ids: Restrict to activities tagged with any of these label UUIDs.
        assigned_user_ids: Restrict to activities assigned to any of these user UUIDs.
        created_by: Restrict to activities created by this user UUID.
        parent_id: Parent-filter behavior — `UNSET` (default) for no filter,
            `None` to return root activities only, or a UUID for direct children
            of that activity.
        limit: Maximum number of results to return (server caps at 500).
        completed_behavior: How completed activities are surfaced; defaults
            server-side to `SHOW_ALL`.

    Returns:
        Matching activities, possibly empty.

    Raises:
        KalbioAPIError: If the API request fails.

    Example:
        ```python
        # Find in-progress experiments assigned to a specific user
        results = client.activities.search_activities(
            activity_types=["experiment"],
            statuses=[ActivityStatusEnum.IN_PROGRESS],
            assigned_user_ids=["user-uuid"],
            limit=50,
        )

        # Find only root activities
        roots = client.activities.search_activities(parent_id=None)
        ```
    """
    params: Dict[str, Any] = {}

    if search_text is not None:
        params["search_text"] = search_text
    if activity_types is not None:
        params["activity_types"] = json.dumps(activity_types)
    if definition_ids is not None:
        params["definition_ids"] = json.dumps(definition_ids)
    if record_ids is not None:
        params["record_ids"] = json.dumps(record_ids)
    if statuses is not None:
        params["statuses"] = json.dumps([[s.value for s in statuses]])
    if label_ids is not None:
        params["label_ids"] = json.dumps([label_ids])
    if assigned_user_ids is not None:
        params["assigned_user_ids"] = json.dumps([assigned_user_ids])
    if created_by is not None:
        params["created_by"] = created_by
    if not isinstance(parent_id, _UnsetType):
        params["parent_id"] = "null" if parent_id is None else parent_id
    if limit is not None:
        params["limit"] = limit
    if completed_behavior is not None:
        params["completed_behavior"] = completed_behavior.value

    resp = self._client._get("/activities/search", params=params)
    if resp is None:
        return []
    return self._create_activity_list(resp)

get_definitions cached

get_definitions() -> list[ActivityDefinition]

Retrieve all available activity definitions.

Returns:

Type	Description
list[ActivityDefinition]	All activity definitions in the workspace.

Raises:

Type	Description
ValidationError	If the data could not be validated as an ActivityDefinition.

Note

This method caches its results. If an exception occurs, logs the error, clears the cache, and returns an empty list.

Example

definitions = client.activities.get_definitions()

Source code in kalbio/activities.py

@lru_cache
def get_definitions(self) -> List[ActivityDefinition]:
    """Retrieve all available activity definitions.

    Returns:
        All activity definitions in the workspace.

    Raises:
        ValidationError: If the data could not be validated as an ActivityDefinition.

    Note:
        This method caches its results. If an exception occurs, logs the error,
        clears the cache, and returns an empty list.

    Example:
        ```python
        definitions = client.activities.get_definitions()
        ```
    """
    resp = self._client._get("/activity_definitions")
    return [self._create_activity_definition(data) for data in resp]

get_definition_by_id

get_definition_by_id(definition_id: DefinitionIdentifier) -> ActivityDefinition | None

Retrieve an activity definition by ID (UUID or name)

Parameters:

Name	Type	Description	Default
definition_id	DefinitionIdentifier	Identifier for the activity definition. This method will accept and resolve any type of DefinitionIdentifier.	required

Returns:

Type	Description
ActivityDefinition | None	The activity definition if found, None otherwise.

Example

definition = client.activities.get_definition_by_id("definition_uuid")

Source code in kalbio/activities.py

def get_definition_by_id(
    self, definition_id: DefinitionIdentifier
) -> ActivityDefinition | None:
    """Retrieve an activity definition by ID (UUID or name)

    Args:
        definition_id: Identifier for the activity definition.

            This method will accept and resolve any type of DefinitionIdentifier.

    Returns:
        The activity definition if found, None otherwise.

    Example:
        ```python
        definition = client.activities.get_definition_by_id("definition_uuid")
        ```
    """
    id_map = self._get_definition_id_map()
    identifier = self._resolve_definition_id(definition_id)

    if identifier is None:
        return None
    else:
        return id_map.get(identifier, None)

get_definitions_by_ids

get_definitions_by_ids(ids: list[DefinitionIdentifier]) -> list[ActivityDefinition]

Retrieve activity definitions by their identifiers

Parameters:

Name	Type	Description	Default
ids	list[DefinitionIdentifier]	List of definition identifiers to retrieve. This method will accept and resolve all types of DefinitionIdentifier.	required

Returns:

Type	Description
list[ActivityDefinition]	List of found activity definitions.

Example

defs = client.activities.get_definitions_by_ids(["def1", "def2"])

Source code in kalbio/activities.py

def get_definitions_by_ids(
    self, ids: List[DefinitionIdentifier]
) -> List[ActivityDefinition]:
    """Retrieve activity definitions by their identifiers

    Args:
        ids: List of definition identifiers to retrieve.

            This method will accept and resolve all types of DefinitionIdentifier.

    Returns:
        List of found activity definitions.

    Example:
        ```python
        defs = client.activities.get_definitions_by_ids(["def1", "def2"])
        ```
    """
    definitions = []

    for definition_id in ids:
        res = self.get_definition_by_id(definition_id)
        if res:
            definitions.append(res)

    return definitions

get_activity_definition_by_external_id

get_activity_definition_by_external_id(external_id: str) -> ActivityDefinition | None

Retrieve an activity definition by its external identifier.

Parameters:

Name	Type	Description	Default
external_id	str	The external identifier of the activity definition to retrieve.	required

Returns:

Type	Description
ActivityDefinition | None	The ActivityDefinition object if found, otherwise None.

Example

definition = client.activities.get_activity_definition_by_external_id("jira-def-7")

Source code in kalbio/activities.py

def get_activity_definition_by_external_id(
    self, external_id: str
) -> ActivityDefinition | None:
    """Retrieve an activity definition by its external identifier.

    Args:
        external_id: The external identifier of the activity definition to retrieve.

    Returns:
        The ActivityDefinition object if found, otherwise None.

    Example:
        ```python
        definition = client.activities.get_activity_definition_by_external_id("jira-def-7")
        ```
    """
    definitions = self.get_definitions()
    return next(
        (d for d in definitions if d.external_id == external_id),
        None,
    )