Skip to content

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

AlbertSession 

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

Bases: Session

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

Parameters:

Name Type Description Default
base_url str

The base URL to prefix to all requests. (e.g., "https://sandbox.albertinvent.com")

 required
retries int

The number of retries for failed requests. Defaults to 3.

 None
client_credentials ClientCredentials | None

The client credentials for programmatic authentication. Optional if token is provided.

 None
token str | None

The JWT token for authentication. Optional if client credentials are provided.

 None

Methods:

Name Description
request
Source code in src/albert/session.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(
    self,
    *,
    base_url: str,
    token: str | None = None,
    client_credentials: ClientCredentials | None = None,
    retries: int | None = None,
):
    super().__init__()
    self.base_url = base_url
    self.headers.update(
        {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": f"albert-SDK V.{albert.__version__}",
        }
    )

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

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

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

base_url instance-attribute 

base_url = base_url

request 

request(
    method: str, path: str, *args, **kwargs
) -> Response
Source code in src/albert/session.py 
75
76
77
78
79
80
81
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 
27
28
def __init__(self, *, session: AlbertSession):
    self.session = session

session instance-attribute 

session = session

BulkPropertyData

Bases: BaseAlbertModel

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

Methods:

Name Description
from_dataframe

Converts a DataFrame to a BulkPropertyData object.

columns class-attribute instance-attribute 

