Property Data

albert.collections.property_data

BlockId `module-attribute`

BlockId = Annotated[str, AfterValidator(ensure_block_id)]

DataColumnId `module-attribute`

DataColumnId = Annotated[
    str, AfterValidator(ensure_data_column_id)
]

DataTemplateId `module-attribute`

DataTemplateId = Annotated[
    str, AfterValidator(ensure_datatemplate_id)
]

IntervalId `module-attribute`

IntervalId = Annotated[
    str, AfterValidator(ensure_interval_id)
]

InventoryId `module-attribute`

InventoryId = Annotated[
    str, AfterValidator(ensure_inventory_id)
]

LotId `module-attribute`

LotId = Annotated[str, AfterValidator(ensure_lot_id)]

SearchInventoryId `module-attribute`

SearchInventoryId = Annotated[
    str, AfterValidator(ensure_search_inventory_id)
]

SearchProjectId `module-attribute`

SearchProjectId = Annotated[
    str, AfterValidator(ensure_project_search_id)
]

TaskId `module-attribute`

TaskId = Annotated[str, AfterValidator(ensure_task_id)]

UserId `module-attribute`

UserId = Annotated[str, AfterValidator(ensure_user_id)]

logger `module-attribute`

logger = create_logger()

AlbertPaginator

AlbertPaginator(
    *,
    path: str,
    mode: PaginationMode,
    session: AlbertSession,
    deserialize: Callable[
        [Iterable[dict]], Iterable[ItemType]
    ],
    params: dict[str, str] | None = None,
)

Bases: Iterator[ItemType]

Helper class for pagination through Albert endpoints.

Two pagination modes are possible: - Offset-based via by the offset query parameter - Key-based via by the startKey query parameter and 'lastKey' response field

A custom deserialize function is provided when additional logic is required to load the raw items returned by the search listing, e.g., making additional Albert API calls.

Source code in src/albert/core/pagination.py

def __init__(
    self,
    *,
    path: str,
    mode: PaginationMode,
    session: AlbertSession,
    deserialize: Callable[[Iterable[dict]], Iterable[ItemType]],
    params: dict[str, str] | None = None,
):
    self.path = path
    self.mode = mode
    self.session = session
    self.deserialize = deserialize

    params = params or {}
    self.params = {k: v for k, v in params.items() if v is not None}

    if "startKey" in self.params:
        self.params["startKey"] = quote_plus(self.params["startKey"])

    self._iterator = self._create_iterator()

deserialize `instance-attribute`

deserialize = deserialize

mode `instance-attribute`

mode = mode

params `instance-attribute`

params = {k: _wfor (k, v) in items() if v is not None}

path `instance-attribute`

path = path

session `instance-attribute`

session = session

AlbertSession

