Lots

albert.collections.lots.LotCollection

LotCollection(*, session: AlbertSession)

Bases: BaseCollection

LotCollection is a collection class for managing Lot entities in the Albert platform.

Parameters:

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

Methods:

Name	Description
create	Create new lots.
get_by_id	Get a lot by its ID.
get_by_ids	Get a list of lots by their IDs.
delete	Delete a lot by its ID.
search	Search for Lot records matching the provided filters.
get_all	Get all Lot entities with optional filters.
update	Update a lot.

Attributes:

Name	Type	Description
base_path

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

def __init__(self, *, session: AlbertSession):
    """A collection for interacting with Lots in Albert.

    Parameters
    ----------
    session : AlbertSession
        An Albert session instance.
    """
    super().__init__(session=session)
    self.base_path = f"/api/{LotCollection._api_version}/lots"

base_path

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

create

create(*, lots: list[Lot]) -> list[Lot]

Create new lots.

Parameters:

Name	Type	Description	Default
lots	list[Lot]	A list of Lot entities to create.	required

Returns:

Type	Description
list[Lot]	A list of created Lot entities.

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

def create(self, *, lots: list[Lot]) -> list[Lot]:
    """Create new lots.

    Parameters
    ----------
    lots : list[Lot]
        A list of Lot entities to create.

    Returns
    -------
    list[Lot]
        A list of created Lot entities.
    """
    payload = [lot.model_dump(by_alias=True, exclude_none=True, mode="json") for lot in lots]
    response = self.session.post(self.base_path, json=payload)
    data = response.json()

    if isinstance(data, list):
        created_raw, failed = data, []
    else:
        created_raw = data.get("CreatedLots") or data.get("CreatedItems") or []
        failed = data.get("FailedItems") or []

    if (response.status_code == 206 or failed) and failed:
        logger.warning("Partial success creating lots", extra={"failed": failed})

    return [Lot(**lot) for lot in created_raw]

get_by_id

get_by_id(*, id: LotId) -> Lot

Get a lot by its ID.

Parameters:

Name	Type	Description	Default
id	str	The ID of the lot to get.	required

Returns:

Type	Description
Lot	The lot with the provided ID.

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

