Skip to content

Batch Data

albert.collections.batch_data.BatchDataCollection 

BatchDataCollection(*, session: AlbertSession)

Bases: BaseCollection

Manage Batch Data for Batch Tasks in the Albert platform.

Batch Data is the tabular record behind a Batch Task (BatchTask): the grid that captures how a physical batch of a formulation was actually made. It is organized as:

  • Rows (BatchDataRow): the formulation components (ingredients) that go into the batch, along with nested child rows for sub-formulas.
  • Product columns (BatchDataColumn): the batch/product being manufactured, carrying batch totals, reference totals, and any lot breakdowns.
  • Values (BatchDataValue): the amount recorded for a given row within a given column.

Batch Data is keyed by the Task ID of its Batch Task (format TAS...); it is not a standalone catalog entity, so there is no free-text search. Retrieve it with get_by_id using the owning Task ID, initialize it for a task with create_batch_data, and record which lots were consumed with update_used_batch_amounts.

This collection is accessed as client.batch_data.

Example

from albert import Albert

client = Albert()
batch_data = client.batch_data.get_by_id(id="TAS123")
for row in batch_data.rows or []:
    print(row.name)

Parameters:

Name Type Description Default
session AlbertSession

The authenticated Albert session used for API calls.

 required

Attributes:

Name Type Description
base_path str

The base API route for batch data requests.

Methods:

Name Description
create_batch_data

Initialize the batch data entry for a batch task.
get_by_id

Get the batch data for a batch task by its ID.
update_used_batch_amounts

Record which lots were used for the batch's recorded amounts.

Parameters:

Name Type Description Default
session AlbertSession

The authenticated Albert session used for API calls.

 required
Source code in src/albert/collections/batch_data.py 
66
67
68
69
70
71
72
73
74
75
def __init__(self, *, session: AlbertSession):
    """Initialize a BatchDataCollection.

    Parameters
    ----------
    session : AlbertSession
        The authenticated Albert session used for API calls.
    """
    super().__init__(session=session)
    self.base_path = f"/api/{BatchDataCollection._api_version}/batchdata"

base_path 

base_path = (
    f"/api/{BatchDataCollection._api_version}/batchdata"
)

create_batch_data 

create_batch_data(*, task_id: TaskId)

Initialize the batch data entry for a batch task.

Sets up the empty batch data grid for the given Batch Task so that amounts and lots can subsequently be recorded. Retrieve the populated grid afterwards with get_by_id.

Example

from albert import Albert

client = Albert()
batch_data = client.batch_data.create_batch_data(task_id="TAS123")

Parameters:

Name Type Description Default
task_id TaskId

The Task ID of the batch task to create batch data for (format TAS...).

 required

Returns:

Type Description
BatchData

The created batch data entry.
Source code in src/albert/collections/batch_data.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
@validate_call
def create_batch_data(self, *, task_id: TaskId):
    """Initialize the batch data entry for a batch task.

    Sets up the empty batch data grid for the given Batch Task so that
    amounts and lots can subsequently be recorded. Retrieve the populated
    grid afterwards with [`get_by_id`][albert.collections.batch_data.BatchDataCollection.get_by_id].

    !!! example
        ```python
        from albert import Albert

        client = Albert()
        batch_data = client.batch_data.create_batch_data(task_id="TAS123")
        ```

    Parameters
    ----------
    task_id : TaskId
        The Task ID of the batch task to create batch data for
        (format ``TAS...``).

    Returns
    -------
    BatchData
        The created batch data entry.
    """
    url = f"{self.base_path}"
    response = self.session.post(url, json={"parentId": task_id})
    return BatchData(**response.json())

get_by_id 

get_by_id(
    *,
    id: TaskId,
    type: BatchDataType = TASK_ID,
    limit: int = 100,
    start_key: str | None = None,
    order_by: OrderBy = DESCENDING,
) -> BatchData

Get the batch data for a batch task by its ID.

Returns the batch data grid (rows, product columns, and recorded values) for the owning Batch Task (BatchTask).

Example

from albert import Albert

client = Albert()
batch_data = client.batch_data.get_by_id(id="TAS123")
batch_data.size
# 12

Parameters:

Name Type Description Default
id TaskId

The identifier to look up, of the kind given by type. By default this is the Task ID of the batch task (format TAS...).

 required
type BatchDataType

The kind of identifier passed as id. Defaults to TASK_ID.

 TASK_ID
limit int

Maximum number of row entries to return per response. Defaults to 100.

 100
start_key str

Pagination cursor identifying the first entry to evaluate; pass the last_key from a previous response to continue where it left off.

 None
order_by OrderBy

Direction in which results are sorted. Defaults to DESCENDING.

 DESCENDING

