Property Data

albert.collections.property_data.PropertyDataCollection

PropertyDataCollection(*, session: AlbertSession)

Bases: BaseCollection

PropertyDataCollection is a collection class for managing Property Data entities in the Albert platform.

Parameters:

Name	Type	Description	Default
session	AlbertSession	The Albert session instance.	required

Methods:

Name	Description
get_properties_on_inventory	Returns all the properties of an inventory item.
add_properties_to_inventory	Add new properties to an inventory item.
update_property_on_inventory	Update a property on an inventory item.
get_task_block_properties	Returns all the properties within a Property Task block for a specific inventory item.
check_for_task_data	Checks if a task has data.
check_block_interval_for_data	Check if a specific block interval has data.
get_all_task_properties	Collect task property data for block/inventory combinations in a task.
update_property_on_task	Updates a specific property on a task.
add_properties_to_task	Add new task properties for a given task.
update_or_create_task_properties	Update or create task properties for a given task.
bulk_load_task_properties	Bulk load task properties for a given task. WARNING: This will overwrite any existing properties!
bulk_delete_task_data	Bulk delete task data for a given task.
search	Search for property data with various filtering options.

Attributes:

Name	Type	Description
base_path

Source code in src/albert/collections/property_data.py

def __init__(self, *, session: AlbertSession):
    """
    Initializes the CompanyCollection with the provided session.

    Parameters
    ----------
    session : AlbertSession
        The Albert session instance.
    """
    super().__init__(session=session)
    self.base_path = f"/api/{PropertyDataCollection._api_version}/propertydata"

base_path

base_path = f'/api/{_api_version}/propertydata'

get_properties_on_inventory

get_properties_on_inventory(
    *, inventory_id: InventoryId
) -> InventoryPropertyData

Returns all the properties of an inventory item.

Parameters:

Name	Type	Description	Default
inventory_id	InventoryId	The ID of the inventory item to retrieve properties for.	required

Returns:

Type	Description
InventoryPropertyData	The properties of the inventory item.

Source code in src/albert/collections/property_data.py

@validate_call
def get_properties_on_inventory(self, *, inventory_id: InventoryId) -> InventoryPropertyData:
    """Returns all the properties of an inventory item.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory item to retrieve properties for.

    Returns
    -------
    InventoryPropertyData
        The properties of the inventory item.
    """
    params = {"entity": "inventory", "id": [inventory_id]}
    response = self.session.get(url=self.base_path, params=params)
    response_json = response.json()
    return InventoryPropertyData(**response_json[0])

add_properties_to_inventory

add_properties_to_inventory(
    *,
    inventory_id: InventoryId,
    properties: list[InventoryDataColumn],
) -> list[InventoryPropertyDataCreate]

Add new properties to an inventory item.

Parameters:

Name	Type	Description	Default
inventory_id	InventoryId	The ID of the inventory item to add properties to.	required
properties	list[InventoryDataColumn]	The properties to add.	required

Returns:

Type	Description
list[InventoryPropertyDataCreate]	The registered properties.

Source code in src/albert/collections/property_data.py

@validate_call
def add_properties_to_inventory(
    self, *, inventory_id: InventoryId, properties: list[InventoryDataColumn]
) -> list[InventoryPropertyDataCreate]:
    """Add new properties to an inventory item.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory item to add properties to.
    properties : list[InventoryDataColumn]
        The properties to add.

    Returns
    -------
    list[InventoryPropertyDataCreate]
        The registered properties.
    """
    returned = []
    for p in properties:
        # Can only add one at a time.
        create_object = InventoryPropertyDataCreate(
            inventory_id=inventory_id, data_columns=[p]
        )
        response = self.session.post(
            self.base_path,
            json=create_object.model_dump(exclude_none=True, by_alias=True, mode="json"),
        )
        response_json = response.json()
        logger.info(response_json.get("message", None))
        returned.append(InventoryPropertyDataCreate(**response_json))
    return returned

update_property_on_inventory

update_property_on_inventory(
    *,
    inventory_id: InventoryId,
    property_data: InventoryDataColumn,
) -> InventoryPropertyData

Update a property on an inventory item.

Parameters:

Name	Type	Description	Default
inventory_id	InventoryId	The ID of the inventory item to update the property on.	required
property_data	InventoryDataColumn	The updated property data.	required

Returns:

Type	Description
InventoryPropertyData	The updated property data as returned by the server.

Source code in src/albert/collections/property_data.py

@validate_call
def update_property_on_inventory(
    self, *, inventory_id: InventoryId, property_data: InventoryDataColumn
) -> InventoryPropertyData:
    """Update a property on an inventory item.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory item to update the property on.
    property_data : InventoryDataColumn
        The updated property data.

    Returns
    -------
    InventoryPropertyData
        The updated property data as returned by the server.
    """
    existing_properties = self.get_properties_on_inventory(inventory_id=inventory_id)
    existing_value = None
    for p in existing_properties.custom_property_data:
        if p.data_column.data_column_id == property_data.data_column_id:
            existing_value = (
                p.data_column.property_data.value
                if p.data_column.property_data.value is not None
                else p.data_column.property_data.string_value
                if p.data_column.property_data.string_value is not None
                else str(p.data_column.property_data.numeric_value)
                if p.data_column.property_data.numeric_value is not None
                else None
            )
            existing_id = p.data_column.property_data.id
            break
    if existing_value is not None:
        payload = [
            PropertyDataPatchDatum(
                operation=PatchOperation.UPDATE,
                id=existing_id,
                attribute="value",
                new_value=property_data.value,
                old_value=existing_value,
            )
        ]
    else:
        payload = [
            PropertyDataPatchDatum(
                operation=PatchOperation.ADD,
                id=existing_id,
                attribute="value",
                new_value=property_data.value,
            )
        ]

    self.session.patch(
        url=f"{self.base_path}/{inventory_id}",
        json=[x.model_dump(exclude_none=True, by_alias=True, mode="json") for x in payload],
    )
    return self.get_properties_on_inventory(inventory_id=inventory_id)

get_task_block_properties

get_task_block_properties(
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
) -> TaskPropertyData

Returns all the properties within a Property Task block for a specific inventory item.

Parameters:

Name	Type	Description	Default
inventory_id	InventoryId	The ID of the inventory.	required
task_id	TaskId	The Property task ID.	required
block_id	BlockId	The Block ID of the block to retrieve properties for.	required
lot_id	LotId | None	The specific Lot of the inventory Item to retrieve lots for, by default None	None

Returns:

Type	Description
TaskPropertyData	The properties of the inventory item within the block.

Source code in src/albert/collections/property_data.py

@validate_call
def get_task_block_properties(
    self,
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
) -> TaskPropertyData:
    """Returns all the properties within a Property Task block for a specific inventory item.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory.
    task_id : TaskId
        The Property task ID.
    block_id : BlockId
        The Block ID of the block to retrieve properties for.
    lot_id : LotId | None, optional
        The specific Lot of the inventory Item to retrieve lots for, by default None

    Returns
    -------
    TaskPropertyData
        The properties of the inventory item within the block.
    """
    params = {
        "entity": "task",
        "blockId": block_id,
        "id": task_id,
        "inventoryId": inventory_id,
        "lotId": lot_id,
    }
    params = {k: v for k, v in params.items() if v is not None}

    response = self.session.get(url=self.base_path, params=params)
    response_json = response.json()
    return TaskPropertyData(**response_json[0])

check_for_task_data

check_for_task_data(
    *, task_id: TaskId
) -> list[CheckPropertyData]

Checks if a task has data.

Parameters:

Name	Type	Description	Default
task_id	TaskId	The ID of the task to check for data.	required

Returns:

Type	Description
list[CheckPropertyData]	A list of CheckPropertyData entities representing the data status of each block + inventory item of the task.

Source code in src/albert/collections/property_data.py

@validate_call
def check_for_task_data(self, *, task_id: TaskId) -> list[CheckPropertyData]:
    """Checks if a task has data.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task to check for data.

    Returns
    -------
    list[CheckPropertyData]
        A list of CheckPropertyData entities representing the data status of each block + inventory item of the task.
    """
    task_info = self._get_task_from_id(id=task_id)

    params = {
        "entity": "block",
        "action": "checkdata",
        "parentId": task_id,
        "id": [x.id for x in task_info.blocks],
    }

    response = self.session.get(url=self.base_path, params=params)
    return [CheckPropertyData(**x) for x in response.json()]

check_block_interval_for_data

check_block_interval_for_data(
    *,
    block_id: BlockId,
    task_id: TaskId,
    interval_id: IntervalId,
) -> CheckPropertyData