AlbertSession(
    *,
    base_url: str,
    token: str | None = None,
    auth_manager: AlbertClientCredentials
    | AlbertSSOClient
    | 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 relative request paths (e.g., "https://app.albertinvent.com").	required
`token`	`str \| None`	A static JWT token for authentication. Ignored if `auth_manager` is provided.	`None`
`auth_manager`	`AlbertClientCredentials \| AlbertSSOClient`	An authentication manager used to dynamically fetch and refresh tokens. If provided, it overrides `token`.	`None`
`retries`	`int`	The number of automatic retries on failed requests (default is 3).	`None`

Methods:

Name	Description
`request`

Source code in src/albert/core/session.py

def __init__(
    self,
    *,
    base_url: str,
    token: str | None = None,
    auth_manager: AlbertClientCredentials | AlbertSSOClient | 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 auth_manager is None:
        raise ValueError("Either `token` or `auth_manager` must be specified.")

    self._auth_manager = auth_manager
    self._provided_token = token

    # 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 `instance-attribute`

base_url = base_url

request

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

Source code in src/albert/core/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

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 `instance-attribute`

session = session

BulkPropertyData `pydantic-model`

Bases: BaseAlbertModel

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

Show JSON schema:

{
  "$defs": {
    "BulkPropertyDataColumn": {
      "description": "A Simple Data Structure representing all the rows of data in a block's data column.",
      "properties": {
        "data_column_name": {
          "default": null,
          "description": "The name of the data column (case sensitive).",
          "title": "Data Column Name",
          "type": "string"
        },
        "data_series": {
          "description": "The values, in order of row number, for the data column.",
          "items": {
            "type": "string"
          },
          "title": "Data Series",
          "type": "array"
        }
      },
      "title": "BulkPropertyDataColumn",
      "type": "object"
    }
  },
  "description": "A Simple Data Structure representing all the columns of data in a block's data column.",
  "properties": {
    "columns": {
      "description": "The columns of data in the block's data column.",
      "items": {
        "$ref": "#/$defs/BulkPropertyDataColumn"
      },
      "title": "Columns",
      "type": "array"
    }
  },
  "title": "BulkPropertyData",
  "type": "object"
}

Fields:

columns (list[BulkPropertyDataColumn])

columns `pydantic-field`

columns: list[BulkPropertyDataColumn]

The columns of data in the block's data column.

from_dataframe `classmethod`

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 `class-attribute` `instance-attribute`

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

data_exists `class-attribute` `instance-attribute`

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

interval_id `class-attribute` `instance-attribute`

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

inventory_id `class-attribute` `instance-attribute`

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

lot_id `class-attribute` `instance-attribute`

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

message `class-attribute` `instance-attribute`

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

DataEntity

Bases: str, Enum

INVENTORY `class-attribute` `instance-attribute`

INVENTORY = 'inventory'

TASK `class-attribute` `instance-attribute`

TASK = 'task'

WORKFLOW `class-attribute` `instance-attribute`

WORKFLOW = 'workflow'

InventoryDataColumn `pydantic-model`

Bases: BaseAlbertModel

Show JSON schema:

{
  "properties": {
    "id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Id"
    },
    "value": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Value"
    }
  },
  "title": "InventoryDataColumn",
  "type": "object"
}

Fields:

data_column_id (DataColumnId | None)
value (str | None)

data_column_id `pydantic-field`

data_column_id: DataColumnId | None = None

value `pydantic-field`

value: str | None = None

InventoryPropertyData

Bases: BaseResource

custom_property_data `class-attribute` `instance-attribute`

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

inventory_id `class-attribute` `instance-attribute`

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

inventory_name `class-attribute` `instance-attribute`

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

task_property_data `class-attribute` `instance-attribute`

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

InventoryPropertyDataCreate

Bases: BaseResource

data_columns `class-attribute` `instance-attribute`

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

entity `class-attribute` `instance-attribute`

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

inventory_id `class-attribute` `instance-attribute`

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

status `class-attribute` `instance-attribute`

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

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

OrderBy

Bases: str, Enum

ASCENDING `class-attribute` `instance-attribute`

ASCENDING = 'asc'

DESCENDING `class-attribute` `instance-attribute`

DESCENDING = 'desc'

PaginationMode

Bases: str, Enum

KEY `class-attribute` `instance-attribute`

KEY = 'key'

OFFSET `class-attribute` `instance-attribute`

OFFSET = 'offset'

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

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 `instance-attribute`

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

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

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)

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)

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

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

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()]

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

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

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

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

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)

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)

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)

PropertyDataPatchDatum

Bases: PatchDatum

property_column_id `class-attribute` `instance-attribute`

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

PropertyDataSearchItem `pydantic-model`

Bases: BaseAlbertModel

Show JSON schema:

