Inventory

albert.collections.inventory.InventoryCollection

InventoryCollection(*, session: AlbertSession)

Bases: BaseCollection

InventoryCollection is a collection class for managing Inventory Item entities in the Albert platform.

Parameters:

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

Methods:

Name	Description
merge	Merge one or multiple child inventory into a parent inventory item.
inventory_exists	Check if an inventory item exists.
get_match_or_none	Get a matching inventory item or return None if not found.
create	Create a new inventory item.
get_by_id	Retrieve an inventory item by its ID.
get_by_ids	Retrieve an set of inventory items by their IDs.
get_specs	Get the specs for a list of inventory items.
add_specs	Add inventory specs to the inventory item.
delete	Delete an inventory item by its ID.
get_all_facets	Get available facets for inventory items based on the provided filters.
get_facet_by_name	Returns a specific facet by its name with all the filters applied to the search.
search	Get a list of inventory items that match the search criteria and
list	List inventory items with optional filters.
update	Update an inventory item.

Attributes:

Name	Type	Description
base_path

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

def __init__(self, *, session: AlbertSession):
    """
    InventoryCollection is a collection class for managing inventory items.

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

base_path

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

merge

merge(
    *,
    parent_id: InventoryId,
    child_id: InventoryId | list[InventoryId],
    modules: list[str] | None = None,
) -> None

Merge one or multiple child inventory into a parent inventory item.

Parameters:

Name	Type	Description	Default
parent_id	InventoryId	The ID of the parent inventory item.	required
child_id	InventoryId | list[InventoryId]	The ID(s) of the child inventory item(s).	required
modules	list[str]	The merge modules to use (default is all).	None

Returns:

Type	Description
None

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

def merge(
    self,
    *,
    parent_id: InventoryId,
    child_id: InventoryId | list[InventoryId],
    modules: list[str] | None = None,
) -> None:
    """
    Merge one or multiple child inventory into a parent inventory item.

    Parameters
    ----------
    parent_id : InventoryId
        The ID of the parent inventory item.
    child_id : InventoryId | list[InventoryId]
        The ID(s) of the child inventory item(s).
    modules : list[str], optional
        The merge modules to use (default is all).

    Returns
    -------
    None
    """

    # assume "all" modules if not specified explicitly
    modules = modules if modules is not None else ALL_MERGE_MODULES

    # define merge endpoint
    url = f"{self.base_path}/merge"

    if isinstance(child_id, list):
        child_inventories = [{"id": i} for i in child_id]
    else:
        child_inventories = [{"id": child_id}]

    # define payload using the class
    payload = MergeInventory(
        parent_id=parent_id,
        child_inventories=child_inventories,
        modules=modules,
    )

    # post request
    self.session.post(url, json=payload.model_dump(mode="json", by_alias=True))

inventory_exists

inventory_exists(*, inventory_item: InventoryItem) -> bool

Check if an inventory item exists.

Parameters:

Name	Type	Description	Default
inventory_item	InventoryItem	The inventory item to check.	required

Returns:

Type	Description
bool	True if the inventory item exists, False otherwise.

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

def inventory_exists(self, *, inventory_item: InventoryItem) -> bool:
    """
    Check if an inventory item exists.

    Parameters
    ----------
    inventory_item : InventoryItem
        The inventory item to check.

    Returns
    -------
    bool
        True if the inventory item exists, False otherwise.
    """
    hit = self.get_match_or_none(inventory_item=inventory_item)
    return bool(hit)

get_match_or_none

get_match_or_none(
    *, inventory_item: InventoryItem
) -> InventoryItem | None

Get a matching inventory item or return None if not found.

Parameters:

Name	Type	Description	Default
inventory_item	InventoryItem	The inventory item to match.	required

Returns:

Type	Description
Union[InventoryItem, None]	The matching inventory item or None if not found.

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

def get_match_or_none(self, *, inventory_item: InventoryItem) -> InventoryItem | None:
    """
    Get a matching inventory item or return None if not found.

    Parameters
    ----------
    inventory_item : InventoryItem
        The inventory item to match.

    Returns
    -------
    Union[InventoryItem, None]
        The matching inventory item or None if not found.
    """
    hits = self.list(text=inventory_item.name, company=[inventory_item.company])
    inv_company = (
        inventory_item.company.name
        if isinstance(inventory_item.company, Company)
        else inventory_item.company
    )
    for inv in hits:
        if inv and inv.name == inventory_item.name and inv.company.name == inv_company:
            return inv
    else:
        return None

create

create(
    *,
    inventory_item: InventoryItem,
    avoid_duplicates: bool = True,
) -> InventoryItem

Create a new inventory item.

Parameters:

Name	Type	Description	Default
inventory_item	InventoryItem	The inventory item to create.	required
avoid_duplicates	bool	Whether to avoid creating duplicate items (default is True).	True

Returns:

Type	Description
InventoryItem	The created inventory item.

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

def create(
    self,
    *,
    inventory_item: InventoryItem,
    avoid_duplicates: bool = True,
) -> InventoryItem:
    """
    Create a new inventory item.

    Parameters
    ----------
    inventory_item : InventoryItem
        The inventory item to create.
    avoid_duplicates : bool, optional
        Whether to avoid creating duplicate items (default is True).

    Returns
    -------
    InventoryItem
        The created inventory item.
    """
    category = (
        inventory_item.category
        if isinstance(inventory_item.category, str)
        else inventory_item.category.value
    )
    if category == InventoryCategory.FORMULAS.value:
        # This will need to interact with worksheets
        raise NotImplementedError("Registrations of formulas not yet implemented")
    tag_collection = TagCollection(session=self.session)
    if inventory_item.tags is not None and inventory_item.tags != []:
        all_tags = [
            tag_collection.create(tag=t) if t.id is None else t for t in inventory_item.tags
        ]
        inventory_item.tags = all_tags
    if inventory_item.company and inventory_item.company.id is None:
        company_collection = CompanyCollection(session=self.session)
        inventory_item.company = company_collection.create(company=inventory_item.company)
    # Check to see if there is a match on name + Company already
    if avoid_duplicates:
        existing = self.get_match_or_none(inventory_item=inventory_item)
        if isinstance(existing, InventoryItem):
            logging.warning(
                f"Inventory item already exists with name {existing.name} and company {existing.company.name}, returning existing item."
            )
            return existing
    response = self.session.post(
        self.base_path,
        json=inventory_item.model_dump(by_alias=True, exclude_none=True, mode="json"),
    )
    return InventoryItem(**response.json())

get_by_id

get_by_id(*, id: InventoryId) -> InventoryItem

Retrieve an inventory item by its ID.

Parameters:

Name	Type	Description	Default
id	InventoryId	The ID of the inventory item.	required

Returns:

Type	Description
InventoryItem	The retrieved inventory item.

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

@validate_call
def get_by_id(self, *, id: InventoryId) -> InventoryItem:
    """
    Retrieve an inventory item by its ID.

    Parameters
    ----------
    id : InventoryId
        The ID of the inventory item.

    Returns
    -------
    InventoryItem
        The retrieved inventory item.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return InventoryItem(**response.json())