Check if a specific block interval has data.

Parameters:

Name	Type	Description	Default
block_id	BlockId	The ID of the block.	required
task_id	TaskId	The ID of the task.	required
interval_id	IntervalId	The ID of the interval.	required

Returns:

Type	Description
CheckPropertyData	description

Source code in src/albert/collections/property_data.py

@validate_call
def check_block_interval_for_data(
    self, *, block_id: BlockId, task_id: TaskId, interval_id: IntervalId
) -> CheckPropertyData:
    """Check if a specific block interval has data.

    Parameters
    ----------
    block_id : BlockId
        The ID of the block.
    task_id : TaskId
        The ID of the task.
    interval_id : IntervalId
        The ID of the interval.

    Returns
    -------
    CheckPropertyData
        _description_
    """
    params = {
        "entity": "block",
        "action": "checkdata",
        "id": block_id,
        "parentId": task_id,
        "intervalId": interval_id,
    }

    response = self.session.get(url=self.base_path, params=params)
    return CheckPropertyData(response.json())

get_all_task_properties

get_all_task_properties(
    *, task_id: TaskId, with_data_only: bool = False
) -> list[TaskPropertyData]

Collect task property data for block/inventory combinations in a task.

Parameters:

Name	Type	Description	Default
task_id	TaskId	The ID of the task to retrieve properties for.	required
with_data_only	bool	When True, only return combinations actually having task data (dataExist flag is true). Defaults to False.	False

Returns:

Type	Description
list[TaskPropertyData]	Task property data for each block/inventory/lot combination. When with_data_only is True, combinations without recorded data are omitted.

Source code in src/albert/collections/property_data.py

@validate_call
def get_all_task_properties(
    self, *, task_id: TaskId, with_data_only: bool = False
) -> list[TaskPropertyData]:
    """Collect task property data for block/inventory combinations in a task.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task to retrieve properties for.
    with_data_only : bool, optional
        When True, only return combinations actually having task data (``dataExist`` flag is true). Defaults to False.

    Returns
    -------
    list[TaskPropertyData]
        Task property data for each block/inventory/lot combination. When
        ``with_data_only`` is True, combinations without recorded data are omitted.
    """
    all_info = []
    task_data_info = self.check_for_task_data(task_id=task_id)
    for combo_info in task_data_info:
        if with_data_only and not combo_info.data_exists:
            continue
        all_info.append(
            self.get_task_block_properties(
                inventory_id=combo_info.inventory_id,
                task_id=task_id,
                block_id=combo_info.block_id,
                lot_id=combo_info.lot_id,
            )
        )
    return all_info

update_property_on_task

update_property_on_task(
    *,
    task_id: TaskId,
    patch_payload: list[PropertyDataPatchDatum],
    inventory_id: InventoryId | None = None,
    block_id: BlockId | None = None,
    lot_id: LotId | None = None,
    return_scope: ReturnScope = "task",
) -> list[TaskPropertyData]

Updates a specific property on a task.

Parameters:

Name	Type	Description	Default
task_id	TaskId	The ID of the task.	required
patch_payload	list[PropertyDataPatchDatum]	The specific patch to make to update the property.	required
inventory_id	InventoryId | None	Required when return_scope="block".	None
block_id	BlockId | None	Required when return_scope="block".	None
lot_id	LotId | None	Optional context for combo fetches.	None
return_scope	Literal[task, block, none]	Controls the response. "task" (default) returns all task properties, "block" returns only the affected block/inventory/lot combination, and "none" skips fetching data.	'task'

Returns:

Type	Description
list[TaskPropertyData]	A list of TaskPropertyData entities representing the properties within the task.

Source code in src/albert/collections/property_data.py