Returns:

Type Description
BatchData

The fully populated batch data.
Source code in src/albert/collections/batch_data.py 
108
109
110
111
112
113
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
@validate_call
def get_by_id(
    self,
    *,
    id: TaskId,
    type: BatchDataType = BatchDataType.TASK_ID,
    limit: int = 100,
    start_key: str | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
) -> BatchData:
    """Get the batch data for a batch task by its ID.

    Returns the batch data grid (rows, product columns, and recorded values)
    for the owning Batch Task ([`BatchTask`][albert.resources.tasks.BatchTask]).

    !!! example
        ```python
        from albert import Albert

        client = Albert()
        batch_data = client.batch_data.get_by_id(id="TAS123")
        batch_data.size
        # 12
        ```

    Parameters
    ----------
    id : TaskId
        The identifier to look up, of the kind given by ``type``. By default
        this is the Task ID of the batch task (format ``TAS...``).
    type : BatchDataType, optional
        The kind of identifier passed as ``id``. Defaults to
        [`TASK_ID`][albert.resources.batch_data.BatchDataType.TASK_ID].
    limit : int, optional
        Maximum number of row entries to return per response. Defaults to 100.
    start_key : str, optional
        Pagination cursor identifying the first entry to evaluate; pass the
        ``last_key`` from a previous response to continue where it left off.
    order_by : OrderBy, optional
        Direction in which results are sorted. Defaults to
        [`DESCENDING`][albert.core.shared.enums.OrderBy.DESCENDING].

    Returns
    -------
    BatchData
        The fully populated batch data.
    """
    params = {
        "id": id,
        "limit": limit,
        "type": type,
        "startKey": start_key,
        "orderBy": order_by,
    }
    response = self.session.get(self.base_path, params=params)
    return BatchData(**response.json())

update_used_batch_amounts 

update_used_batch_amounts(
    *,
    task_id: TaskId,
    patches: list[BatchValuePatchPayload],
) -> None

Record which lots were used for a batch task's recorded amounts.

Applies patch entries that set the lot consumed for individual cells of the batch data grid, identified by their row (and optional column). Each patch targets a value via a BatchValueId and describes the change with one or more BatchValuePatchDatum entries.

Example

from albert import Albert
from albert.resources.batch_data import (
    BatchValueId,
    BatchValuePatchDatum,
    BatchValuePatchPayload,
)

client = Albert()
patch = BatchValuePatchPayload(
    id=BatchValueId(row_id="ROW1", col_id="COL9999999"),
    data=[
        BatchValuePatchDatum(
            operation="update",
            new_value="LOT123",
            old_value="LOT100",
        )
    ],
)
client.batch_data.update_used_batch_amounts(task_id="TAS123", patches=[patch])

Parameters:

Name Type Description Default
task_id TaskId

The Task ID of the batch task to update (format TAS...).

 required
patches list[BatchValuePatchPayload]

Patch entries describing which batch values to update and the lot to assign to each.

 required

Returns:

Type Description
None
Source code in src/albert/collections/batch_data.py 
165
166
167
168
169
170
171
172
173
174
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
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
@validate_call
def update_used_batch_amounts(
    self, *, task_id: TaskId, patches: list[BatchValuePatchPayload]
) -> None:
    """Record which lots were used for a batch task's recorded amounts.

    Applies patch entries that set the lot consumed for individual cells of
    the batch data grid, identified by their row (and optional column). Each
    patch targets a value via a
    [`BatchValueId`][albert.resources.batch_data.BatchValueId] and describes the
    change with one or more
    [`BatchValuePatchDatum`][albert.resources.batch_data.BatchValuePatchDatum] entries.

    !!! example
        ```python
        from albert import Albert
        from albert.resources.batch_data import (
            BatchValueId,
            BatchValuePatchDatum,
            BatchValuePatchPayload,
        )

        client = Albert()
        patch = BatchValuePatchPayload(
            id=BatchValueId(row_id="ROW1", col_id="COL9999999"),
            data=[
                BatchValuePatchDatum(
                    operation="update",
                    new_value="LOT123",
                    old_value="LOT100",
                )
            ],
        )
        client.batch_data.update_used_batch_amounts(task_id="TAS123", patches=[patch])
        ```

    Parameters
    ----------
    task_id : TaskId
        The Task ID of the batch task to update (format ``TAS...``).
    patches : list[BatchValuePatchPayload]
        Patch entries describing which batch values to update and the lot to
        assign to each.

    Returns
    -------
    None
    """
    url = f"{self.base_path}/{task_id}/values"
    self.session.patch(
        url,
        json=[
            patch.model_dump(exclude_none=True, by_alias=True, mode="json")
            for patch in patches
        ],
    )