Skip to content

Activities

activities

Activities Module for Kaleidoscope API Client.

This module provides comprehensive 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 activites
  • Related programs
  • Parent-child activity relationships
  • Activity dependencies and scheduling

Classes:

Name Description
ActivityStatusEnum

Enumeration of possible activity statuses used across activity workflows.

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.

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.

ActivityType

ActivityType = 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 = {'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.

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.

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
Source code in kalpy/activities.py
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def update_property(self, property_value: Any) -> None:
    """Update the property with a new value.

    Args:
        property_value (Any): The new value to set for the property.
    """
    try:
        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)
    except Exception as e:
        _logger.error(f"Error updating property {self.id}: {e}")
        return None

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 or None

A dict of response JSON data (contains reference to the uploaded file) if request successful, otherwise None.

Source code in kalpy/activities.py
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
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 (str): The name of the file to be updated.
        file_data (BinaryIO): The binary data of the file to be updated.
        file_type (str): The MIME type of the file to be updated.

    Returns:
        (dict or None): A dict of response JSON data (contains reference to the
            uploaded file) if request successful, otherwise None.
    """
    try:
        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
    except Exception as e:
        _logger.error(f"Error adding file to property {self.id}: {e}")
        return None

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

activities cached property

activities: 'Activity' | None

Get the activities for this activity definition.

Returns:

Type Description
'Activity' | None

List[Activity]: The activities associated with this activity definition.

Note

This is a cached property.

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

record_ids list[str]

List of record IDs linked to this activity.

Methods:

Name Description
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.

activity_definition cached property

activity_definition: 'ActivityDefinition' | None

Get the activity definition for this activity.

Returns:

Type Description
ActivityDefinition or None

The activity definition associated with this activity. If the activity has no definition, returns None.

Note

This is a cached property.

assigned_users cached property

assigned_users: list[WorkspaceUser]

Get the assigned users for this activity.

Returns:

Type Description
list[WorkspaceUser]

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]

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]

List[Label]: The labels associated with this activity.

Note

This is a cached property.

programs cached property

programs: list[Program]

Retrieve the programs associated with this activity.

Returns:

Type Description
list[Program]

List[Program]: A list of Program instances fetched by their IDs.

Note

This is a cached property.

child_activities cached property

child_activities: list['Activity']

Retrieve the child activities associated with this activity.

Returns:

Type Description
list['Activity']

List[Activity]: A list of Activity objects representing the child activities.

Note

This is a cached property.

records cached property

records: list['Record']

Retrieve the records associated with this activity.

Returns:

Type Description
list['Record']

List[Record]: A list of Record objects corresponding to the activity.

Note

This is a cached property.

get_record

get_record(identifier: str) -> bool

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

Returns:

Type Description
Record or None

The record if it is in the operation, otherwise None

Source code in kalpy/activities.py
461
462
463
464
465
466
467
468
469
470
def get_record(self, identifier: str) -> bool:
    """Retrieves the record with the given identifier if it is in the operation

    Returns:
        (Record or None): The record if it is in the operation, otherwise None
    """
    return next(
        (r for r in self.records if r.record_identifier == identifier),
        None,
    )

has_record

has_record(identifier: str) -> bool

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

Returns:

Name Type Description
bool bool

Whether the record is in the operation

Source code in kalpy/activities.py
472
473
474
475
476
477
478
def has_record(self, identifier: str) -> bool:
    """Retrieve whether a record with the given identifier is in the operation

    Returns:
        bool: Whether the record is in the operation
    """
    return self.get_record(identifier) != 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.

{}
Source code in kalpy/activities.py
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
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.
    """
    try:
        resp = self._client._put("/activities/" + self.id, kwargs)
        if resp:
            for key, value in resp.items():
                if hasattr(self, key):
                    setattr(self, key, value)
    except Exception as e:
        _logger.error(f"Error updating activity: {e}")
        return None

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
Source code in kalpy/activities.py
497
498
499
500
501
502
503
504
505
506
507
508
509
def add_records(self, record_ids: List[str]) -> None:
    """Add a list of record IDs to the activity.

    Args:
        record_ids (List[str]): A list of record IDs to be added to the activity.
    """
    try:
        self._client._put(
            "/operations/" + self.id + "/records", {"record_ids": record_ids}
        )
    except Exception as e:
        _logger.error(f"Error adding record: {e}")
        return None

get_record_data

get_record_data() -> list[dict]

Retrieve data from all this activity's associated records.

Returns:

Type Description
list[dict]

List[dict]: A list containing the activity data for each record, obtained by calling get_activity_data with the current activity's ID.