columns: list[BulkPropertyDataColumn] = Field(
    default_factory=list,
    description="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 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
@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'

Bases: BaseAlbertModel

Methods:

Name Description
to_entity_link

id instance-attribute 

id: str

name class-attribute instance-attribute 

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

to_entity_link() -> EntityLink
Source code in src/albert/resources/base.py 
40
41
42
def to_entity_link(self) -> "EntityLink":
    # Convience method to return self, so you can call this method on objects that are already entity links
    return self

InventoryDataColumn

Bases: BaseAlbertModel

data_column_id class-attribute instance-attribute 

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

value class-attribute instance-attribute 

value: str | None = Field(default=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 
18
19
20
21
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'

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 
51
52
53
54
55
56
57
58
59
60
61
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 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@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 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
@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 
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
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 
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
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 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
@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 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
@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 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
@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 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@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 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
@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 
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
@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 
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
@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 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@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 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
@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

Bases: BaseAlbertModel

category instance-attribute 

category: str

created_by class-attribute instance-attribute 

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

data_template_id class-attribute instance-attribute 

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

data_template_name class-attribute instance-attribute 

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

id instance-attribute 

id: PropertyDataId

inventory_id class-attribute instance-attribute 

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

parent_id class-attribute instance-attribute 

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

project_id class-attribute instance-attribute 

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

result instance-attribute 

result: PropertyDataResult

task_id class-attribute instance-attribute 

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

workflow instance-attribute 

workflow: list[WorkflowItem]

workflow_id class-attribute instance-attribute 

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

workflow_name class-attribute instance-attribute 

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

PropertyTask

Bases: BaseTask

Represents a batch task.

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

Attributes:

Name Type Description
name str

The name of the batch task.
inventory_information list[InventoryInformation]

Information about the inventory associated with the batch task.
location SerializeAsEntityLink[Location]

The location where the batch task is performed.
parent_id str

The ID of the parent project.
blocks list[Block]

A list of blocks associated with the batch task.
id (str, optional)

The ID of the batch task, by default None.
metadata (dict[str, MetadataItem], optional)

Metadata associated with the batch task, by default an empty dictionary.
due_date (str, optional)

The due date of the batch task. YYY-MM-DD format, by default None.
notes (str, optional)

Notes associated with the batch task, by default None.
priority (TaskPriority, optional)

The priority of the batch task, by default None.
assigned_to (SerializeAsEntityLink[User], optional)

The user assigned to the batch task, by default None.
state (TaskState, optional)

The state of the batch task, by default None.
sources (list[TaskSource], optional)

A list of sources associated with the batch task, by default an empty list.
security_class (SecurityClass, optional)

The security class of the batch task, by default None.
start_date str, read only

The start date of the batch task, by default None.
claimed_date str, read only

The claimed date of the batch task, by default None.
completed_date str, read only

The completed date of the batch task, by default None.
closed_date str, read only

The closed date of the batch task, by default None.

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

Bases: BaseAlbertModel

calculation class-attribute instance-attribute 

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

data_column_unique_id class-attribute instance-attribute 

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

hidden class-attribute instance-attribute 

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

id class-attribute instance-attribute 

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

name class-attribute instance-attribute 

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

numeric_value class-attribute instance-attribute 

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

property_data class-attribute instance-attribute 

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

sequence class-attribute instance-attribute 

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

string_value class-attribute instance-attribute 

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

unit class-attribute instance-attribute 

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

value class-attribute instance-attribute 

value: str | None = Field(default=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_by_id

Retrieve a task by its ID.
get_history
list

Search for tasks matching the given criteria.
remove_block

Remove a block from a Property task.
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 
45
46
47
48
49
50
51
52
53
54
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 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@validate_call
def add_block(
    self, *, task_id: TaskId, data_template_id: DataTemplateId, workflow_id: WorkflowId
) -> None:
    """Add a block to a Property task.

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

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

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

create 

create(*, task: BaseTask) -> BaseTask

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

Parameters:

Name Type Description Default
task BaseTask

The task object to create.

 required

Returns:

Type Description
BaseTask

The registered task object.
Source code in src/albert/collections/tasks.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def create(self, *, task: BaseTask) -> BaseTask:
    """Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.

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

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

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 
206
207
208
209
210
211
212
213
214
215
216
@validate_call
def delete(self, *, id: TaskId) -> None:
    """Delete a task.

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

get_by_id 

get_by_id(*, id: TaskId) -> BaseTask

Retrieve a task by its ID.

Parameters:

Name Type Description Default
id TaskId

The ID of the task to retrieve.

 required

Returns:

Type Description
BaseTask

The task object with the provided ID.
Source code in src/albert/collections/tasks.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
@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 
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
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())

list 

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

Search for tasks matching the given criteria.

Parameters:

Name Type Description Default
order OrderBy

The order in which to return results, by default OrderBy.DESCENDING

 DESCENDING
text str | None

The text to search for, by default None

 None
sort_by str | None

The attribute to sort by, by default None

 None
tags list[str] | None

The tags to search for, by default None

 None
task_id list[str] | None

The related task IDs to search for, by default None

 None
linked_task list[str] | None

The Linked Task IDs to search for, by default None

 None
category TaskCategory | None

The category of the task to search for, by default None

 None
albert_id list[str] | None

The Albert IDs to search for, by default None

 None
data_template list[str] | None

The data template IDs to search for, by default None

 None
assigned_to list[str] | None

The User IDs to search for, by default None

 None
location list[str] | None

The Locations names to search for, by default None

 None
priority list[str] | None

The Priority levels to search for, by default None

 None
status list[str] | None

The Task Statuses to search for, by default None

 None
parameter_group list[str] | None

The related Parameter Group IDs to search for, by default None

 None
created_by list[str] | None

The User IDs of the task creators to search for, by default None

 None
project_id str | None

The Project ID to search for, by default None

 None

Yields:

Type Description
Iterator[BaseTask]

An iterator of matching Task objects.
Source code in src/albert/collections/tasks.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
def list(
    self,
    *,
    order: OrderBy = OrderBy.DESCENDING,
    text: str | None = None,
    sort_by: str | None = None,
    tags: list[str] | None = None,
    task_id: list[str] | None = None,
    linked_task: list[str] | None = None,
    category: TaskCategory | None = None,
    albert_id: list[str] | None = None,
    data_template: list[str] | None = None,
    assigned_to: list[str] | None = None,
    location: list[str] | None = None,
    priority: list[str] | None = None,
    status: list[str] | None = None,
    parameter_group: list[str] | None = None,
    created_by: list[str] | None = None,
    project_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[BaseTask]:
    """Search for tasks matching the given criteria.

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

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

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

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

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

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 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@validate_call
def remove_block(self, *, task_id: TaskId, block_id: BlockId) -> None:
    """Remove a block from a Property task.

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

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

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 
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
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 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@validate_call
def update_block_workflow(
    self, *, task_id: TaskId, block_id: BlockId, workflow_id: WorkflowId
) -> None:
    """
    Update the workflow of a specific block within a task.

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

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

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

TaskDataColumn

Bases: BaseAlbertModel

column_sequence class-attribute instance-attribute 

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

data_column_id class-attribute instance-attribute 

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

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 
271
272
273
274
275
276
277
278
@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: InventoryInformation | 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

Bases: BaseAlbertModel

data_columns class-attribute instance-attribute 

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

trial_number class-attribute instance-attribute 

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

visible_trial_number class-attribute instance-attribute 

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

void class-attribute instance-attribute 

void: bool = Field(default=False)