get_by_ids

get_by_ids(
    *, ids: list[InventoryId]
) -> list[InventoryItem]

Retrieve an set of inventory items by their IDs.

Parameters:

Name	Type	Description	Default
ids	list[InventoryId]	The list of IDs of the inventory items.	required

Returns:

Type	Description
list[InventoryItem]	The retrieved inventory items.

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

@validate_call
def get_by_ids(self, *, ids: list[InventoryId]) -> list[InventoryItem]:
    """
    Retrieve an set of inventory items by their IDs.

    Parameters
    ----------
    ids : list[InventoryId]
        The list of IDs of the inventory items.

    Returns
    -------
    list[InventoryItem]
        The retrieved inventory items.
    """
    batch_size = 250
    batches = [ids[i : i + batch_size] for i in range(0, len(ids), batch_size)]
    inventory = []
    for batch in batches:
        response = self.session.get(f"{self.base_path}/ids", params={"id": batch})
        inventory.extend([InventoryItem(**item) for item in response.json()["Items"]])
    return inventory

get_specs

get_specs(
    *, ids: list[InventoryId]
) -> list[InventorySpecList]

Get the specs for a list of inventory items.

Parameters:

Name	Type	Description	Default
ids	list[InventoryId]	List of Inventory IDs to get the specs for.	required