Source code in kalpy/activities.py
511
512
513
514
515
516
517
518
519
520
521
def get_record_data(self) -> List[dict]:
    """Retrieve data from all this activity's associated records.

    Returns:
        List[dict]: A list containing the activity data for each record,
            obtained by calling get_activity_data with the current activity's ID.
    """
    data = []
    for record in self.records:
        data.append(record.get_activity_data(self.id))
    return data

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
create_activity

Create a new activity.

get_activities

Retrieve all activities in the workspace, including experiments.

get_activity_by_id

Retrieve an activity by its unique identifier.

get_activity_by_external_id

Retrieve an activity by its external identifier.

get_activities_by_ids

Fetch activities by their IDs in batches.

get_definitions

Retrieve all available activity definitions.

get_definition_by_name

Retrieve an activity definition by name.

get_definition_by_id

Retrieve an activity definition by ID.

get_activity_definition_by_external_id

Retrieve an activity definition by its external identifier.

get_activities_with_record

Retrieve all activities that contain a specific record.

Source code in kalpy/activities.py
536
537
def __init__(self, client: KaleidoscopeClient):
    self._client = client

create_activity

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

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.

[]
activity_definition_id str | None

ID of the activity definition. Defaults to None.

None
assigned_user_ids list[str] | None

List of user IDs to assign to the activity. Defaults to 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 or None

The newly created activity instance or None if activity creation was not successful.

Source code in kalpy/activities.py
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
def create_activity(
    self,
    title: str,
    activity_type: ActivityType,
    program_ids: Optional[list[str]] = [],
    activity_definition_id: Optional[str] = None,
    assigned_user_ids: Optional[List[str]] = [],
    start_date: Optional[datetime] = None,
    duration: Optional[int] = None,
) -> Activity | None:
    """Create a new activity.

    Args:
        title (str): The title/name of the activity.
        activity_type (ActivityType): The type of activity (e.g. task, experiment, etc.).
        program_ids (Optional[list[str]]): List of program IDs to associate with
            the activity. Defaults to None.
        activity_definition_id (Optional[str]): ID of the activity definition.
            Defaults to None.
        assigned_user_ids (Optional[List[str]]): List of user IDs to assign to
            the activity. Defaults to None.
        start_date (Optional[datetime]): Start date for the activity. Defaults to None.
        duration (Optional[int]): Duration in days for the activity. Defaults to None.

    Returns:
        (Activity or None): The newly created activity instance or None if activity
            creation was not successful.
    """
    try:
        start_date_formatted = start_date.isoformat() if start_date else None
        payload = {
            "program_ids": program_ids,
            "title": title,
            "activity_type": activity_type,
            "definition_id": activity_definition_id,
            "record_ids": [],
            "assigned_user_ids": assigned_user_ids,
            "start_date": start_date_formatted,
            "duration": duration,
        }
        resp = self._client._post("/activities", payload)
        return self._create_activity(resp[0])
    except Exception as e:
        _logger.error(f"Error creating activity {title}: {e}")
        return None

get_activities cached

get_activities() -> list[Activity]

Retrieve all activities in the workspace, including experiments.

Returns:

Type Description
list[Activity]

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.

Source code in kalpy/activities.py
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
@lru_cache
def get_activities(self) -> List[Activity]:
    """Retrieve all activities in the workspace, including experiments.

    Returns:
        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.
    """
    try:
        resp = self._client._get("/activities")
        return self._create_activity_list(resp)
    except Exception as e:
        _logger.error(f"Error fetching activities: {e}")
        self.get_activities.cache_clear()
        return []

get_activity_by_id

get_activity_by_id(activity_id: str) -> Activity | None

Retrieve an activity by its unique identifier.

Parameters:

Name Type Description Default
activity_id str

The unique identifier of the activity to retrieve.

required

Returns:

Type Description
Activity or None

The Activity object if found, otherwise None.

Source code in kalpy/activities.py
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
def get_activity_by_id(self, activity_id: str) -> Activity | None:
    """Retrieve an activity by its unique identifier.

    Args:
        activity_id (str): The unique identifier of the activity to retrieve.

    Returns:
        (Activity or None): The Activity object if found, otherwise None.
    """
    try:
        resp = self._client._get("/activities/" + activity_id)
        if resp is None:
            return None

        return self._create_activity(resp)
    except Exception as e:
        _logger.error(f"Error fetching activity {activity_id}: {e}")
        return None

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 or None

The Activity object if found, otherwise None.

Source code in kalpy/activities.py
661
662
663
664
665
666
667
668
669
670
671
672
673
674
def get_activity_by_external_id(self, external_id: str) -> Activity | None:
    """Retrieve an activity by its external identifier.

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

    Returns:
        (Activity or None): The Activity object if found, otherwise None.
    """
    activities = self.get_definitions()
    return next(
        (a for a in activities if a.external_id == external_id),
        None,
    )

get_activities_by_ids

