Property Data

albert.collections.property_data

BlockId

BlockId = Annotated[str, AfterValidator(ensure_block_id)]

DataColumnId

DataColumnId = Annotated[
    str, AfterValidator(ensure_data_column_id)
]

DataTemplateId

DataTemplateId = Annotated[
    str, AfterValidator(ensure_datatemplate_id)
]

IntervalId

IntervalId = Annotated[
    str, AfterValidator(ensure_interval_id)
]

InventoryId

InventoryId = Annotated[
    str, AfterValidator(ensure_inventory_id)
]

LotId

LotId = Annotated[str, AfterValidator(ensure_lot_id)]

SearchInventoryId

SearchInventoryId = Annotated[
    str, AfterValidator(ensure_search_inventory_id)
]

SearchProjectId

SearchProjectId = Annotated[
    str, AfterValidator(ensure_project_search_id)
]

TaskId

TaskId = Annotated[str, AfterValidator(ensure_task_id)]

UserId

UserId = Annotated[str, AfterValidator(ensure_user_id)]

BaseCollection

BaseCollection(*, session: AlbertSession)

BaseCollection is the base class for all collection classes.

Parameters:

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

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

def __init__(self, *, session: AlbertSession):
    self.session = session

session

session = session

OrderBy

Bases: str, Enum

DESCENDING

DESCENDING = 'desc'

ASCENDING

ASCENDING = 'asc'

TaskCollection

TaskCollection(*, session: AlbertSession)

Bases: BaseCollection

TaskCollection is a collection class for managing Task entities in the Albert platform.

Parameters:

Name	Type	Description	Default
session	AlbertSession	The Albert Session information	required

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

def __init__(self, *, session: AlbertSession):
    """Initialize the TaskCollection.

    Parameters
    ----------
    session : AlbertSession
        The Albert Session information
    """
    super().__init__(session=session)
    self.base_path = f"/api/{TaskCollection._api_version}/tasks"

base_path

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

create

create(*, task: BaseTask) -> BaseTask

Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.

Parameters:

Name	Type	Description	Default
task	BaseTask	The task object to create.	required

Returns:

Type	Description
BaseTask	The registered task object.

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

def create(self, *, task: BaseTask) -> BaseTask:
    """Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.

    Parameters
    ----------
    task : BaseTask
        The task object to create.

    Returns
    -------
    BaseTask
        The registered task object.
    """
    payload = [task.model_dump(mode="json", by_alias=True, exclude_none=True)]
    url = f"{self.base_path}/multi?category={task.category.value}"
    if task.parent_id is not None:
        url = f"{url}&parentId={task.parent_id}"
    response = self.session.post(url=url, json=payload)
    task_data = response.json()[0]
    return TaskAdapter.validate_python(task_data)

add_block

add_block(
    *,
    task_id: TaskId,
    data_template_id: DataTemplateId,
    workflow_id: WorkflowId,
) -> None

Add a block to a Property task.

Parameters:

Name	Type	Description	Default
task_id	TaskId	The ID of the task to add the block to.	required
data_template_id	DataTemplateId	The ID of the data template to use for the block.	required
workflow_id	WorkflowId	The ID of the workflow to assign to the block.	required

Returns:

Type	Description
None	This method does not return any value.

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

@validate_call
def add_block(
    self, *, task_id: TaskId, data_template_id: DataTemplateId, workflow_id: WorkflowId
) -> None:
    """Add a block to a Property task.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task to add the block to.
    data_template_id : DataTemplateId
        The ID of the data template to use for the block.
    workflow_id : WorkflowId
        The ID of the workflow to assign to the block.

    Returns
    -------
    None
        This method does not return any value.

    """
    url = f"{self.base_path}/{task_id}"
    payload = [
        {
            "id": task_id,
            "data": [
                {
                    "operation": "add",
                    "attribute": "Block",
                    "newValue": [{"datId": data_template_id, "Workflow": {"id": workflow_id}}],
                }
            ],
        }
    ]
    self.session.patch(url=url, json=payload)
    return None

update_block_workflow

update_block_workflow(
    *,
    task_id: TaskId,
    block_id: BlockId,
    workflow_id: WorkflowId,
) -> None

Update the workflow of a specific block within a task.

