Batch Data

albert.collections.batch_data

TaskId module-attribute

TaskId = Annotated[str, AfterValidator(ensure_task_id)]

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

BatchData

Bases: BaseResource

id class-attribute instance-attribute

id: TaskId | None = Field(default=None, alias='albertId')

last_key class-attribute instance-attribute

last_key: str | None = Field(default=None, alias='lastKey')

product class-attribute instance-attribute

product: list[BatchDataColumn] | None = Field(
    default=None, alias="Product"
)

rows class-attribute instance-attribute

rows: list[BatchDataRow] | None = Field(
    default=None, alias="Rows"
)

size class-attribute instance-attribute

size: int | None = Field(default=None)

BatchDataCollection

BatchDataCollection(*, session: AlbertSession)

Bases: BaseCollection

BatchDataCollection is a collection class for managing BatchData entities in the Albert platform.

Parameters:

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

Methods:

Name	Description
create_batch_data	Create a new batch data entry.
get_by_id	Retrieve BatchData by ID.
update_used_batch_amounts	Update the used batch amounts for a given task ID.

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

def __init__(self, *, session: AlbertSession):
    """
    Initializes the BatchDataCollection with the provided session.

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

base_path instance-attribute

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

create_batch_data

create_batch_data(*, task_id: TaskId)

Create a new batch data entry.

Parameters:

Name	Type	Description	Default
task_id	TaskId	The ID of the task for which the batch data is being created.	required

Returns:

Type	Description
BatchData	The created BatchData object.

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

def create_batch_data(self, *, task_id: TaskId):
    """
    Create a new batch data entry.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task for which the batch data is being created.

    Returns
    -------
    BatchData
        The created BatchData object.
    """
    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

Retrieve BatchData by ID.

Parameters:

Name	Type	Description	Default
id	TaskId	Unique Id of the selected type.	required
type	BatchDataType	Type of Id for which BatchData will be fetched.	TASK_ID
limit	int	The maximum number of list entities to return.	100
start_key	str	The primary key of the first item that this operation will evaluate.	None
order_by	OrderBy	The order by which to sort the results, by default OrderBy.DESCENDING	DESCENDING

Returns:

Type	Description
BatchData	The BatchData object.

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

@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:
    """
    Retrieve BatchData by ID.

    Parameters
    ----------
    id : TaskId
        Unique Id of the selected type.
    type : BatchDataType
        Type of Id for which BatchData will be fetched.
    limit : int, optional
        The maximum number of list entities to return.
    start_key : str, optional
        The primary key of the first item that this operation will evaluate.
    order_by : OrderBy, optional
        The order by which to sort the results, by default OrderBy.DESCENDING
    Returns
    ------
    BatchData
        The BatchData object.
    """
    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: str, patches=list[BatchValuePatchPayload]
) -> None

Update the used batch amounts for a given task ID.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to update.	required
patches	list[BatchValuePatchPayload]	The patch payloads containing the data to update.	list[BatchValuePatchPayload]

Returns:

Type	Description
None	This method does not return anything.

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

def update_used_batch_amounts(
    self, *, task_id: str, patches=list[BatchValuePatchPayload]
) -> None:
    """
    Update the used batch amounts for a given task ID.

    Parameters
    ----------
    task_id : str
        The ID of the task to update.
    patches : list[BatchValuePatchPayload]
        The patch payloads containing the data to update.

    Returns
    -------
    None
        This method does not return anything.
    """
    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
        ],
    )

BatchDataType

Bases: str, Enum

TASK_ID class-attribute instance-attribute

TASK_ID = 'taskId'

BatchValuePatchPayload pydantic-model

Bases: BaseAlbertModel

Show JSON schema:

{
  "$defs": {
    "BatchValueId": {
      "properties": {
        "colId": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Colid"
        },
        "rowId": {
          "title": "Rowid",
          "type": "string"
        }
      },
      "required": [
        "rowId"
      ],
      "title": "BatchValueId",
      "type": "object"
    },
    "BatchValuePatchDatum": {
      "properties": {
        "attribute": {
          "default": "lotId",
          "title": "Attribute",
          "type": "string"
        },
        "lotId": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Lotid"
        },
        "newValue": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Newvalue"
        },
        "oldValue": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Oldvalue"
        },
        "operation": {
          "title": "Operation",
          "type": "string"
        }
      },
      "required": [
        "operation"
      ],
      "title": "BatchValuePatchDatum",
      "type": "object"
    }
  },
  "properties": {
    "Id": {
      "$ref": "#/$defs/BatchValueId"
    },
    "data": {
      "items": {
        "$ref": "#/$defs/BatchValuePatchDatum"
      },
      "title": "Data",
      "type": "array"
    },
    "lotId": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Lotid"
    }
  },
  "required": [
    "Id"
  ],
  "title": "BatchValuePatchPayload",
  "type": "object"
}

Fields:

• id (BatchValueId)
• data (list[BatchValuePatchDatum])
• lot_id (str | None)

data pydantic-field

data: list[BatchValuePatchDatum]

id pydantic-field

id: BatchValueId

lot_id pydantic-field

lot_id: str | None = None

OrderBy

Bases: str, Enum

ASCENDING class-attribute instance-attribute

ASCENDING = 'asc'

DESCENDING class-attribute instance-attribute

DESCENDING = 'desc'