@validate_call
def get_by_id(self, *, id: LotId) -> Lot:
    """Get a lot by its ID.

    Parameters
    ----------
    id : str
        The ID of the lot to get.

    Returns
    -------
    Lot
        The lot with the provided ID.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return Lot(**response.json())

get_by_ids

get_by_ids(*, ids: list[LotId]) -> list[Lot]

Get a list of lots by their IDs.

Parameters:

Name	Type	Description	Default
ids	list[str]	A list of lot IDs to get.	required

Returns:

Type	Description
list[Lot]	A list of lots with the provided IDs.

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

@validate_call
def get_by_ids(self, *, ids: list[LotId]) -> list[Lot]:
    """Get a list of lots by their IDs.

    Parameters
    ----------
    ids : list[str]
        A list of lot IDs to get.

    Returns
    -------
    list[Lot]
        A list of lots with the provided IDs.
    """
    url = f"{self.base_path}/ids"
    response = self.session.get(url, params={"id": ids})
    return [Lot(**lot) for lot in response.json()["Items"]]

delete

delete(*, id: LotId) -> None

Delete a lot by its ID.

Parameters:

Name	Type	Description	Default
id	str	The ID of the lot to delete.	required

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

@validate_call
def delete(self, *, id: LotId) -> None:
    """Delete a lot by its ID.

    Parameters
    ----------
    id : str
        The ID of the lot to delete.
    """
    url = f"{self.base_path}?id={id}"
    self.session.delete(url)

search

search(
    *,
    text: str | None = None,
    inventory_id: InventoryId
    | list[InventoryId]
    | None = None,
    location_id: str | list[str] | None = None,
    storage_location_id: str | list[str] | None = None,
    task_id: TaskId | list[TaskId] | None = None,
    category: InventoryCategory
    | str
    | list[InventoryCategory | str]
    | None = None,
    external_barcode_id: str | list[str] | None = None,
    search_field: str | list[str] | None = None,
    source_field: str | list[str] | None = None,
    additional_field: str | list[str] | None = None,
    is_drop_down: bool | None = None,
    order_by: OrderBy = DESCENDING,
    sort_by: str | None = None,
    offset: int | None = None,
    max_items: int | None = None,
) -> Iterator[LotSearchItem]

Search for Lot records matching the provided filters.

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

Parameters:

Name	Type	Description	Default
text	str	Free-text query matched against lot fields.	None
inventory_id	InventoryId or list[InventoryId]	Filter by parent inventory IDs.	None
location_id	str or list[str]	Filter by specific location IDs.	None
storage_location_id	str or list[str]	Filter by storage location IDs.	None
task_id	TaskId or list[TaskId]	Filter by source task IDs.	None
category	InventoryCategory or list[str]	Filter by parent inventory categories.	None
external_barcode_id	str or list[str]	Filter by external barcode IDs.	None
search_field	str or list[str]	Restrict the fields the text query searches.	None
source_field	str or list[str]	Restrict which fields are returned in the response.	None
additional_field	str or list[str]	Request additional columns from the search index.	None
is_drop_down	bool	Use dropdown sanitization for the search text when True.	None
order_by	OrderBy	Sort order for the results, default DESCENDING.	DESCENDING
sort_by	str	Attribute to sort by.	None
offset	int	Pagination offset to start from.	None
max_items	int	Maximum number of items to return in total. If None, fetches all available items.	None

Returns:

Type	Description
Iterator[LotSearchItem]	An iterator of matching partial (unhydrated) lot entities.

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

@validate_call
def search(
    self,
    *,
    text: str | None = None,
    inventory_id: InventoryId | list[InventoryId] | None = None,
    location_id: str | list[str] | None = None,
    storage_location_id: str | list[str] | None = None,
    task_id: TaskId | list[TaskId] | None = None,
    category: InventoryCategory | str | list[InventoryCategory | str] | None = None,
    external_barcode_id: str | list[str] | None = None,
    search_field: str | list[str] | None = None,
    source_field: str | list[str] | None = None,
    additional_field: str | list[str] | None = None,
    is_drop_down: bool | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    sort_by: str | None = None,
    offset: int | None = None,
    max_items: int | None = None,
) -> Iterator[LotSearchItem]:
    """
    Search for Lot records matching the provided filters.

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

    Parameters
    ----------
    text : str, optional
        Free-text query matched against lot fields.
    inventory_id : InventoryId or list[InventoryId], optional
        Filter by parent inventory IDs.
    location_id : str or list[str], optional
        Filter by specific location IDs.
    storage_location_id : str or list[str], optional
        Filter by storage location IDs.
    task_id : TaskId or list[TaskId], optional
        Filter by source task IDs.
    category : InventoryCategory or list[str], optional
        Filter by parent inventory categories.
    external_barcode_id : str or list[str], optional
        Filter by external barcode IDs.
    search_field : str or list[str], optional
        Restrict the fields the `text` query searches.
    source_field : str or list[str], optional
        Restrict which fields are returned in the response.
    additional_field : str or list[str], optional
        Request additional columns from the search index.
    is_drop_down : bool, optional
        Use dropdown sanitization for the search text when True.
    order_by : OrderBy, optional
        Sort order for the results, default DESCENDING.
    sort_by : str, optional
        Attribute to sort by.
    offset : int, optional
        Pagination offset to start from.
    max_items : int, optional
        Maximum number of items to return in total. If None, fetches all available items.

    Returns
    -------
    Iterator[LotSearchItem]
        An iterator of matching partial (unhydrated) lot entities.
    """

    search_text = text if (text is None or len(text) < 50) else text[:50]

    def _ensure_list(value):
        if value is None:
            return None
        if isinstance(value, list | tuple | set):
            return list(value)
        return [value]

    def _format_categories(value):
        raw = _ensure_list(value)
        if raw is None:
            return None
        formatted: list[str] = []
        for category in raw:
            formatted.append(
                category.value if isinstance(category, InventoryCategory) else category
            )
        return formatted

    params = {
        "offset": offset,
        "order": order_by.value,
        "text": search_text,
        "sortBy": sort_by,
        "isDropDown": is_drop_down,
        "inventoryId": _ensure_list(inventory_id),
        "locationId": _ensure_list(location_id),
        "storageLocationId": _ensure_list(storage_location_id),
        "taskId": _ensure_list(task_id),
        "category": _format_categories(category),
        "externalBarcodeId": _ensure_list(external_barcode_id),
        "searchField": _ensure_list(search_field),
        "sourceField": _ensure_list(source_field),
        "additionalField": _ensure_list(additional_field),
    }
    params = {key: value for key, value in params.items() if value is not None}

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        params=params,
        max_items=max_items,
        deserialize=lambda items: [
            LotSearchItem(**item)._bind_collection(self) for item in items
        ],
    )

get_all

get_all(
    *,
    parent_id: InventoryId | None = None,
    inventory_id: InventoryId | None = None,
    barcode_id: str | None = None,
    parent_id_category: str | None = None,
    inventory_on_hand: str | None = None,
    location_id: str | None = None,
    exact_match: bool = False,
    begins_with: bool = False,
    start_key: str | None = None,
    max_items: int | None = None,
) -> Iterator[Lot]

Get all Lot entities with optional filters.

Parameters:

Name	Type	Description	Default
parent_id	str	Fetch lots for the given parentId (inventory).	None
inventory_id	str	Fetch lots for the given inventoryId.	None
barcode_id	str	Fetch lots for the given barcodeId.	None
parent_id_category	str	Filter by parentIdCategory (e.g., RawMaterials, Consumables).	None
inventory_on_hand	str	Filter by inventoryOnHand (lteZero, gtZero, eqZero).	None
location_id	str	Filter by locationId.	None
exact_match	bool	Whether to match barcodeId exactly. Default is False.	False
begins_with	bool	Whether to match barcodeId as prefix. Default is False.	False
start_key	str	The pagination key to continue listing from.	None
max_items	int	Maximum number of items to return in total. If None, fetches all available items.	None

Returns:

Type	Description
Iterator[Lot]	An iterator of Lot entities matching the filters.

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

@validate_call
def get_all(
    self,
    *,
    parent_id: InventoryId | None = None,
    inventory_id: InventoryId | None = None,
    barcode_id: str | None = None,
    parent_id_category: str | None = None,
    inventory_on_hand: str | None = None,
    location_id: str | None = None,
    exact_match: bool = False,
    begins_with: bool = False,
    start_key: str | None = None,
    max_items: int | None = None,
) -> Iterator[Lot]:
    """
    Get all Lot entities with optional filters.

    Parameters
    ----------
    parent_id : str, optional
        Fetch lots for the given parentId (inventory).
    inventory_id : str, optional
        Fetch lots for the given inventoryId.
    barcode_id : str, optional
        Fetch lots for the given barcodeId.
    parent_id_category : str, optional
        Filter by parentIdCategory (e.g., RawMaterials, Consumables).
    inventory_on_hand : str, optional
        Filter by inventoryOnHand (lteZero, gtZero, eqZero).
    location_id : str, optional
        Filter by locationId.
    exact_match : bool, optional
        Whether to match barcodeId exactly. Default is False.
    begins_with : bool, optional
        Whether to match barcodeId as prefix. Default is False.
    start_key : str, optional
        The pagination key to continue listing from.
    max_items : int, optional
        Maximum number of items to return in total. If None, fetches all available items.

    Returns
    -------
    Iterator[Lot]
        An iterator of Lot entities matching the filters.
    """
    params = {
        "parentId": parent_id,
        "inventoryId": inventory_id,
        "barcodeId": barcode_id,
        "parentIdCategory": parent_id_category,
        "inventoryOnHand": inventory_on_hand,
        "locationId": location_id,
        "startKey": start_key,
        "exactMatch": exact_match,
        "beginsWith": begins_with,
    }

    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        max_items=max_items,
        deserialize=lambda items: [Lot(**item) for item in items],
    )

update

update(*, lot: Lot) -> Lot

Update a lot.

Parameters:

Name	Type	Description	Default
lot	Lot	The updated lot object.	required

Returns:

Type	Description
Lot	The updated Lot entity as returned by the server.

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

def update(self, *, lot: Lot) -> Lot:
    """Update a lot.

    Parameters
    ----------
    lot : Lot
        The updated lot object.

    Returns
    -------
    Lot
        The updated Lot entity as returned by the server.
    """
    existing_lot = self.get_by_id(id=lot.id)
    patch_data = self._generate_lots_patch_payload(existing=existing_lot, updated=lot)
    url = f"{self.base_path}/{lot.id}"
    if patch_data.data:
        self.session.patch(url, json=patch_data.model_dump(mode="json", by_alias=True))

    return self.get_by_id(id=lot.id)