Returns:

Type	Description
list[InventorySpecList]	A list of InventorySpecList objects, each containing the specs for an inventory item.

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

@validate_call
def get_specs(self, *, ids: list[InventoryId]) -> list[InventorySpecList]:
    """Get the specs for a list of inventory items.

    Parameters
    ----------
    ids : list[InventoryId]
        List of Inventory IDs to get the specs for.

    Returns
    -------
    list[InventorySpecList]
        A list of InventorySpecList objects, each containing the specs for an inventory item.
    """
    url = f"{self.base_path}/specs"
    batches = [ids[i : i + 250] for i in range(0, len(ids), 250)]
    ta = TypeAdapter(InventorySpecList)
    return [
        ta.validate_python(item)
        for batch in batches
        for item in self.session.get(url, params={"id": batch}).json()
    ]

add_specs

add_specs(
    *,
    inventory_id: InventoryId,
    specs: InventorySpec | list[InventorySpec],
) -> InventorySpecList

Add inventory specs to the inventory item.

An InventorySpec is a property that was not directly measured via a task, but is a generic property of that inentory item.

Parameters:

Name	Type	Description	Default
inventory_id	InventoryId	The Albert ID of the inventory item to add the specs to	required
specs	list[InventorySpec]	List of InventorySpec objects to add to the inventory item, which described the value and, optionally, the conditions associated with the value (via workflow).	required

Returns:

Type	Description
InventorySpecList	The list of InventorySpecs attached to the InventoryItem.

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

@validate_call
def add_specs(
    self,
    *,
    inventory_id: InventoryId,
    specs: InventorySpec | list[InventorySpec],
) -> InventorySpecList:
    """Add inventory specs to the inventory item.

    An `InventorySpec` is a property that was not directly measured via a task,
    but is a generic property of that inentory item.

    Parameters
    ----------
    inventory_id : InventoryId
        The Albert ID of the inventory item to add the specs to
    specs : list[InventorySpec]
        List of InventorySpec objects to add to the inventory item,
        which described the value and, optionally,
        the conditions associated with the value (via workflow).

    Returns
    -------
    InventorySpecList
        The list of InventorySpecs attached to the InventoryItem.
    """
    if isinstance(specs, InventorySpec):
        specs = [specs]
    response = self.session.put(
        url=f"{self.base_path}/{inventory_id}/specs",
        json=[x.model_dump(exclude_unset=True, by_alias=True, mode="json") for x in specs],
    )
    return InventorySpecList(**response.json())

delete

delete(*, id: InventoryId) -> None

Delete an inventory item by its ID.

Parameters:

Name	Type	Description	Default
id	InventoryId	The ID of the inventory item.	required

Returns:

Type	Description
None

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

@validate_call
def delete(self, *, id: InventoryId) -> None:
    """
    Delete an inventory item by its ID.

    Parameters
    ----------
    id : InventoryId
        The ID of the inventory item.

    Returns
    -------
    None
    """

    url = f"{self.base_path}/{id}"
    self.session.delete(url)

get_all_facets

get_all_facets(
    *,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory]
    | InventoryCategory
    | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation]
    | StorageLocation
    | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> list[FacetItem]

Get available facets for inventory items based on the provided filters.

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

@validate_call
def get_all_facets(
    self,
    *,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory] | InventoryCategory | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation] | StorageLocation | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> list[FacetItem]:
    """
    Get available facets for inventory items based on the provided filters.
    """

    params = self._prepare_parameters(
        limit=1,
        text=text,
        cas=cas,
        category=category,
        company=company,
        location=location,
        storage_location=storage_location,
        project_id=project_id,
        sheet_id=sheet_id,
        created_by=created_by,
        lot_owner=lot_owner,
        tags=tags,
    )
    response = self.session.get(
        url=f"{self.base_path}/llmsearch"
        if match_all_conditions
        else f"{self.base_path}/search",
        params=params,
    )
    return [FacetItem.model_validate(x) for x in response.json()["Facets"]]