@validate_call
def update_property_on_task(
    self,
    *,
    task_id: TaskId,
    patch_payload: list[PropertyDataPatchDatum],
    inventory_id: InventoryId | None = None,
    block_id: BlockId | None = None,
    lot_id: LotId | None = None,
    return_scope: ReturnScope = "task",
) -> list[TaskPropertyData]:
    """Updates a specific property on a task.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task.
    patch_payload : list[PropertyDataPatchDatum]
        The specific patch to make to update the property.
    inventory_id : InventoryId | None, optional
        Required when return_scope="block".
    block_id : BlockId | None, optional
        Required when return_scope="block".
    lot_id : LotId | None, optional
        Optional context for combo fetches.
    return_scope : Literal["task", "block", "none"], optional
        Controls the response. "task" (default) returns all task properties,
        "block" returns only the affected block/inventory/lot combination, and "none" skips fetching data.

    Returns
    -------
    list[TaskPropertyData]
        A list of TaskPropertyData entities representing the properties within the task.
    """
    if len(patch_payload) > 0:
        self.session.patch(
            url=f"{self.base_path}/{task_id}",
            json=[
                x.model_dump(exclude_none=True, by_alias=True, mode="json")
                for x in patch_payload
            ],
        )
    return self._resolve_return_scope(
        task_id=task_id,
        return_scope=return_scope,
        inventory_id=inventory_id,
        block_id=block_id,
        lot_id=lot_id,
    )

add_properties_to_task

add_properties_to_task(
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
    properties: list[TaskPropertyCreate],
    return_scope: ReturnScope = "task",
) -> list[TaskPropertyData]

Add new task properties for a given task.

This method only works for new values. If a trial number is provided in the TaskPropertyCreate, it must relate to an existing trial. New trials must be added with no trial number provided. Do not try to create multiple new trials in one call as this will lead to unexpected behavior. Build out new trials in a loop if many new trials are needed.

Parameters:

Name	Type	Description	Default
inventory_id	InventoryId	The ID of the inventory.	required
task_id	TaskId	The ID of the task.	required
block_id	BlockId	The ID of the block.	required
lot_id	LotId	The ID of the lot, by default None.	None
properties	list[TaskPropertyCreate]	A list of TaskPropertyCreate entities representing the properties to add.	required
return_scope	Literal[task, block, none]	Controls the response. "task" (default) returns all task properties, "block" returns only the affected block/inventory/lot combination, and "none" skips fetching data.	'task'

Returns:

Type	Description
list[TaskPropertyData]	The newly created task properties.

Source code in src/albert/collections/property_data.py

@validate_call
def add_properties_to_task(
    self,
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
    properties: list[TaskPropertyCreate],
    return_scope: ReturnScope = "task",
) -> list[TaskPropertyData]:
    """
    Add new task properties for a given task.

    This method only works for new values. If a trial number is provided in the TaskPropertyCreate,
    it must relate to an existing trial. New trials must be added with no trial number provided.
    Do not try to create multiple new trials in one call as this will lead to unexpected behavior.
    Build out new trials in a loop if many new trials are needed.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory.
    task_id : TaskId
        The ID of the task.
    block_id : BlockId
        The ID of the block.
    lot_id : LotId, optional
        The ID of the lot, by default None.
    properties : list[TaskPropertyCreate]
        A list of TaskPropertyCreate entities representing the properties to add.
    return_scope : Literal["task", "block", "none"], optional
        Controls the response. "task" (default) returns all task properties,
        "block" returns only the affected block/inventory/lot combination, and "none" skips fetching data.

    Returns
    -------
    list[TaskPropertyData]
        The newly created task properties.
    """
    params = {
        "blockId": block_id,
        "inventoryId": inventory_id,
        "lotId": lot_id,
        "autoCalculate": "true",
        "history": "true",
    }
    params = {k: v for k, v in params.items() if v is not None}
    response = self.session.post(
        url=f"{self.base_path}/{task_id}",
        json=[x.model_dump(exclude_none=True, by_alias=True, mode="json") for x in properties],
        params=params,
    )

    registered_properties = [
        TaskPropertyCreate(**x) for x in response.json() if "DataTemplate" in x
    ]
    existing_data_rows = self.get_task_block_properties(
        inventory_id=inventory_id, task_id=task_id, block_id=block_id, lot_id=lot_id
    )
    patches = self._form_calculated_task_property_patches(
        existing_data_rows=existing_data_rows, properties=registered_properties
    )
    if len(patches) > 0:
        return self.update_property_on_task(
            task_id=task_id,
            patch_payload=patches,
            return_scope=return_scope,
            inventory_id=inventory_id,
            block_id=block_id,
            lot_id=lot_id,
        )

    return self._resolve_return_scope(
        task_id=task_id,
        return_scope=return_scope,
        inventory_id=inventory_id,
        block_id=block_id,
        lot_id=lot_id,
        prefetched_block=existing_data_rows,
    )

update_or_create_task_properties