get_activities_by_ids(ids: list[str], batch_size: int = 250) -> list[Activity]

Fetch activities by their IDs in batches.

Parameters:

Name Type Description Default
ids list[str]

A list of activity ID strings to fetch.

required
batch_size int

The number of IDs to include in each batch request. Defaults to 250.

250

Returns:

Type Description
list[Activity]

List[Activity]: A list of Activity objects corresponding to the provided IDs.

Note

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

Source code in kalpy/activities.py
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
def get_activities_by_ids(
    self, ids: List[str], batch_size: int = 250
) -> List[Activity]:
    """Fetch activities by their IDs in batches.

    Args:
        ids (List[str]): A list of activity ID strings to fetch.
        batch_size (int): The number of IDs to include in each batch request.
            Defaults to 250.

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

    Note:
        If an exception occurs, logs the error and returns an empty list.
    """
    try:
        all_activities = []

        for i in range(0, len(ids), batch_size):
            batch = ids[i : i + batch_size]
            resp = self._client._get(f"/activities?activity_ids={",".join(batch)}")
            activities = self._create_activity_list(resp)
            all_activities.extend(activities)

        return all_activities
    except Exception as e:
        _logger.error(f"Error fetching activities: {e}")
        return []

get_definitions cached

get_definitions() -> list[ActivityDefinition]

Retrieve all available activity definitions.

Returns:

Type Description
list[ActivityDefinition]

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.

Source code in kalpy/activities.py
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
@lru_cache
def get_definitions(self) -> List[ActivityDefinition]:
    """Retrieve all available activity definitions.

    Returns:
        List[ActivityDefinition]: 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.
    """
    try:
        resp = self._client._get("/activity_definitions")
        return TypeAdapter(List[ActivityDefinition]).validate_python(resp)
    except Exception as e:
        _logger.error(f"Error fetching activity definitions: {e}")
        self.get_definitions.cache_clear()
        return []

get_definition_by_name

get_definition_by_name(name: str) -> ActivityDefinition | None

Retrieve an activity definition by name.

Parameters:

Name Type Description Default
name str

The name of the activity definition.

required

Returns:

Type Description
ActivityDefinition or None

The activity definition if found, None otherwise.

Source code in kalpy/activities.py
728
729
730
731
732
733
734
735
736
737
738
739
740
741
def get_definition_by_name(self, name: str) -> ActivityDefinition | None:
    """Retrieve an activity definition by name.

    Args:
        name (str): The name of the activity definition.

    Returns:
        (ActivityDefinition or None): The activity definition if found, None otherwise.
    """
    definitions = self.get_definitions()
    return next(
        (d for d in definitions if d.title == name),
        None,
    )

get_definition_by_id

get_definition_by_id(definition_id: str) -> ActivityDefinition | None

Retrieve an activity definition by ID.

Parameters:

Name Type Description Default
definition_id str

Unique identifier for the activity definition.

required

Returns:

Type Description
ActivityDefinition or None

The activity definition if found, None otherwise.

Source code in kalpy/activities.py
743
744
745
746
747
748
749
750
751
752
753
754
755
756
def get_definition_by_id(self, definition_id: str) -> ActivityDefinition | None:
    """Retrieve an activity definition by ID.

    Args:
        definition_id (str): Unique identifier for the activity definition.

    Returns:
        (ActivityDefinition or None): The activity definition if found, None otherwise.
    """
    definitions = self.get_definitions()
    return next(
        (d for d in definitions if d.id == definition_id),
        None,
    )

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 or None

The ActivityDefinition object if found, otherwise None.

Source code in kalpy/activities.py
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
def get_activity_definition_by_external_id(
    self, external_id: str
) -> ActivityDefinition | None:
    """Retrieve an activity definition by its external identifier.

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

    Returns:
        (ActivityDefinition or None): The ActivityDefinition object if found, otherwise None.
    """
    definitions = self.get_definitions()
    return next(
        (d for d in definitions if d.external_id == external_id),
        None,
    )

get_activities_with_record

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

Retrieve all activities that contain a specific record.

Parameters:

Name Type Description Default
record_id str

Unique identifier for the record.

required

Returns:

Type Description
list[Activity]

List[Activity]: Activities that include the specified record.

Note

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

Source code in kalpy/activities.py
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
def get_activities_with_record(self, record_id: str) -> List[Activity]:
    """Retrieve all activities that contain a specific record.

    Args:
        record_id (str): Unique identifier for the record.

    Returns:
        List[Activity]: Activities that include the specified record.

    Note:
        If an exception occurs, logs the error and returns an empty list.
    """
    try:
        resp = self._client._get("/records/" + record_id + "/operations")
        return self._create_activity_list(resp)
    except Exception as e:
        _logger.error(f"Error fetching activity with record {record_id}: {e}")
        return []