get_facet_by_name

get_facet_by_name(
    name: str | list[str],
    *,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory]
    | InventoryCategory
    | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation]
    | StorageLocation
    | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> list[FacetItem]

Returns a specific facet by its name with all the filters applied to the search. This can be used for example to fetch all remaining tags as part of an iterative refinement of a search.

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

@validate_call
def get_facet_by_name(
    self,
    name: str | list[str],
    *,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory] | InventoryCategory | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation] | StorageLocation | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> list[FacetItem]:
    """
    Returns a specific facet by its name with all the filters applied to the search.
    This can be used for example to fetch all remaining tags as part of an iterative
    refinement of a search.
    """
    if isinstance(name, str):
        name = [name]

    facets = self.get_all_facets(
        text=text,
        cas=cas,
        category=category,
        company=company,
        location=location,
        storage_location=storage_location,
        project_id=project_id,
        sheet_id=sheet_id,
        created_by=created_by,
        lot_owner=lot_owner,
        tags=tags,
        match_all_conditions=match_all_conditions,
    )
    filtered_facets = []
    for facet in facets:
        if facet.name in name or facet.name.lower() in name:
            filtered_facets.append(facet)

    return filtered_facets

search

search(
    *,
    limit: int = 100,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory]
    | InventoryCategory
    | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation]
    | StorageLocation
    | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> Iterator[InventorySearchItem]

Get a list of inventory items that match the search criteria and return the raw search records. These are not full inventory item objects, but are special short documents intended for fast summary results

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

@validate_call
def search(
    self,
    *,
    limit: int = 100,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory] | InventoryCategory | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation] | StorageLocation | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> Iterator[InventorySearchItem]:
    """
    Get a list of inventory items that match the search criteria and
    return the raw search records. These are not full inventory item
    objects, but are special short documents intended for fast summary results
    """

    def deserialize(items: list[dict]):
        return [InventorySearchItem.model_validate(x) for x in items]

    params = self._prepare_parameters(
        limit=limit,
        text=text,
        cas=cas,
        category=category,
        company=company,
        location=location,
        storage_location=storage_location,
        project_id=project_id,
        sheet_id=sheet_id,
        created_by=created_by,
        lot_owner=lot_owner,
        tags=tags,
    )
    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/llmsearch"
        if match_all_conditions
        else f"{self.base_path}/search",
        params=params,
        session=self.session,
        deserialize=deserialize,
    )

list

list(
    *,
    limit: int = 100,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory]
    | InventoryCategory
    | None = None,
    company: list[Company] | Company | None = None,
    order: OrderBy = DESCENDING,
    sort_by: str | None = "createdAt",
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation]
    | StorageLocation
    | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> Iterator[InventoryItem]

List inventory items with optional filters.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of items to return (default is 100)	100
text	str	Text to search for in inventory names and descriptions	None
cas	list[Cas] | Cas | None	Filter by CAS number(s)	None
category	list[InventoryCategory] | InventoryCategory | None	Filter by inventory category/categories	None
company	list[Company] | Company | None	Filter by manufacturing company/companies	None
order	OrderBy	Sort order, either ASCENDING or DESCENDING (default is DESCENDING)	DESCENDING
sort_by	str	Field to sort by (default is "createdAt")	'createdAt'
location	list[Location] | None	Filter by location(s)	None
storage_location	list[StorageLocation] | None	Filter by storage location(s)	None
project_id	str	Filter by project ID	None
sheet_id	str	Filter by sheet ID	None
created_by	list[User]	Filter by creator(s)	None
lot_owner	list[User]	Filter by lot owner(s)	None
tags	list[str]	Filter by tag(s)	None
match_all_conditions	bool	Whether to match all conditions (default is False -- e.g. OR between conditions)	False