update_or_create_task_properties(
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
    properties: list[TaskPropertyCreate],
    return_scope: ReturnScope = "task",
) -> list[TaskPropertyData]

Update or create task properties for a given task.

If a trial number is provided in the TaskPropertyCreate, it must relate to an existing trial. New trials must be added with no trial number provided. Do not try to create multiple new trials in one call as this will lead to unexpected behavior. Build out new trials in a loop if many new trials are needed.

Parameters:

Name	Type	Description	Default
inventory_id	InventoryId	The ID of the inventory.	required
task_id	TaskId	The ID of the task.	required
block_id	BlockId	The ID of the block.	required
lot_id	LotId	The ID of the lot, by default None.	None
properties	list[TaskPropertyCreate]	A list of TaskPropertyCreate entities representing the properties to update or create.	required
return_scope	Literal[task, block, none]	Controls the response. "task" (default) returns all task properties, "block" returns only the affected block/inventory/lot combination, and "none" skips fetching data.	'task'

Returns:

Type	Description
list[TaskPropertyData]	The updated or newly created task properties.

Source code in src/albert/collections/property_data.py

@validate_call
def update_or_create_task_properties(
    self,
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
    properties: list[TaskPropertyCreate],
    return_scope: ReturnScope = "task",
) -> list[TaskPropertyData]:
    """
    Update or create task properties for a given task.

    If a trial number is provided in the TaskPropertyCreate, it must relate to an existing trial.
    New trials must be added with no trial number provided. Do not try to create multiple new trials
    in one call as this will lead to unexpected behavior. Build out new trials in a loop if many new
    trials are needed.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory.
    task_id : TaskId
        The ID of the task.
    block_id : BlockId
        The ID of the block.
    lot_id : LotId, optional
        The ID of the lot, by default None.
    properties : list[TaskPropertyCreate]
        A list of TaskPropertyCreate entities representing the properties to update or create.
    return_scope : Literal["task", "block", "none"], optional
        Controls the response. "task" (default) returns all task properties,
        "block" returns only the affected block/inventory/lot combination, and "none" skips fetching data.

    Returns
    -------
    list[TaskPropertyData]
        The updated or newly created task properties.

    """
    existing_data_rows = self.get_task_block_properties(
        inventory_id=inventory_id, task_id=task_id, block_id=block_id, lot_id=lot_id
    )
    update_patches, new_values = self._form_existing_row_value_patches(
        existing_data_rows=existing_data_rows, properties=properties
    )

    calculated_patches = self._form_calculated_task_property_patches(
        existing_data_rows=existing_data_rows, properties=properties
    )
    all_patches = update_patches + calculated_patches
    if len(new_values) > 0:
        if len(all_patches) > 0:
            self.update_property_on_task(
                task_id=task_id,
                patch_payload=all_patches,
                return_scope="none",
                inventory_id=inventory_id,
                block_id=block_id,
                lot_id=lot_id,
            )
        return self.add_properties_to_task(
            inventory_id=inventory_id,
            task_id=task_id,
            block_id=block_id,
            lot_id=lot_id,
            properties=new_values,
            return_scope=return_scope,
        )
    else:
        return self.update_property_on_task(
            task_id=task_id,
            patch_payload=all_patches,
            return_scope=return_scope,
            inventory_id=inventory_id,
            block_id=block_id,
            lot_id=lot_id,
        )

bulk_load_task_properties

bulk_load_task_properties(
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    property_data: BulkPropertyData,
    interval="default",
    lot_id: LotId = None,
    return_scope: ReturnScope = "task",
) -> list[TaskPropertyData]

Bulk load task properties for a given task. WARNING: This will overwrite any existing properties! BulkPropertyData column names must exactly match the names of the data columns (Case Sensitive).

Parameters:

Name	Type	Description	Default
inventory_id	InventoryId	The ID of the inventory.	required
task_id	TaskId	The ID of the task.	required
block_id	BlockId	The ID of the block.	required
lot_id	LotId	The ID of the lot, by default None.	None
interval	str	The interval to use for the properties, by default "default". Can be obtained using Workflow.get_interval_id().	'default'
property_data	BulkPropertyData	A list of columnwise data containing all your rows of data for a single interval. Can be created using BulkPropertyData.from_dataframe().	required
return_scope	Literal[task, block, none]	Controls the response. "task" (default) returns all task properties, "block" returns only the affected block/inventory/lot combination, and "none" skips fetching data.	'task'