This method updates the workflow of a specified block within a task.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task.	required
block_id	str	The ID of the block within the task.	required
workflow_id	str	The ID of the new workflow to be assigned to the block.	required

Returns:

Type	Description
None	This method does not return any value.

Notes

• The method asserts that the retrieved task is an instance of PropertyTask.
• If the block's current workflow matches the new workflow ID, no update is performed.
• The method handles the case where the block has a default workflow named "No Parameter Group".

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

@validate_call
def update_block_workflow(
    self, *, task_id: TaskId, block_id: BlockId, workflow_id: WorkflowId
) -> None:
    """
    Update the workflow of a specific block within a task.

    This method updates the workflow of a specified block within a task.
    Parameters
    ----------
    task_id : str
        The ID of the task.
    block_id : str
        The ID of the block within the task.
    workflow_id : str
        The ID of the new workflow to be assigned to the block.

    Returns
    -------
    None
        This method does not return any value.

    Notes
    -----
    - The method asserts that the retrieved task is an instance of `PropertyTask`.
    - If the block's current workflow matches the new workflow ID, no update is performed.
    - The method handles the case where the block has a default workflow named "No Parameter Group".
    """
    url = f"{self.base_path}/{task_id}"
    task = self.get_by_id(id=task_id)
    if not isinstance(task, PropertyTask):
        logger.error(f"Task {task_id} is not an instance of PropertyTask")
        raise TypeError(f"Task {task_id} is not an instance of PropertyTask")
    for b in task.blocks:
        if b.id != block_id:
            continue
        for w in b.workflow:
            if w.name == "No Parameter Group" and len(b.workflow) > 1:
                # hardcoded default workflow
                continue
            existing_workflow_id = w.id
    if existing_workflow_id == workflow_id:
        logger.info(f"Block {block_id} already has workflow {workflow_id}")
        return None
    patch = [
        {
            "data": [
                {
                    "operation": "update",
                    "attribute": "workflow",
                    "oldValue": existing_workflow_id,
                    "newValue": workflow_id,
                    "blockId": block_id,
                }
            ],
            "id": task_id,
        }
    ]
    self.session.patch(url=url, json=patch)
    return None

remove_block

remove_block(*, task_id: TaskId, block_id: BlockId) -> None

Remove a block from a Property task.

Parameters:

Name	Type	Description	Default
task_id	str	ID of the Task to remove the block from (e.g., TASFOR1234)	required
block_id	str	ID of the Block to remove (e.g., BLK1)	required

Returns:

Type	Description
None

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

@validate_call
def remove_block(self, *, task_id: TaskId, block_id: BlockId) -> None:
    """Remove a block from a Property task.

    Parameters
    ----------
    task_id : str
        ID of the Task to remove the block from (e.g., TASFOR1234)
    block_id : str
        ID of the Block to remove (e.g., BLK1)

    Returns
    -------
    None
    """
    url = f"{self.base_path}/{task_id}"
    payload = [
        {
            "id": task_id,
            "data": [
                {
                    "operation": "delete",
                    "attribute": "Block",
                    "oldValue": [block_id],
                }
            ],
        }
    ]
    self.session.patch(url=url, json=payload)
    return None

delete

delete(*, id: TaskId) -> None

Delete a task.

Parameters:

Name	Type	Description	Default
id	TaskId	The ID of the task to delete.	required

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

@validate_call
def delete(self, *, id: TaskId) -> None:
    """Delete a task.

    Parameters
    ----------
    id : TaskId
        The ID of the task to delete.
    """
    url = f"{self.base_path}/{id}"
    self.session.delete(url)

get_by_id

get_by_id(*, id: TaskId) -> BaseTask

Retrieve a task by its ID.

Parameters:

Name	Type	Description	Default
id	TaskId	The ID of the task to retrieve.	required

Returns:

Type	Description
BaseTask	The task object with the provided ID.

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

@validate_call
def get_by_id(self, *, id: TaskId) -> BaseTask:
    """Retrieve a task by its ID.

    Parameters
    ----------
    id : TaskId
        The ID of the task to retrieve.

    Returns
    -------
    BaseTask
        The task object with the provided ID.
    """
    url = f"{self.base_path}/multi/{id}"
    response = self.session.get(url)
    return TaskAdapter.validate_python(response.json())