{
  "$defs": {
    "PropertyDataResult": {
      "properties": {
        "valueNumeric": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Valuenumeric"
        },
        "name": {
          "title": "Name",
          "type": "string"
        },
        "id": {
          "title": "Id",
          "type": "string"
        },
        "value": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Value"
        },
        "trial": {
          "title": "Trial",
          "type": "string"
        },
        "valueString": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Valuestring"
        }
      },
      "required": [
        "name",
        "id",
        "trial"
      ],
      "title": "PropertyDataResult",
      "type": "object"
    },
    "WorkflowItem": {
      "properties": {
        "name": {
          "title": "Name",
          "type": "string"
        },
        "id": {
          "title": "Id",
          "type": "string"
        },
        "value": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Value"
        },
        "parameterGroupId": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Parametergroupid"
        },
        "valueNumeric": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Valuenumeric"
        },
        "unitName": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Unitname"
        },
        "unitId": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Unitid"
        }
      },
      "required": [
        "name",
        "id"
      ],
      "title": "WorkflowItem",
      "type": "object"
    }
  },
  "properties": {
    "id": {
      "title": "Id",
      "type": "string"
    },
    "category": {
      "title": "Category",
      "type": "string"
    },
    "workflow": {
      "items": {
        "$ref": "#/$defs/WorkflowItem"
      },
      "title": "Workflow",
      "type": "array"
    },
    "result": {
      "$ref": "#/$defs/PropertyDataResult"
    },
    "dataTemplateId": {
      "title": "Datatemplateid",
      "type": "string"
    },
    "workflowName": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Workflowname"
    },
    "parentId": {
      "title": "Parentid",
      "type": "string"
    },
    "dataTemplateName": {
      "title": "Datatemplatename",
      "type": "string"
    },
    "createdBy": {
      "title": "Createdby",
      "type": "string"
    },
    "inventoryId": {
      "title": "Inventoryid",
      "type": "string"
    },
    "projectId": {
      "title": "Projectid",
      "type": "string"
    },
    "workflowId": {
      "title": "Workflowid",
      "type": "string"
    },
    "taskId": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Taskid"
    }
  },
  "required": [
    "id",
    "category",
    "workflow",
    "result",
    "dataTemplateId",
    "parentId",
    "dataTemplateName",
    "createdBy",
    "inventoryId",
    "projectId",
    "workflowId"
  ],
  "title": "PropertyDataSearchItem",
  "type": "object"
}

Fields:

id (PropertyDataId)
category (str)
workflow (list[WorkflowItem])
result (PropertyDataResult)
data_template_id (DataTemplateId)
workflow_name (str | None)
parent_id (TaskId | InventoryId)
data_template_name (str)
created_by (str)
inventory_id (InventoryId)
project_id (ProjectId)
workflow_id (WorkflowId)
task_id (TaskId | None)

category `pydantic-field`

category: str

created_by `pydantic-field`

created_by: str

data_template_id `pydantic-field`

data_template_id: DataTemplateId

data_template_name `pydantic-field`

data_template_name: str

id `pydantic-field`

id: PropertyDataId

inventory_id `pydantic-field`

inventory_id: InventoryId

parent_id `pydantic-field`

parent_id: TaskId | InventoryId

project_id `pydantic-field`

project_id: ProjectId

result `pydantic-field`

result: PropertyDataResult

task_id `pydantic-field`

task_id: TaskId | None = None

workflow `pydantic-field`

workflow: list[WorkflowItem]

workflow_id `pydantic-field`

workflow_id: WorkflowId

workflow_name `pydantic-field`