Returns:

Type	Description
list[TaskPropertyData]	The updated or newly created task properties.

Example

from albert.resources.property_data import BulkPropertyData

data = BulkPropertyData.from_dataframe(df=my_dataframe)
res = client.property_data.bulk_load_task_properties(
    block_id="BLK1",
    inventory_id="INVEXP102748-042",
    property_data=data,
    task_id="TASFOR291760",
)

[TaskPropertyData(id="TASFOR291760", ...)]

Source code in src/albert/collections/property_data.py

def bulk_load_task_properties(
    self,
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    property_data: BulkPropertyData,
    interval="default",
    lot_id: LotId = None,
    return_scope: ReturnScope = "task",
) -> list[TaskPropertyData]:
    """
    Bulk load task properties for a given task. WARNING: This will overwrite any existing properties!
    BulkPropertyData column names must exactly match the names of the data columns (Case Sensitive).

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory.
    task_id : TaskId
        The ID of the task.
    block_id : BlockId
        The ID of the block.
    lot_id : LotId, optional
        The ID of the lot, by default None.
    interval : str, optional
        The interval to use for the properties, by default "default". Can be obtained using Workflow.get_interval_id().
    property_data : BulkPropertyData
        A list of columnwise data containing all your rows of data for a single interval. Can be created using BulkPropertyData.from_dataframe().
    return_scope : Literal["task", "block", "none"], optional
        Controls the response. "task" (default) returns all task properties,
        "block" returns only the affected block/inventory/lot combination, and "none" skips fetching data.

    Returns
    -------
    list[TaskPropertyData]
        The updated or newly created task properties.

    Example
    -------

    ```python
    from albert.resources.property_data import BulkPropertyData

    data = BulkPropertyData.from_dataframe(df=my_dataframe)
    res = client.property_data.bulk_load_task_properties(
        block_id="BLK1",
        inventory_id="INVEXP102748-042",
        property_data=data,
        task_id="TASFOR291760",
    )

    [TaskPropertyData(id="TASFOR291760", ...)]
    ```
    """
    property_df = pd.DataFrame(
        {x.data_column_name: x.data_series for x in property_data.columns}
    )

    def _get_column_map(dataframe: pd.DataFrame, property_data: TaskPropertyData):
        data_col_info = property_data.data[0].trials[0].data_columns  # PropertyValue
        column_map = {}
        for col in dataframe.columns:
            column = [x for x in data_col_info if x.name == col]
            if len(column) == 1:
                column_map[col] = column[0]
            else:
                raise ValueError(
                    f"Column '{col}' not found in block data columns or multiple matches found."
                )
        return column_map

    def _df_to_task_prop_create_list(
        dataframe: pd.DataFrame,
        column_map: dict[str, PropertyValue],
        data_template_id: DataTemplateId,
    ) -> list[TaskPropertyCreate]:
        task_prop_create_list = []
        for i, row in dataframe.iterrows():
            for col_name, col_info in column_map.items():
                if col_name not in dataframe.columns:
                    raise ValueError(f"Column '{col_name}' not found in DataFrame.")

                task_prop_create = TaskPropertyCreate(
                    data_column=TaskDataColumn(
                        data_column_id=col_info.id,
                        column_sequence=col_info.sequence,
                    ),
                    value=str(row[col_name]),
                    visible_trial_number=i + 1,
                    interval_combination=interval,
                    data_template=EntityLink(id=data_template_id),
                )
                task_prop_create_list.append(task_prop_create)
        return task_prop_create_list

    task_prop_data = self.get_task_block_properties(
        inventory_id=inventory_id, task_id=task_id, block_id=block_id, lot_id=lot_id
    )
    column_map = _get_column_map(property_df, task_prop_data)
    all_task_prop_create = _df_to_task_prop_create_list(
        dataframe=property_df,
        column_map=column_map,
        data_template_id=task_prop_data.data_template.id,
    )
    with suppress(NotFoundError):
        # This is expected if the task is new and has no data yet.
        self.bulk_delete_task_data(
            task_id=task_id,
            block_id=block_id,
            inventory_id=inventory_id,
            lot_id=lot_id,
            interval_id=interval,
        )
    return self.add_properties_to_task(
        inventory_id=inventory_id,
        task_id=task_id,
        block_id=block_id,
        lot_id=lot_id,
        properties=all_task_prop_create,
        return_scope=return_scope,
    )

