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

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

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

get(
    *,
    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

def get(
    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

Bases: BaseAlbertModel

data class-attribute instance-attribute

data: list[BatchValuePatchDatum] = Field(
    default_factory=list
)

id class-attribute instance-attribute

id: BatchValueId = Field(alias='Id')

lot_id class-attribute instance-attribute

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

OrderBy

Bases: str, Enum

ASCENDING class-attribute instance-attribute

ASCENDING = 'asc'

DESCENDING class-attribute instance-attribute

DESCENDING = 'desc'