workflow_name: str | None = None

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[TaskInventoryInformation]`	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.

batch_task_id `class-attribute` `instance-attribute`

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

blocks `class-attribute` `instance-attribute`

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

category `class-attribute` `instance-attribute`

category: Literal[PROPERTY] = PROPERTY

qc_task `class-attribute` `instance-attribute`

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

target `class-attribute` `instance-attribute`

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

PropertyValue `pydantic-model`

Bases: BaseAlbertModel

Show JSON schema:

{
  "$defs": {
    "AuditFields": {
      "description": "The audit fields for a resource",
      "properties": {
        "by": {
          "default": null,
          "title": "By",
          "type": "string"
        },
        "byName": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Byname"
        },
        "at": {
          "anyOf": [
            {
              "format": "date-time",
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "At"
        }
      },
      "title": "AuditFields",
      "type": "object"
    },
    "EntityLink": {
      "properties": {
        "id": {
          "title": "Id",
          "type": "string"
        },
        "name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        }
      },
      "required": [
        "id"
      ],
      "title": "EntityLink",
      "type": "object"
    },
    "PropertyData": {
      "properties": {
        "id": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Id"
        },
        "value": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Value"
        }
      },
      "title": "PropertyData",
      "type": "object"
    },
    "Status": {
      "description": "The status of a resource",
      "enum": [
        "active",
        "inactive"
      ],
      "title": "Status",
      "type": "string"
    },
    "Unit": {
      "description": "Unit is a Pydantic model representing a unit entity.\n\nAttributes\n----------\nid : str | None\n    The Albert ID of the unit. Set when the unit is retrieved from Albert.\nname : str\n    The name of the unit.\nsymbol : str | None\n    The symbol of the unit.\nsynonyms : List[str] | None\n    The list of synonyms for the unit.\ncategory : UnitCategory\n    The category of the unit.\nverified : bool | None\n    Whether the unit is verified.\nstatus : Status | None\n    The status of the unit. Allowed values are `active`, and `inactive`",
      "properties": {
        "status": {
          "anyOf": [
            {
              "$ref": "#/$defs/Status"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "Created": {
          "anyOf": [
            {
              "$ref": "#/$defs/AuditFields"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "Updated": {
          "anyOf": [
            {
              "$ref": "#/$defs/AuditFields"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "albertId": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Albertid"
        },
        "name": {
          "title": "Name",
          "type": "string"
        },
        "symbol": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Symbol"
        },
        "Synonyms": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "title": "Synonyms"
        },
        "category": {
          "anyOf": [
            {
              "$ref": "#/$defs/UnitCategory"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "verified": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": false,
          "title": "Verified"
        }
      },
      "required": [
        "name"
      ],
      "title": "Unit",
      "type": "object"
    },
    "UnitCategory": {
      "description": "UnitCategory is an enumeration of possible unit categories.\n\nAttributes\n----------\nLENGTH : str\n    Represents length units.\nVOLUME : str\n    Represents volume units.\nLIQUID_VOLUME : str\n    Represents liquid volume units.\nANGLES : str\n    Represents angle units.\nTIME : str\n    Represents time units.\nFREQUENCY : str\n    Represents frequency units.\nMASS : str\n    Represents mass units.\nCURRENT : str\n    Represents electric current units.\nTEMPERATURE : str\n    Represents temperature units.\nAMOUNT : str\n    Represents amount of substance units.\nLUMINOSITY : str\n    Represents luminous intensity units.\nFORCE : str\n    Represents force units.\nENERGY : str\n    Represents energy units.\nPOWER : str\n    Represents power units.\nPRESSURE : str\n    Represents pressure units.\nELECTRICITY_AND_MAGNETISM : str\n    Represents electricity and magnetism units.\nOTHER : str\n    Represents other units.\nWEIGHT : str\n    Represents weight units.",
      "enum": [
        "Length",
        "Volume",
        "Liquid volume",
        "Angles",
        "Time",
        "Frequency",
        "Mass",
        "Electric current",
        "Temperature",
        "Amount of substance",
        "Luminous intensity",
        "Force",
        "Energy",
        "Power",
        "Pressure",
        "Electricity and magnetism",
        "Other",
        "Weight",
        "Area",
        "Surface Area",
        "Binary",
        "Capacitance",
        "Speed",
        "Electrical conductivity",
        "Electrical permitivitty",
        "Density",
        "Resistance"
      ],
      "title": "UnitCategory",
      "type": "string"
    }
  },
  "properties": {
    "id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Id"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "sequence": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Sequence"
    },
    "calculation": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Calculation"
    },
    "valueNumeric": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Valuenumeric"
    },
    "valueString": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Valuestring"
    },
    "value": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Value"
    },
    "Unit": {
      "anyOf": [
        {
          "$ref": "#/$defs/Unit"
        },
        {
          "$ref": "#/$defs/EntityLink"
        },
        {
          "additionalProperties": true,
          "type": "object"
        }
      ],
      "title": "Unit"
    },
    "PropertyData": {
      "anyOf": [
        {
          "$ref": "#/$defs/PropertyData"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "dataColumnUniqueId": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Datacolumnuniqueid"
    },
    "hidden": {
      "anyOf": [
        {
          "type": "boolean"
        },
        {
          "type": "null"
        }
      ],
      "default": false,
      "title": "Hidden"
    }
  },
  "title": "PropertyValue",
  "type": "object"
}

Fields:

id (str | None)
name (str | None)
sequence (str | None)
calculation (str | None)
numeric_value (float | None)
string_value (str | None)
value (str | None)
unit (SerializeAsEntityLink[Unit] | dict)
property_data (PropertyData | None)
data_column_unique_id (str | None)
hidden (bool | None)

calculation `pydantic-field`

calculation: str | None = None

data_column_unique_id `pydantic-field`

data_column_unique_id: str | None = None

hidden `pydantic-field`

hidden: bool | None = False

id `pydantic-field`

id: str | None = None

name `pydantic-field`

name: str | None = None

numeric_value `pydantic-field`

numeric_value: float | None = None

property_data `pydantic-field`

property_data: PropertyData | None = None

sequence `pydantic-field`

sequence: str | None = None

string_value `pydantic-field`

string_value: str | None = None

unit `pydantic-field`

unit: SerializeAsEntityLink[Unit] | dict

value `pydantic-field`

value: str | None = None

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

Methods:

Name	Description
`add_block`	Add a block to a Property task.
`create`	Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.
`delete`	Delete a task.
`get_all`	Retrieve fully hydrated Task entities with optional filters.
`get_by_id`	Retrieve a task by its ID.
`get_history`
`remove_block`	Remove a block from a Property task.
`search`	Search for Task matching the provided criteria.
`update`	Update a task.
`update_block_workflow`	Update the workflow of a specific block within a task.

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 `instance-attribute`

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

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)

create

create(
    *, task: PropertyTask | GeneralTask | BatchTask
) -> 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`	`PropertyTask \| GeneralTask \| BatchTask`	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: PropertyTask | GeneralTask | BatchTask) -> BaseTask:
    """Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.

    Parameters
    ----------
    task : PropertyTask | GeneralTask | BatchTask
        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)

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_all

get_all(
    *, params: TaskFilterParams | None = None
) -> Iterator[BaseTask]

Retrieve fully hydrated Task entities with optional filters.

This method returns complete entity data using get_by_id or get_by_ids. Use :meth:search for faster retrieval when you only need lightweight, partial (unhydrated) entities.

Parameters:

Name	Type	Description	Default
`params`	`TaskFilterParams`	Filter and pagination options passed to the search query.	`None`

Yields:

Type	Description
`Iterator[BaseTask]`	A stream of fully hydrated Task objects (PropertyTask, BatchTask, or GeneralTask).

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

def get_all(self, *, params: TaskFilterParams | None = None) -> Iterator[BaseTask]:
    """Retrieve fully hydrated Task entities with optional filters.

    This method returns complete entity data using `get_by_id` or `get_by_ids`.
    Use :meth:`search` for faster retrieval when you only need lightweight, partial (unhydrated) entities.

    Parameters
    ----------
    params : TaskFilterParams, optional
        Filter and pagination options passed to the search query.

    Yields
    ------
    Iterator[BaseTask]
        A stream of fully hydrated Task objects (PropertyTask, BatchTask, or GeneralTask).
    """
    params = params or TaskFilterParams()

    for task in self.search(params=params):
        task_id = getattr(task, "id", None)
        if not task_id:
            continue

        try:
            yield self.get_by_id(id=task_id)
        except (AlbertHTTPError, RetryError) as e:
            logger.warning(f"Error fetching task '{id}': {e}")

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

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

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)

search

search(
    *, params: TaskFilterParams | None = None
) -> Iterator[TaskSearchItem]

Search for Task matching the provided criteria.

⚠️ This method returns partial (unhydrated) entities to optimize performance. To retrieve fully detailed entities, use :meth:get_all instead.

Parameters:

Name	Type	Description	Default
`params`	`TaskFilterParams`	Structured query parameters including filters, sort order, and pagination.	`None`

Yields:

Type	Description
`Iterator[BaseTask]`	An iterator of matching, fully hydrated Task objects.

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

def search(self, *, params: TaskFilterParams | None = None) -> Iterator[TaskSearchItem]:
    """Search for Task matching the provided criteria.

    ⚠️ This method returns partial (unhydrated) entities to optimize performance.
    To retrieve fully detailed entities, use :meth:`get_all` instead.

    Parameters
    ----------
    params : TaskFilterParams, optional
        Structured query parameters including filters, sort order, and pagination.

    Yields
    ------
    Iterator[BaseTask]
        An iterator of matching, fully hydrated Task objects.
    """
    params = params or TaskFilterParams()

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

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        deserialize=lambda items: [
            TaskSearchItem(**item)._bind_collection(self) for item in items
        ],
        params=query_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)

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)

TaskDataColumn `pydantic-model`

Bases: BaseAlbertModel

Show JSON schema:

{
  "properties": {
    "id": {
      "title": "Id",
      "type": "string"
    },
    "columnId": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Columnid"
    }
  },
  "required": [
    "id"
  ],
  "title": "TaskDataColumn",
  "type": "object"
}

Fields:

data_column_id (DataColumnId)
column_sequence (str | None)

column_sequence `pydantic-field`

column_sequence: str | None = None

data_column_id `pydantic-field`

data_column_id: DataColumnId

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.

Methods:

Name	Description
`set_visible_trial_number`

data_column `class-attribute` `instance-attribute`

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

data_template `class-attribute` `instance-attribute`

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

entity `class-attribute` `instance-attribute`

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

interval_combination `class-attribute` `instance-attribute`

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`.",
)

trial_number `class-attribute` `instance-attribute`

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

value `class-attribute` `instance-attribute`

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

visible_trial_number `class-attribute` `instance-attribute`

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

block_id `class-attribute` `instance-attribute`

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

category `class-attribute` `instance-attribute`

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

data `class-attribute` `instance-attribute`

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

data_template `class-attribute` `instance-attribute`

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

entity `class-attribute` `instance-attribute`

entity: Literal[TASK] = TASK

finial_workflow `class-attribute` `instance-attribute`

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

initial_workflow `class-attribute` `instance-attribute`

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

inventory `class-attribute` `instance-attribute`

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

parent_id `class-attribute` `instance-attribute`

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

task_id `class-attribute` `instance-attribute`

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

Trial `pydantic-model`

Bases: BaseAlbertModel

Show JSON schema:

{
  "$defs": {
    "AuditFields": {
      "description": "The audit fields for a resource",
      "properties": {
        "by": {
          "default": null,
          "title": "By",
          "type": "string"
        },
        "byName": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Byname"
        },
        "at": {
          "anyOf": [
            {
              "format": "date-time",
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "At"
        }
      },
      "title": "AuditFields",
      "type": "object"
    },
    "EntityLink": {
      "properties": {
        "id": {
          "title": "Id",
          "type": "string"
        },
        "name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        }
      },
      "required": [
        "id"
      ],
      "title": "EntityLink",
      "type": "object"
    },
    "PropertyData": {
      "properties": {
        "id": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Id"
        },
        "value": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Value"
        }
      },
      "title": "PropertyData",
      "type": "object"
    },
    "PropertyValue": {
      "properties": {
        "id": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Id"
        },
        "name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "sequence": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Sequence"
        },
        "calculation": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Calculation"
        },
        "valueNumeric": {
          "anyOf": [
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Valuenumeric"
        },
        "valueString": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Valuestring"
        },
        "value": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Value"
        },
        "Unit": {
          "anyOf": [
            {
              "$ref": "#/$defs/Unit"
            },
            {
              "$ref": "#/$defs/EntityLink"
            },
            {
              "additionalProperties": true,
              "type": "object"
            }
          ],
          "title": "Unit"
        },
        "PropertyData": {
          "anyOf": [
            {
              "$ref": "#/$defs/PropertyData"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "dataColumnUniqueId": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Datacolumnuniqueid"
        },
        "hidden": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": false,
          "title": "Hidden"
        }
      },
      "title": "PropertyValue",
      "type": "object"
    },
    "Status": {
      "description": "The status of a resource",
      "enum": [
        "active",
        "inactive"
      ],
      "title": "Status",
      "type": "string"
    },
    "Unit": {
      "description": "Unit is a Pydantic model representing a unit entity.\n\nAttributes\n----------\nid : str | None\n    The Albert ID of the unit. Set when the unit is retrieved from Albert.\nname : str\n    The name of the unit.\nsymbol : str | None\n    The symbol of the unit.\nsynonyms : List[str] | None\n    The list of synonyms for the unit.\ncategory : UnitCategory\n    The category of the unit.\nverified : bool | None\n    Whether the unit is verified.\nstatus : Status | None\n    The status of the unit. Allowed values are `active`, and `inactive`",
      "properties": {
        "status": {
          "anyOf": [
            {
              "$ref": "#/$defs/Status"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "Created": {
          "anyOf": [
            {
              "$ref": "#/$defs/AuditFields"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "Updated": {
          "anyOf": [
            {
              "$ref": "#/$defs/AuditFields"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "albertId": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Albertid"
        },
        "name": {
          "title": "Name",
          "type": "string"
        },
        "symbol": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Symbol"
        },
        "Synonyms": {
          "anyOf": [
            {
              "items": {
                "type": "string"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "title": "Synonyms"
        },
        "category": {
          "anyOf": [
            {
              "$ref": "#/$defs/UnitCategory"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "verified": {
          "anyOf": [
            {
              "type": "boolean"
            },
            {
              "type": "null"
            }
          ],
          "default": false,
          "title": "Verified"
        }
      },
      "required": [
        "name"
      ],
      "title": "Unit",
      "type": "object"
    },
    "UnitCategory": {
      "description": "UnitCategory is an enumeration of possible unit categories.\n\nAttributes\n----------\nLENGTH : str\n    Represents length units.\nVOLUME : str\n    Represents volume units.\nLIQUID_VOLUME : str\n    Represents liquid volume units.\nANGLES : str\n    Represents angle units.\nTIME : str\n    Represents time units.\nFREQUENCY : str\n    Represents frequency units.\nMASS : str\n    Represents mass units.\nCURRENT : str\n    Represents electric current units.\nTEMPERATURE : str\n    Represents temperature units.\nAMOUNT : str\n    Represents amount of substance units.\nLUMINOSITY : str\n    Represents luminous intensity units.\nFORCE : str\n    Represents force units.\nENERGY : str\n    Represents energy units.\nPOWER : str\n    Represents power units.\nPRESSURE : str\n    Represents pressure units.\nELECTRICITY_AND_MAGNETISM : str\n    Represents electricity and magnetism units.\nOTHER : str\n    Represents other units.\nWEIGHT : str\n    Represents weight units.",
      "enum": [
        "Length",
        "Volume",
        "Liquid volume",
        "Angles",
        "Time",
        "Frequency",
        "Mass",
        "Electric current",
        "Temperature",
        "Amount of substance",
        "Luminous intensity",
        "Force",
        "Energy",
        "Power",
        "Pressure",
        "Electricity and magnetism",
        "Other",
        "Weight",
        "Area",
        "Surface Area",
        "Binary",
        "Capacitance",
        "Speed",
        "Electrical conductivity",
        "Electrical permitivitty",
        "Density",
        "Resistance"
      ],
      "title": "UnitCategory",
      "type": "string"
    }
  },
  "properties": {
    "trialNo": {
      "title": "Trialno",
      "type": "integer"
    },
    "visibleTrialNo": {
      "default": 1,
      "title": "Visibletrialno",
      "type": "integer"
    },
    "void": {
      "default": false,
      "title": "Void",
      "type": "boolean"
    },
    "DataColumns": {
      "items": {
        "$ref": "#/$defs/PropertyValue"
      },
      "title": "Datacolumns",
      "type": "array"
    }
  },
  "required": [
    "trialNo"
  ],
  "title": "Trial",
  "type": "object"
}

Fields:

trial_number (int)
visible_trial_number (int)
void (bool)
data_columns (list[PropertyValue])

data_columns `pydantic-field`

data_columns: list[PropertyValue]

trial_number `pydantic-field`

trial_number: int

visible_trial_number `pydantic-field`

visible_trial_number: int = 1

void `pydantic-field`

void: bool = False

Property Data

albert.collections.property_data

BlockId module-attribute

DataColumnId module-attribute

DataTemplateId module-attribute

IntervalId module-attribute

InventoryId module-attribute

LotId module-attribute

SearchInventoryId module-attribute

SearchProjectId module-attribute

TaskId module-attribute

UserId module-attribute

logger module-attribute

AlbertPaginator

deserialize instance-attribute

mode instance-attribute

params instance-attribute

path instance-attribute

session instance-attribute

AlbertSession

base_url instance-attribute

request

BaseCollection

session instance-attribute

BulkPropertyData pydantic-model

columns pydantic-field

from_dataframe classmethod

CheckPropertyData

block_id class-attribute instance-attribute

data_exists class-attribute instance-attribute

interval_id class-attribute instance-attribute

inventory_id class-attribute instance-attribute

lot_id class-attribute instance-attribute

message class-attribute instance-attribute

DataEntity

INVENTORY class-attribute instance-attribute

TASK class-attribute instance-attribute

WORKFLOW class-attribute instance-attribute

InventoryDataColumn pydantic-model

data_column_id pydantic-field

value pydantic-field

InventoryPropertyData

custom_property_data class-attribute instance-attribute

inventory_id class-attribute instance-attribute

inventory_name class-attribute instance-attribute

task_property_data class-attribute instance-attribute

InventoryPropertyDataCreate

data_columns class-attribute instance-attribute

entity class-attribute instance-attribute

inventory_id class-attribute instance-attribute

status class-attribute instance-attribute

NotFoundError

OrderBy

ASCENDING class-attribute instance-attribute

DESCENDING class-attribute instance-attribute

PaginationMode

KEY class-attribute instance-attribute

OFFSET class-attribute instance-attribute

PropertyDataCollection

base_path instance-attribute

add_properties_to_inventory

add_properties_to_task

bulk_delete_task_data

bulk_load_task_properties

check_block_interval_for_data

check_for_task_data

get_all_task_properties

get_properties_on_inventory

get_task_block_properties

search

update_or_create_task_properties

update_property_on_inventory

update_property_on_task

PropertyDataPatchDatum

property_column_id class-attribute instance-attribute

PropertyDataSearchItem pydantic-model

category pydantic-field

created_by pydantic-field

data_template_id pydantic-field

data_template_name pydantic-field

BlockId `module-attribute`

DataColumnId `module-attribute`

DataTemplateId `module-attribute`

IntervalId `module-attribute`

InventoryId `module-attribute`

LotId `module-attribute`

SearchInventoryId `module-attribute`

SearchProjectId `module-attribute`

TaskId `module-attribute`

UserId `module-attribute`

logger `module-attribute`

deserialize `instance-attribute`

mode `instance-attribute`

params `instance-attribute`

path `instance-attribute`

session `instance-attribute`

base_url `instance-attribute`

session `instance-attribute`

BulkPropertyData `pydantic-model`

columns `pydantic-field`

from_dataframe `classmethod`

block_id `class-attribute` `instance-attribute`

data_exists `class-attribute` `instance-attribute`

interval_id `class-attribute` `instance-attribute`

inventory_id `class-attribute` `instance-attribute`

lot_id `class-attribute` `instance-attribute`

message `class-attribute` `instance-attribute`

INVENTORY `class-attribute` `instance-attribute`

TASK `class-attribute` `instance-attribute`

WORKFLOW `class-attribute` `instance-attribute`

InventoryDataColumn `pydantic-model`

data_column_id `pydantic-field`

value `pydantic-field`

custom_property_data `class-attribute` `instance-attribute`

inventory_id `class-attribute` `instance-attribute`

inventory_name `class-attribute` `instance-attribute`

task_property_data `class-attribute` `instance-attribute`

data_columns `class-attribute` `instance-attribute`

entity `class-attribute` `instance-attribute`

inventory_id `class-attribute` `instance-attribute`

status `class-attribute` `instance-attribute`

ASCENDING `class-attribute` `instance-attribute`

DESCENDING `class-attribute` `instance-attribute`

KEY `class-attribute` `instance-attribute`

OFFSET `class-attribute` `instance-attribute`

base_path `instance-attribute`

property_column_id `class-attribute` `instance-attribute`

PropertyDataSearchItem `pydantic-model`

category `pydantic-field`

created_by `pydantic-field`

data_template_id `pydantic-field`

data_template_name `pydantic-field`

id `pydantic-field`

inventory_id `pydantic-field`

parent_id `pydantic-field`

project_id `pydantic-field`

result `pydantic-field`

task_id `pydantic-field`

workflow `pydantic-field`

workflow_id `pydantic-field`

workflow_name `pydantic-field`

batch_task_id `class-attribute` `instance-attribute`

blocks `class-attribute` `instance-attribute`

category `class-attribute` `instance-attribute`

qc_task `class-attribute` `instance-attribute`

target `class-attribute` `instance-attribute`

PropertyValue `pydantic-model`

calculation `pydantic-field`

data_column_unique_id `pydantic-field`

hidden `pydantic-field`

id `pydantic-field`

name `pydantic-field`

numeric_value `pydantic-field`

property_data `pydantic-field`

sequence `pydantic-field`

string_value `pydantic-field`

unit `pydantic-field`

value `pydantic-field`

base_path `instance-attribute`

TaskDataColumn `pydantic-model`