bulk_delete_task_data

bulk_delete_task_data(
    *,
    task_id: TaskId,
    block_id: BlockId,
    inventory_id: InventoryId,
    lot_id: LotId | None = None,
    interval_id=None,
) -> None

Bulk delete task data for a given task.

Parameters:

Name	Type	Description	Default
task_id	TaskId	The ID of the task.	required
block_id	BlockId	The ID of the block.	required
inventory_id	InventoryId	The ID of the inventory.	required
lot_id	LotId	The ID of the lot, by default None.	None
interval_id	IntervalId	The ID of the interval, by default None. If provided, will delete data for this specific interval.	None

Returns:

Type	Description
None

Source code in src/albert/collections/property_data.py

def bulk_delete_task_data(
    self,
    *,
    task_id: TaskId,
    block_id: BlockId,
    inventory_id: InventoryId,
    lot_id: LotId | None = None,
    interval_id=None,
) -> None:
    """
    Bulk delete task data for a given task.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task.
    block_id : BlockId
        The ID of the block.
    inventory_id : InventoryId
        The ID of the inventory.
    lot_id : LotId, optional
        The ID of the lot, by default None.
    interval_id : IntervalId, optional
        The ID of the interval, by default None. If provided, will delete data for this specific interval.

    Returns
    -------
    None
    """
    params = {
        "inventoryId": inventory_id,
        "blockId": block_id,
        "lotId": lot_id,
        "intervalRow": interval_id if interval_id != "default" else None,
    }
    params = {k: v for k, v in params.items() if v is not None}
    self.session.delete(f"{self.base_path}/{task_id}", params=params)

search

search(
    *,
    result: str | None = None,
    text: str | None = None,
    order: OrderBy | None = None,
    sort_by: str | None = None,
    inventory_ids: list[SearchInventoryId]
    | SearchInventoryId
    | None = None,
    project_ids: list[SearchProjectId]
    | SearchProjectId
    | None = None,
    lot_ids: list[LotId] | LotId | None = None,
    data_template_ids: DataTemplateId
    | list[DataTemplateId]
    | None = None,
    data_column_ids: DataColumnId
    | list[DataColumnId]
    | None = None,
    category: list[DataEntity] | DataEntity | None = None,
    data_templates: list[str] | str | None = None,
    data_columns: list[str] | str | None = None,
    parameters: list[str] | str | None = None,
    parameter_group: list[str] | str | None = None,
    unit: list[str] | str | None = None,
    created_by: list[UserId] | UserId | None = None,
    task_created_by: list[UserId] | UserId | None = None,
    return_fields: list[str] | str | None = None,
    return_facets: list[str] | str | None = None,
    max_items: int | None = None,
) -> Iterator[PropertyDataSearchItem]

Search for property data with various filtering options.

Parameters:

Name	Type	Description	Default
result	str	Query using syntax, e.g. result=viscosity(<200)@temperature(25).	None
text	str	Free text search across all fields.	None
order	OrderBy	Sort order (ascending/descending).	None
sort_by	str	Field to sort results by.	None
inventory_ids	SearchInventoryId or list[SearchInventoryId]	Filter by inventory IDs.	None
project_ids	SearchProjectId or list[SearchProjectId]	Filter by project IDs.	None
lot_ids	LotId or list[LotId]	Filter by lot IDs.	None
data_template_ids	DataTemplateId or list[DataTemplateId]	Filter by data template IDs.	None
data_column_ids	DataColumnId or list[DataColumnId]	Filter by data column IDs.	None
category	DataEntity or list[DataEntity]	Filter by data entity categories.	None
data_templates	str or list[str]	Filter by data template names.	None
data_columns	str or list[str]	Filter by data column names.	None
parameters	str or list[str]	Filter by parameter names.	None
parameter_group	str or list[str]	Filter by parameter group names.	None
unit	str or list[str]	Filter by unit names.	None
created_by	UserId or list[UserId]	Filter by user IDs who created the data.	None
task_created_by	UserId or list[UserId]	Filter by user IDs who created the task.	None
return_fields	str or list[str]	Specific fields to return.	None
return_facets	str or list[str]	Specific facets to return.	None
max_items	int	Maximum number of items to return in total. If None, fetches all available items.	None

Returns:

Type	Description
Iterator[PropertyDataSearchItem]	An iterator of search results matching the specified filters.

Source code in src/albert/collections/property_data.py

@validate_call
def search(
    self,
    *,
    result: str | None = None,
    text: str | None = None,
    # Sorting/pagination
    order: OrderBy | None = None,
    sort_by: str | None = None,
    # Core platform identifiers
    inventory_ids: list[SearchInventoryId] | SearchInventoryId | None = None,
    project_ids: list[SearchProjectId] | SearchProjectId | None = None,
    lot_ids: list[LotId] | LotId | None = None,
    data_template_ids: DataTemplateId | list[DataTemplateId] | None = None,
    data_column_ids: DataColumnId | list[DataColumnId] | None = None,
    # Data structure filters
    category: list[DataEntity] | DataEntity | None = None,
    data_templates: list[str] | str | None = None,
    data_columns: list[str] | str | None = None,
    # Data content filters
    parameters: list[str] | str | None = None,
    parameter_group: list[str] | str | None = None,
    unit: list[str] | str | None = None,
    # User filters
    created_by: list[UserId] | UserId | None = None,
    task_created_by: list[UserId] | UserId | None = None,
    # Response customization
    return_fields: list[str] | str | None = None,
    return_facets: list[str] | str | None = None,
    # Pagination
    max_items: int | None = None,
) -> Iterator[PropertyDataSearchItem]:
    """
    Search for property data with various filtering options.

    Parameters
    ----------
    result : str, optional
        Query using syntax, e.g. result=viscosity(<200)@temperature(25).
    text : str, optional
        Free text search across all fields.
    order : OrderBy, optional
        Sort order (ascending/descending).
    sort_by : str, optional
        Field to sort results by.
    inventory_ids : SearchInventoryId or list[SearchInventoryId], optional
        Filter by inventory IDs.
    project_ids : SearchProjectId or list[SearchProjectId], optional
        Filter by project IDs.
    lot_ids : LotId or list[LotId], optional
        Filter by lot IDs.
    data_template_ids : DataTemplateId or list[DataTemplateId], optional
        Filter by data template IDs.
    data_column_ids : DataColumnId or list[DataColumnId], optional
        Filter by data column IDs.
    category : DataEntity or list[DataEntity], optional
        Filter by data entity categories.
    data_templates : str or list[str], optional
        Filter by data template names.
    data_columns : str or list[str], optional
        Filter by data column names.
    parameters : str or list[str], optional
        Filter by parameter names.
    parameter_group : str or list[str], optional
        Filter by parameter group names.
    unit : str or list[str], optional
        Filter by unit names.
    created_by : UserId or list[UserId], optional
        Filter by user IDs who created the data.
    task_created_by : UserId or list[UserId], optional
        Filter by user IDs who created the task.
    return_fields : str or list[str], optional
        Specific fields to return.
    return_facets : str or list[str], optional
        Specific facets to return.
    max_items : int, optional
        Maximum number of items to return in total. If None, fetches all available items.

    Returns
    -------
    Iterator[PropertyDataSearchItem]
        An iterator of search results matching the specified filters.
    """

    def deserialize(items: list[dict]) -> list[PropertyDataSearchItem]:
        return [PropertyDataSearchItem.model_validate(x) for x in items]

    def ensure_list(v):
        if v is None:
            return None
        return [v] if isinstance(v, str | Enum) else v

    params = {
        "result": result,
        "text": text,
        "order": order.value if order else None,
        "sortBy": sort_by,
        "inventoryIds": ensure_list(inventory_ids),
        "projectIds": ensure_list(project_ids),
        "lotIds": ensure_list(lot_ids),
        "dataTemplateId": ensure_list(data_template_ids),
        "dataColumnId": ensure_list(data_column_ids),
        "category": [c.value for c in ensure_list(category)] if category else None,
        "dataTemplates": ensure_list(data_templates),
        "dataColumns": ensure_list(data_columns),
        "parameters": ensure_list(parameters),
        "parameterGroup": ensure_list(parameter_group),
        "unit": ensure_list(unit),
        "createdBy": ensure_list(created_by),
        "taskCreatedBy": ensure_list(task_created_by),
        "returnFields": ensure_list(return_fields),
        "returnFacets": ensure_list(return_facets),
    }

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        params=params,
        max_items=max_items,
        deserialize=deserialize,
    )