Returns:

Type	Description
Iterator[InventoryItem]	An iterator over the matching inventory items

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

@validate_call
def list(
    self,
    *,
    limit: int = 100,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory] | InventoryCategory | None = None,
    company: list[Company] | Company | None = None,
    order: OrderBy = OrderBy.DESCENDING,
    sort_by: str | None = "createdAt",
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation] | StorageLocation | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> Iterator[InventoryItem]:
    """
    List inventory items with optional filters.

    Parameters
    ----------
    limit : int, optional
        Maximum number of items to return (default is 100)
    text : str, optional
        Text to search for in inventory names and descriptions
    cas : list[Cas] | Cas | None, optional
        Filter by CAS number(s)
    category : list[InventoryCategory] | InventoryCategory | None, optional
        Filter by inventory category/categories
    company : list[Company] | Company | None, optional
        Filter by manufacturing company/companies
    order : OrderBy, optional
        Sort order, either ASCENDING or DESCENDING (default is DESCENDING)
    sort_by : str, optional
        Field to sort by (default is "createdAt")
    location : list[Location] | None, optional
        Filter by location(s)
    storage_location : list[StorageLocation] | None, optional
        Filter by storage location(s)
    project_id : str, optional
        Filter by project ID
    sheet_id : str, optional
        Filter by sheet ID
    created_by : list[User], optional
        Filter by creator(s)
    lot_owner : list[User], optional
        Filter by lot owner(s)
    tags : list[str], optional
        Filter by tag(s)
    match_all_conditions : bool, optional
        Whether to match all conditions (default is False -- e.g. OR between conditions)

    Returns
    -------
    Iterator[InventoryItem]
        An iterator over the matching inventory items
    """

    def deserialize(items: list[dict]) -> list[InventoryItem]:
        return self.get_by_ids(ids=[x["albertId"] for x in items])

    search_text = text if (text is None or len(text) < 50) else text[0:50]
    params = self._prepare_parameters(
        limit=limit,
        text=search_text,
        cas=cas,
        category=category,
        company=company,
        order=order,
        sort_by=sort_by,
        location=location,
        storage_location=storage_location,
        project_id=project_id,
        sheet_id=sheet_id,
        created_by=created_by,
        lot_owner=lot_owner,
        tags=tags,
    )
    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/llmsearch"
        if match_all_conditions
        else f"{self.base_path}/search",
        params=params,
        session=self.session,
        deserialize=deserialize,
    )

update

update(*, inventory_item: InventoryItem) -> InventoryItem

Update an inventory item.

Parameters:

Name	Type	Description	Default
inventory_item	InventoryItem	The updated inventory item object.	required

Returns:

Type	Description
InventoryItem	The updated inventory item retrieved from the server.

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

def update(self, *, inventory_item: InventoryItem) -> InventoryItem:
    """
    Update an inventory item.

    Parameters
    ----------
    inventory_item : InventoryItem
        The updated inventory item object.

    Returns
    -------
    InventoryItem
        The updated inventory item retrieved from the server.
    """
    # Fetch the current object state from the server or database
    current_object = self.get_by_id(id=inventory_item.id)

    # Generate the PATCH payload
    patch_payload = self._generate_inventory_patch_payload(
        existing=current_object, updated=inventory_item
    )

    # Complex patching is not working, so I'm going to do this in a loop :(
    # https://teams.microsoft.com/l/message/19:de4a48c366664ce1bafcdbea02298810@thread.tacv2/1724856117312?tenantId=98aab90e-764b-48f1-afaa-02e3c7300653&groupId=35a36a3d-fc25-4899-a1dd-ad9c7d77b5b3&parentMessageId=1724856117312&teamName=Product%20%2B%20Engineering&channelName=General%20-%20API&createdTime=1724856117312
    url = f"{self.base_path}/{inventory_item.id}"
    for change in patch_payload["data"]:
        change_payload = {"data": [change]}
        self.session.patch(url, json=change_payload)
    updated_inv = self.get_by_id(id=inventory_item.id)
    return updated_inv