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	Returns all the properties for a specific 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 objects 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 objects 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
) -> list[TaskPropertyData]

Returns all the properties for a specific task.

Parameters:

Name	Type	Description	Default
task_id	TaskId	The ID of the task to retrieve properties for.	required

Returns:

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

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

@validate_call
def get_all_task_properties(self, *, task_id: TaskId) -> list[TaskPropertyData]:
    """Returns all the properties for a specific task.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task to retrieve properties for.

    Returns
    -------
    list[TaskPropertyData]
        A list of TaskPropertyData objects representing the properties within the task.
    """
    all_info = []
    task_data_info = self.check_for_task_data(task_id=task_id)
    for combo_info in task_data_info:
        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],
) -> 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

Returns:

Type	Description
list[TaskPropertyData]	A list of TaskPropertyData objects 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]
) -> 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.

    Returns
    -------
    list[TaskPropertyData]
        A list of TaskPropertyData objects 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.get_all_task_properties(task_id=task_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],
)

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 objects representing the properties to add.	required

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],
):
    """
    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 objects representing the properties to add.

    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)
    else:
        return self.get_all_task_properties(task_id=task_id)

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],
) -> 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 objects representing the properties to update or create.	required

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],
) -> 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 objects representing the properties to update or create.

    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:
        self.update_property_on_task(task_id=task_id, patch_payload=all_patches)
        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,
        )
    else:
        return self.update_property_on_task(task_id=task_id, patch_payload=all_patches)

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,
) -> 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

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,
) -> 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().

    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,
    )

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(
    *,
    limit: int = 100,
    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,
) -> Iterator[PropertyDataSearchItem]

Search for property data with various filtering options.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of results to return.	100
result	str	Find results using search syntax. e.g. to find all results with viscosity < 200 at a temperature of 25 we would do result=viscosity(<200)@temperature(25)	None
text	str	Free text search across all searchable fields.	None
order	OrderBy	Sort order (ascending/descending).	None
sort_by	str	Field to sort results by.	None
inventory_ids	SearchInventoryIdType or list of SearchInventoryIdType	Filter by inventory IDs.	None
project_ids	ProjectIdType or list of ProjectIdType	Filter by project IDs.	None
lot_ids	LotIdType or list of LotIdType	Filter by lot IDs.	None
data_template_ids	DataTemplateId or list of DataTemplateId	Filter by data template IDs.	None
data_column_ids	DataColumnId | list[DataColumnId] | None	Filter by data column IDs.	None
category	DataEntity or list of DataEntity	Filter by data entity categories.	None
data_templates	str or list of str (exact match)	Filter by data template names.	None
data_columns	str or list of str (exact match)	Filter by data column names (currently non-functional).	None
parameters	str or list of str (exact match)	Filter by parameter names.	None
parameter_group	str or list of str (exact match)	Filter by parameter group names.	None
unit	str or list of str (exact match)	Filter by unit names.	None
created_by	UserIdType or list of UserIdType	Filter by creator user IDs.	None
task_created_by	UserIdType or list of UserIdType	Filter by task creator user IDs.	None
return_fields	str or list of str	Specific fields to include in results. If None, returns all fields.	None
return_facets	str or list of str	Specific facets to include in results.	None

Returns:

Type	Description
dict	Search results matching the specified criteria.

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

@validate_call
def search(
    self,
    *,
    limit: int = 100,
    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,
) -> Iterator[PropertyDataSearchItem]:
    """Search for property data with various filtering options.

    Parameters
    ----------
    limit : int, default=100
        Maximum number of results to return.
    result : str, optional
        Find results using search syntax. e.g. to find all results with viscosity < 200 at a temperature of 25 we would do
        result=viscosity(<200)@temperature(25)
    text : str, optional
        Free text search across all searchable fields.
    order : OrderBy, optional
        Sort order (ascending/descending).
    sort_by : str, optional
        Field to sort results by.
    inventory_ids : SearchInventoryIdType or list of SearchInventoryIdType, optional
        Filter by inventory IDs.
    project_ids : ProjectIdType or list of ProjectIdType, optional
        Filter by project IDs.
    lot_ids : LotIdType or list of LotIdType, optional
        Filter by lot IDs.
    data_template_ids : DataTemplateId or list of DataTemplateId, optional
        Filter by data template IDs.
    data_column_ids: DataColumnId or list of DataColumnId, optional
        Filter by data column IDs.
    category : DataEntity or list of DataEntity, optional
        Filter by data entity categories.
    data_templates : str or list of str (exact match), optional
        Filter by data template names.
    data_columns : str or list of str (exact match), optional
        Filter by data column names (currently non-functional).
    parameters : str or list of str (exact match), optional
        Filter by parameter names.
    parameter_group : str or list of str (exact match), optional
        Filter by parameter group names.
    unit : str or list of str (exact match), optional
        Filter by unit names.
    created_by : UserIdType or list of UserIdType, optional
        Filter by creator user IDs.
    task_created_by : UserIdType or list of UserIdType, optional
        Filter by task creator user IDs.
    return_fields : str or list of str, optional
        Specific fields to include in results. If None, returns all fields.
    return_facets : str or list of str, optional
        Specific facets to include in results.

    Returns
    -------
    dict
        Search results matching the specified criteria.
    """

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

    if isinstance(inventory_ids, str):
        inventory_ids = [inventory_ids]
    if isinstance(project_ids, str):
        project_ids = [project_ids]
    if isinstance(lot_ids, str):
        lot_ids = [lot_ids]
    if isinstance(data_template_ids, str):
        data_template_ids = [data_template_ids]
    if isinstance(data_column_ids, str):
        data_column_ids = [data_column_ids]
    if isinstance(category, DataEntity):
        category = [category]
    if isinstance(data_templates, str):
        data_templates = [data_templates]
    if isinstance(data_columns, str):
        data_columns = [data_columns]
    if isinstance(parameters, str):
        parameters = [parameters]
    if isinstance(parameter_group, str):
        parameter_group = [parameter_group]
    if isinstance(unit, str):
        unit = [unit]
    if isinstance(created_by, str):
        created_by = [created_by]
    if isinstance(task_created_by, str):
        task_created_by = [task_created_by]
    if isinstance(return_fields, str):
        return_fields = [return_fields]
    if isinstance(return_facets, str):
        return_facets = [return_facets]

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

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