list

list(
    *,
    order: OrderBy = DESCENDING,
    text: str | None = None,
    sort_by: str | None = None,
    tags: list[str] | None = None,
    task_id: list[str] | None = None,
    linked_task: list[str] | None = None,
    category: TaskCategory | None = None,
    albert_id: list[str] | None = None,
    data_template: list[str] | None = None,
    assigned_to: list[str] | None = None,
    location: list[str] | None = None,
    priority: list[str] | None = None,
    status: list[str] | None = None,
    parameter_group: list[str] | None = None,
    created_by: list[str] | None = None,
    project_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[BaseTask]

Search for tasks matching the given criteria.

Parameters:

Name	Type	Description	Default
order	OrderBy	The order in which to return results, by default OrderBy.DESCENDING	DESCENDING
text	str | None	The text to search for, by default None	None
sort_by	str | None	The attribute to sort by, by default None	None
tags	list[str] | None	The tags to search for, by default None	None
task_id	list[str] | None	The related task IDs to search for, by default None	None
linked_task	list[str] | None	The Linked Task IDs to search for, by default None	None
category	TaskCategory | None	The category of the task to search for, by default None	None
albert_id	list[str] | None	The Albert IDs to search for, by default None	None
data_template	list[str] | None	The data template IDs to search for, by default None	None
assigned_to	list[str] | None	The User IDs to search for, by default None	None
location	list[str] | None	The Locations names to search for, by default None	None
priority	list[str] | None	The Priority levels to search for, by default None	None
status	list[str] | None	The Task Statuses to search for, by default None	None
parameter_group	list[str] | None	The related Parameter Group IDs to search for, by default None	None
created_by	list[str] | None	The User IDs of the task creators to search for, by default None	None
project_id	str | None	The Project ID to search for, by default None	None

Yields:

Type	Description
Iterator[BaseTask]	An iterator of matching Task objects.

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

def list(
    self,
    *,
    order: OrderBy = OrderBy.DESCENDING,
    text: str | None = None,
    sort_by: str | None = None,
    tags: list[str] | None = None,
    task_id: list[str] | None = None,
    linked_task: list[str] | None = None,
    category: TaskCategory | None = None,
    albert_id: list[str] | None = None,
    data_template: list[str] | None = None,
    assigned_to: list[str] | None = None,
    location: list[str] | None = None,
    priority: list[str] | None = None,
    status: list[str] | None = None,
    parameter_group: list[str] | None = None,
    created_by: list[str] | None = None,
    project_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[BaseTask]:
    """Search for tasks matching the given criteria.

    Parameters
    ----------
    order : OrderBy, optional
        The order in which to return results, by default OrderBy.DESCENDING
    text : str | None, optional
        The text to search for, by default None
    sort_by : str | None, optional
        The attribute to sort by, by default None
    tags : list[str] | None, optional
        The tags to search for, by default None
    task_id : list[str] | None, optional
        The related task IDs to search for, by default None
    linked_task : list[str] | None, optional
        The Linked Task IDs to search for, by default None
    category : TaskCategory | None, optional
        The category of the task to search for, by default None
    albert_id : list[str] | None, optional
        The Albert IDs to search for, by default None
    data_template : list[str] | None, optional
        The data template IDs to search for, by default None
    assigned_to : list[str] | None, optional
        The User IDs to search for, by default None
    location : list[str] | None, optional
        The Locations names to search for, by default None
    priority : list[str] | None, optional
        The Priority levels to search for, by default None
    status : list[str] | None, optional
        The Task Statuses to search for, by default None
    parameter_group : list[str] | None, optional
        The related Parameter Group IDs to search for, by default None
    created_by : list[str] | None, optional
        The User IDs of the task creators to search for, by default None
    project_id : str | None, optional
        The Project ID to search for, by default None

    Yields
    ------
    Iterator[BaseTask]
        An iterator of matching Task objects.
    """

    def deserialize(items: list[dict]) -> Iterator[BaseTask]:
        for item in items:
            id = item["albertId"]
            try:
                yield self.get_by_id(id=id)
            except (
                AlbertHTTPError,
                RetryError,
            ) as e:  # some legacy poorly formed Tasks raise 500s. The allowance on Retry error to also ignore these.
                logger.warning(f"Error fetching task '{id}': {e}")

    params = {
        "limit": limit,
        "offset": offset,
        "order": OrderBy(order).value if order else None,
        "text": text,
        "sortBy": sort_by,
        "tags": tags,
        "taskId": task_id,
        "linkedTask": linked_task,
        "category": category,
        "albertId": albert_id,
        "dataTemplate": data_template,
        "assignedTo": assigned_to,
        "location": location,
        "priority": priority,
        "status": status,
        "parameterGroup": parameter_group,
        "createdBy": created_by,
        "projectId": project_id,
    }

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

update

update(*, task: BaseTask) -> BaseTask

Update a task.

Parameters:

Name	Type	Description	Default
task	BaseTask	The updated Task object.	required

Returns:

Type	Description
BaseTask	The updated Task object as it exists in the Albert platform.

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

def update(self, *, task: BaseTask) -> BaseTask:
    """Update a task.

    Parameters
    ----------
    task : BaseTask
        The updated Task object.

    Returns
    -------
    BaseTask
        The updated Task object as it exists in the Albert platform.
    """
    existing = self.get_by_id(id=task.id)
    patch_payload, list_metadata_updates = self._generate_adv_patch_payload(
        updated=task, existing=existing
    )
    patch_operations = patch_payload.get("data", [])

    if len(patch_operations) == 0 and len(list_metadata_updates) == 0:
        logger.info(f"Task {task.id} is already up to date")
        return task
    path = f"{self.base_path}/{task.id}"

    for datum in patch_operations:
        patch_payload = TaskPatchPayload(data=[datum], id=task.id)
        self.session.patch(
            url=path,
            json=[patch_payload.model_dump(mode="json", by_alias=True, exclude_none=True)],
        )

    # For metadata list field updates, we clear, then update
    # since duplicate attribute values are not allowed in single patch request.
    for attribute, values in list_metadata_updates.items():
        entity_links = existing.metadata.get(attribute.split(".")[1])
        old_values = [item.id if hasattr(item, "id") else item for item in entity_links]
        clear_datum = PatchDatum(
            operation=PatchOperation.DELETE, attribute=attribute, oldValue=old_values
        )
        clear_payload = TaskPatchPayload(data=[clear_datum], id=task.id)
        self.session.patch(
            url=path,
            json=[clear_payload.model_dump(mode="json", by_alias=True, exclude_none=True)],
        )
        if values:
            update_datum = PatchDatum(
                operation=PatchOperation.UPDATE,
                attribute=attribute,
                newValue=values,
                oldValue=[],
            )

            update_payload = TaskPatchPayload(data=[update_datum], id=task.id)
            self.session.patch(
                url=path,
                json=[
                    update_payload.model_dump(mode="json", by_alias=True, exclude_none=False)
                ],
            )
    return self.get_by_id(id=task.id)

get_history

get_history(
    *,
    id: TaskId,
    order: OrderBy = DESCENDING,
    limit: int = 1000,
    entity: HistoryEntity | None = None,
    blockId: str | None = None,
    startKey: str | None = None,
) -> TaskHistory

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

def get_history(
    self,
    *,
    id: TaskId,
    order: OrderBy = OrderBy.DESCENDING,
    limit: int = 1000,
    entity: HistoryEntity | None = None,
    blockId: str | None = None,
    startKey: str | None = None,
) -> TaskHistory:
    params = {
        "limit": limit,
        "orderBy": OrderBy(order).value if order else None,
        "entity": entity,
        "blockId": blockId,
        "startKey": startKey,
    }
    url = f"{self.base_path}/{id}/history"
    response = self.session.get(url, params=params)
    return TaskHistory(**response.json())

NotFoundError

NotFoundError(response: Response)

Bases: AlbertClientError

HTTP Error due to a 404 Not Found response.

Source code in src/albert/exceptions.py

def __init__(self, response: requests.Response):
    message = self._format_message(response)
    super().__init__(message)
    self.response = response

EntityLink

Bases: BaseAlbertModel

id

id: str

name

name: str | None = Field(default=None, exclude=True)

to_entity_link

to_entity_link() -> EntityLink

Source code in src/albert/resources/base.py

def to_entity_link(self) -> "EntityLink":
    # Convience method to return self, so you can call this method on objects that are already entity links
    return self

BulkPropertyData

Bases: BaseAlbertModel

A Simple Data Structure representing all the columns of data in a block's data column.

columns

columns: list[BulkPropertyDataColumn] = Field(
    default_factory=list,
    description="The columns of data in the block's data column.",
)

from_dataframe

from_dataframe(df: DataFrame) -> BulkPropertyData

Converts a DataFrame to a BulkPropertyData object.

Parameters:

Name	Type	Description	Default
df	DataFrame	The DataFrame to convert.	required

Returns:

Type	Description
BulkPropertyData	The BulkPropertyData object that represents the data in the DataFrame.

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

@classmethod
def from_dataframe(cls, df: pd.DataFrame) -> "BulkPropertyData":
    """
    Converts a DataFrame to a BulkPropertyData object.

    Parameters
    ----------
    df : pd.DataFrame
        The DataFrame to convert.

    Returns
    -------
    BulkPropertyData
        The BulkPropertyData object that represents the data in the DataFrame.
    """
    # Convert all the values to strings, since all albert values are string typed in Albert
    df = df.fillna("").astype(str)
    columns = []
    for column in df.columns:
        data_column = BulkPropertyDataColumn(
            data_column_name=column, data_series=df[column].tolist()
        )
        columns.append(data_column)
    return BulkPropertyData(columns=columns)

CheckPropertyData

Bases: BaseResource

block_id

block_id: str | None = Field(default=None, alias='blockId')

interval_id

interval_id: str | None = Field(
    default=None, alias="interval"
)

inventory_id

inventory_id: str | None = Field(
    default=None, alias="inventoryId"
)

lot_id

lot_id: str | None = Field(default=None, alias='lotId')

data_exists

data_exists: bool | None = Field(
    default=None, alias="dataExist"
)

message

message: str | None = Field(default=None)

DataEntity

Bases: str, Enum

TASK

TASK = 'task'

WORKFLOW

WORKFLOW = 'workflow'

INVENTORY

INVENTORY = 'inventory'

InventoryDataColumn

Bases: BaseAlbertModel

data_column_id

data_column_id: DataColumnId | None = Field(
    alias="id", default=None
)

value

value: str | None = Field(default=None)

InventoryPropertyData

Bases: BaseResource

inventory_id

inventory_id: str = Field(alias='inventoryId')

inventory_name

inventory_name: str | None = Field(
    default=None, alias="inventoryName"
)

task_property_data

task_property_data: list[TaskData] = Field(
    default_factory=list, alias="Task"
)

custom_property_data

custom_property_data: list[CustomData] = Field(
    default_factory=list, alias="NoTask"
)

InventoryPropertyDataCreate

Bases: BaseResource

entity

entity: Literal[INVENTORY] = Field(default=INVENTORY)

inventory_id

inventory_id: InventoryId = Field(alias='parentId')

data_columns

data_columns: list[InventoryDataColumn] = Field(
    default_factory=list, max_length=1, alias="DataColumn"
)

status

status: PropertyDataStatus | None = Field(default=None)

PropertyDataPatchDatum

Bases: PatchDatum

property_column_id

property_column_id: DataColumnId | PropertyDataId = Field(
    alias="id"
)

PropertyDataSearchItem

Bases: BaseAlbertModel

id

id: PropertyDataId

category

category: str

workflow

workflow: list[WorkflowItem]

result

result: PropertyDataResult

data_template_id

data_template_id: DataTemplateId = Field(
    ..., alias="dataTemplateId"
)

workflow_name

workflow_name: str | None = Field(
    default=None, alias="workflowName"
)

parent_id

parent_id: TaskId | InventoryId = Field(
    ..., alias="parentId"
)

data_template_name

data_template_name: str = Field(
    ..., alias="dataTemplateName"
)

created_by

created_by: str = Field(..., alias='createdBy')

inventory_id

inventory_id: InventoryId = Field(..., alias='inventoryId')

project_id

project_id: ProjectId = Field(..., alias='projectId')

workflow_id

workflow_id: WorkflowId = Field(..., alias='workflowId')

task_id

task_id: TaskId | None = Field(default=None, alias="taskId")

PropertyValue

Bases: BaseAlbertModel

id

id: str | None = Field(default=None)

name

name: str | None = Field(default=None)

sequence

sequence: str | None = Field(default=None)

calculation

calculation: str | None = Field(default=None)

numeric_value

numeric_value: float | None = Field(
    default=None, alias="valueNumeric"
)

string_value

string_value: str | None = Field(
    default=None, alias="valueString"
)

value

value: str | None = Field(default=None)

unit

unit: SerializeAsEntityLink[Unit] | dict = Field(
    default_factory=dict, alias="Unit"
)

property_data

property_data: PropertyData | None = Field(
    default=None, alias="PropertyData"
)

data_column_unique_id

data_column_unique_id: str | None = Field(
    default=None, alias="dataColumnUniqueId"
)

hidden

hidden: bool | None = Field(default=False)

TaskDataColumn

Bases: BaseAlbertModel

data_column_id

data_column_id: DataColumnId = Field(alias='id')

column_sequence

column_sequence: str | None = Field(
    default=None, alias="columnId"
)

TaskPropertyCreate

Bases: BaseResource

Represents a task property to be created.

This class is used to create new task properties. Users can use the Workflowe.get_interval_id method to find the correct interval given the names and setpoints of the parameters.

Notes

• Users can use Workflow.get_interval_id(parameter_values={"name1":"value1", "name2":"value2"}) to find the correct interval given the names and setpoints of the parameters.
• Leave trial_number blank to create a new row/trial.
• visible_trial_number can be used to set the relative row number, allowing you to pass multiple rows of data at once.

entity

entity: Literal[TASK] = Field(
    default=TASK,
    description="The entity type, which is always `DataEntity.TASK` for task properties.",
)

interval_combination

interval_combination: str = Field(
    alias="intervalCombination",
    examples=["default", "ROW4XROW2", "ROW2"],
    default="default",
    description="The interval combination, which can be found using `Workflow.get_interval_id`.",
)

data_column

data_column: TaskDataColumn = Field(
    ...,
    alias="DataColumns",
    description="The data column associated with the task property.",
)

value

value: str | None = Field(
    default=None,
    description="The value of the task property.",
)

trial_number

trial_number: int = Field(
    alias="trialNo",
    default=None,
    description="The trial number/ row number. Leave blank to create a new row/trial.",
)

data_template

data_template: SerializeAsEntityLink[DataTemplate] = Field(
    ...,
    alias="DataTemplate",
    description="The data template associated with the task property.",
)

visible_trial_number

visible_trial_number: int | None = Field(
    alias="visibleTrialNo",
    default=None,
    description="Can be used to set the relative row number, allowing you to pass multiple rows of data at once.",
)

set_visible_trial_number

set_visible_trial_number() -> TaskPropertyCreate

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

@model_validator(mode="after")
def set_visible_trial_number(self) -> "TaskPropertyCreate":
    if self.visible_trial_number is None:
        if self.trial_number is not None:
            self.visible_trial_number = self.trial_number
        else:
            self.visible_trial_number = "1"
    return self

TaskPropertyData

Bases: BaseResource

entity

entity: Literal[TASK] = TASK

parent_id

parent_id: str = Field(..., alias='parentId')

task_id

task_id: str | None = Field(default=None, alias='id')

inventory

inventory: InventoryInformation | None = Field(
    default=None, alias="Inventory"
)

category

category: DataEntity | None = Field(default=None)

initial_workflow

initial_workflow: SerializeAsEntityLink[Workflow] | None = (
    Field(default=None, alias="InitialWorkflow")
)

finial_workflow

finial_workflow: SerializeAsEntityLink[Workflow] | None = (
    Field(default=None, alias="FinalWorkflow")
)

data_template

data_template: (
    SerializeAsEntityLink[DataTemplate] | None
) = Field(default=None, alias="DataTemplate")

data

data: list[DataInterval] = Field(
    alias="Data", frozen=True, exclude=True
)

block_id

block_id: str | None = Field(alias='blockId', default=None)

Trial

Bases: BaseAlbertModel

trial_number

trial_number: int = Field(alias='trialNo')

visible_trial_number

visible_trial_number: int = Field(
    default=1, alias="visibleTrialNo"
)

void

void: bool = Field(default=False)

data_columns

data_columns: list[PropertyValue] = Field(
    default_factory=list, alias="DataColumns"
)

PropertyTask

Bases: BaseTask

Represents a batch task.

This class is used to create and manage batch tasks. It includes the base task attributes and additional attributes specific to batch tasks.

Attributes:

Name	Type	Description
name	str	The name of the batch task.
inventory_information	list[InventoryInformation]	Information about the inventory associated with the batch task.
location	SerializeAsEntityLink[Location]	The location where the batch task is performed.
parent_id	str	The ID of the parent project.
blocks	list[Block]	A list of blocks associated with the batch task.
id	(str, optional)	The ID of the batch task, by default None.
metadata	(dict[str, MetadataItem], optional)	Metadata associated with the batch task, by default an empty dictionary.
due_date	(str, optional)	The due date of the batch task. YYY-MM-DD format, by default None.
notes	(str, optional)	Notes associated with the batch task, by default None.
priority	(TaskPriority, optional)	The priority of the batch task, by default None.
assigned_to	(SerializeAsEntityLink[User], optional)	The user assigned to the batch task, by default None.
state	(TaskState, optional)	The state of the batch task, by default None.
sources	(list[TaskSource], optional)	A list of sources associated with the batch task, by default an empty list.
security_class	(SecurityClass, optional)	The security class of the batch task, by default None.
start_date	str, read only	The start date of the batch task, by default None.
claimed_date	str, read only	The claimed date of the batch task, by default None.
completed_date	str, read only	The completed date of the batch task, by default None.
closed_date	str, read only	The closed date of the batch task, by default None.

category

category: Literal[PROPERTY] = PROPERTY

blocks

blocks: list[Block] | None = Field(
    alias="Blocks", default=None
)

qc_task

qc_task: bool | None = Field(alias='qcTask', default=None)

batch_task_id

batch_task_id: str | None = Field(
    alias="batchTaskId", default=None
)

target

target: str | None = Field(default=None)

AlbertSession

AlbertSession(
    *,
    base_url: str,
    token: str | None = None,
    client_credentials: ClientCredentials | None = None,
    retries: int | None = None,
)

Bases: Session

A session that has a base URL, which is prefixed to all request URLs.

Parameters:

Name	Type	Description	Default
base_url	str	The base URL to prefix to all requests. (e.g., "https://sandbox.albertinvent.com")	required
retries	int	The number of retries for failed requests. Defaults to 3.	None
client_credentials	ClientCredentials | None	The client credentials for programmatic authentication. Optional if token is provided.	None
token	str | None	The JWT token for authentication. Optional if client credentials are provided.	None

Source code in src/albert/session.py

def __init__(
    self,
    *,
    base_url: str,
    token: str | None = None,
    client_credentials: ClientCredentials | None = None,
    retries: int | None = None,
):
    super().__init__()
    self.base_url = base_url
    self.headers.update(
        {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": f"albert-SDK V.{albert.__version__}",
        }
    )

    if token is None and client_credentials is None:
        raise ValueError("Either client credentials or token must be specified.")

    self._provided_token = token
    self._token_manager = (
        TokenManager(base_url, client_credentials) if client_credentials is not None else None
    )

    # Set up retry logic
    retries = retries if retries is not None else 3
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=0.3,
        status_forcelist=(500, 502, 503, 504, 403),
        raise_on_status=False,
    )
    adapter = HTTPAdapter(max_retries=retry)
    self.mount("http://", adapter)
    self.mount("https://", adapter)

base_url

base_url = base_url

request

request(
    method: str, path: str, *args, **kwargs
) -> Response

Source code in src/albert/session.py

def request(self, method: str, path: str, *args, **kwargs) -> requests.Response:
    self.headers["Authorization"] = f"Bearer {self._access_token}"
    full_url = urljoin(self.base_url, path) if not path.startswith("http") else path
    with handle_http_errors():
        response = super().request(method, full_url, *args, **kwargs)
        response.raise_for_status()
        return response

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

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