Skip to content

Inventory

albert.collections.inventory

InventoryId

InventoryId = Annotated[
    str, AfterValidator(ensure_inventory_id)
]

ProjectId

ProjectId = Annotated[
    str, AfterValidator(ensure_project_id)
]

SearchProjectId

SearchProjectId = Annotated[
    str, AfterValidator(ensure_project_search_id)
]

WorksheetId

WorksheetId = Annotated[
    str, AfterValidator(ensure_worksheet_id)
]

ALL_MERGE_MODULES

ALL_MERGE_MODULES = [
    "PRICING",
    "NOTES",
    "SDS",
    "PD",
    "BD",
    "LOT",
    "CAS",
    "TAS",
    "WFL",
    "PRG",
    "PTD",
]

All modules selectable for inventory merge.

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

session = session

OrderBy

Bases: str, Enum

DESCENDING

DESCENDING = 'desc'

ASCENDING

ASCENDING = 'asc'

Cas

Bases: BaseModel

Represents a CAS entity.

Show JSON schema:
{
  "$defs": {
    "CasCategory": {
      "enum": [
        "User",
        "Verisk",
        "TSCA - Public",
        "TSCA - Private",
        "not TSCA",
        "CAS linked to External Database",
        "Unknown (Trade Secret)",
        "CL_Inventory Upload"
      ],
      "title": "CasCategory",
      "type": "string"
    },
    "Hazard": {
      "description": "Represents a chemical hazard.",
      "properties": {
        "subCategory": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Hazard subcategory",
          "title": "Subcategory"
        },
        "hCode": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Hazard code",
          "title": "Hcode"
        },
        "category": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Hazard category",
          "title": "Category"
        },
        "class": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Hazard classification",
          "title": "Class"
        },
        "hCodeText": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Hazard code text",
          "title": "Hcodetext"
        }
      },
      "title": "Hazard",
      "type": "object"
    }
  },
  "description": "Represents a CAS entity.",
  "properties": {
    "number": {
      "description": "The CAS number.",
      "title": "Number",
      "type": "string"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Name of the CAS.",
      "title": "Name"
    },
    "description": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The description or name of the CAS.",
      "title": "Description"
    },
    "notes": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Notes related to the CAS.",
      "title": "Notes"
    },
    "category": {
      "anyOf": [
        {
          "$ref": "#/$defs/CasCategory"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The category of the CAS."
    },
    "casSmiles": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "CAS SMILES notation.",
      "title": "Cassmiles"
    },
    "inchiKey": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "InChIKey of the CAS.",
      "title": "Inchikey"
    },
    "iUpacName": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "IUPAC name of the CAS.",
      "title": "Iupacname"
    },
    "albertId": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The AlbertID of the CAS.",
      "title": "Albertid"
    },
    "hazards": {
      "anyOf": [
        {
          "items": {
            "$ref": "#/$defs/Hazard"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Hazards associated with the CAS.",
      "title": "Hazards"
    },
    "wgk": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "German Water Hazard Class (WGK) number.",
      "title": "Wgk"
    },
    "ecListNo": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "European Community (EC) number.",
      "title": "Eclistno"
    },
    "type": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Type of the CAS.",
      "title": "Type"
    },
    "classificationType": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Classification type of the CAS.",
      "title": "Classificationtype"
    },
    "order": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "CAS order.",
      "title": "Order"
    }
  },
  "required": [
    "number"
  ],
  "title": "Cas",
  "type": "object"
}

Fields:

number

number: str

The CAS number.

name

name: str | None = None

Name of the CAS.

description

description: str | None = None

The description or name of the CAS.

notes

notes: str | None = None

Notes related to the CAS.

category

category: CasCategory | None = None

The category of the CAS.

smiles

smiles: str | None = None

CAS SMILES notation.

inchi_key

inchi_key: str | None = None

InChIKey of the CAS.

iupac_name

iupac_name: str | None = None

IUPAC name of the CAS.

id

id: str | None = None

The AlbertID of the CAS.

hazards

hazards: list[Hazard] | None = None

Hazards associated with the CAS.

wgk

wgk: str | None = None

German Water Hazard Class (WGK) number.

ec_number

ec_number: str | None = None

European Community (EC) number.

type

type: str | None = None

Type of the CAS.

classification_type

classification_type: str | None = None

Classification type of the CAS.

order

order: str | None = None

CAS order.

from_string

from_string(*, number: str) -> Cas

Creates a Cas object from a string.

Parameters:

Name Type Description Default
number str

The CAS number.

required

Returns:

Type Description
Cas

The Cas object created from the string.

Source code in src/albert/resources/cas.py
@classmethod
def from_string(cls, *, number: str) -> "Cas":
    """
    Creates a Cas object from a string.

    Parameters
    ----------
    number : str
        The CAS number.

    Returns
    -------
    Cas
        The Cas object created from the string.
    """
    return cls(number=number)

Company

Bases: BaseResource

Company is a Pydantic model representing a company entity.

Attributes:

Name Type Description
name str

The name of the company.

id str | None

The Albert ID of the company. Set when the company is retrieved from Albert.

distance float | None

The scores of a company in a search result, optional. Read-only.

name

name: str

id

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

distance

distance: float | None = Field(
    default=None, exclude=True, frozen=True
)

CompanyCollection

CompanyCollection(*, session: AlbertSession)

Bases: BaseCollection

CompanyCollection is a collection class for managing Company entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required
Source code in src/albert/collections/companies.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the CompanyCollection with the provided session.

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

base_path

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

list

list(
    *,
    limit: int = 50,
    name: str | list[str] = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Company]

Lists company entities with optional filters.

Parameters:

Name Type Description Default
limit int

The maximum number of companies to return, by default 50.

50
name Union[str, None]

The name of the company to filter by, by default None.

None
exact_match bool

Whether to match the name exactly, by default True.

True

Returns:

Type Description
Iterator

An iterator of Company objects.

Source code in src/albert/collections/companies.py
def list(
    self,
    *,
    limit: int = 50,
    name: str | list[str] = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Company]:
    """
    Lists company entities with optional filters.

    Parameters
    ----------
    limit : int, optional
        The maximum number of companies to return, by default 50.
    name : Union[str, None], optional
        The name of the company to filter by, by default None.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.

    Returns
    -------
    Iterator
        An iterator of Company objects.
    """
    params = {"limit": limit, "dupDetection": "false", "startKey": start_key}
    if name:
        params["name"] = name if isinstance(name, list) else [name]
        params["exactMatch"] = str(exact_match).lower()
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [Company(**item) for item in items],
    )

company_exists

company_exists(
    *, name: str, exact_match: bool = True
) -> bool

Checks if a company exists by its name.

Parameters:

Name Type Description Default
name str

The name of the company to check.

required
exact_match bool

Whether to match the name exactly, by default True.

True

Returns:

Type Description
bool

True if the company exists, False otherwise.

Source code in src/albert/collections/companies.py
def company_exists(self, *, name: str, exact_match: bool = True) -> bool:
    """
    Checks if a company exists by its name.

    Parameters
    ----------
    name : str
        The name of the company to check.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.

    Returns
    -------
    bool
        True if the company exists, False otherwise.
    """
    companies = self.get_by_name(name=name, exact_match=exact_match)
    return bool(companies)

get_by_id

get_by_id(*, id: str) -> Company

Get a company by its ID.

Parameters:

Name Type Description Default
id str

The ID of the company to retrieve.

required

Returns:

Type Description
Company

The Company object.

Source code in src/albert/collections/companies.py
def get_by_id(self, *, id: str) -> Company:
    """
    Get a company by its ID.

    Parameters
    ----------
    id : str
        The ID of the company to retrieve.

    Returns
    -------
    Company
        The Company object.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    company = response.json()
    found_company = Company(**company)
    return found_company

get_by_name

get_by_name(
    *, name: str, exact_match: bool = True
) -> Company | None

Retrieves a company by its name.

Parameters:

Name Type Description Default
name str

The name of the company to retrieve.

required
exact_match bool

Whether to match the name exactly, by default True.

True

Returns:

Type Description
Company

The Company object if found, None otherwise.

Source code in src/albert/collections/companies.py
def get_by_name(self, *, name: str, exact_match: bool = True) -> Company | None:
    """
    Retrieves a company by its name.

    Parameters
    ----------
    name : str
        The name of the company to retrieve.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.

    Returns
    -------
    Company
        The Company object if found, None otherwise.
    """
    found = self.list(name=name, exact_match=exact_match)
    return next(found, None)

create

create(
    *, company: str | Company, check_if_exists: bool = True
) -> Company

Creates a new company entity.

Parameters:

Name Type Description Default
company Union[str, Company]

The company name or Company object to create.

required
check_if_exists bool

Whether to check if the company already exists, by default True.

True

Returns:

Type Description
Company

The created Company object.

Source code in src/albert/collections/companies.py
def create(self, *, company: str | Company, check_if_exists: bool = True) -> Company:
    """
    Creates a new company entity.

    Parameters
    ----------
    company : Union[str, Company]
        The company name or Company object to create.
    check_if_exists : bool, optional
        Whether to check if the company already exists, by default True.

    Returns
    -------
    Company
        The created Company object.
    """
    if isinstance(company, str):
        company = Company(name=company)
    hit = self.get_by_name(name=company.name, exact_match=True)
    if check_if_exists and hit:
        logging.warning(f"Company {company.name} already exists with id {hit.id}.")
        return hit

    payload = company.model_dump(by_alias=True, exclude_unset=True, mode="json")
    response = self.session.post(self.base_path, json=payload)
    this_company = Company(**response.json())
    return this_company

delete

delete(*, id: str) -> None

Deletes a company entity.

Parameters:

Name Type Description Default
id str

The ID of the company to delete.

required
Source code in src/albert/collections/companies.py
def delete(self, *, id: str) -> None:
    """Deletes a company entity.

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

rename

rename(*, old_name: str, new_name: str) -> Company

Renames an existing company entity.

Parameters:

Name Type Description Default
old_name str

The current name of the company.

required
new_name str

The new name of the company.

required

Returns:

Type Description
Company

The renamed Company object

Source code in src/albert/collections/companies.py
def rename(self, *, old_name: str, new_name: str) -> Company:
    """
    Renames an existing company entity.

    Parameters
    ----------
    old_name : str
        The current name of the company.
    new_name : str
        The new name of the company.

    Returns
    -------
    Company
        The renamed Company object
    """
    company = self.get_by_name(name=old_name, exact_match=True)
    if not company:
        msg = f'Company "{old_name}" not found.'
        logger.error(msg)
        raise AlbertException(msg)
    company_id = company.id
    endpoint = f"{self.base_path}/{company_id}"
    payload = {
        "data": [
            {
                "operation": "update",
                "attribute": "name",
                "oldValue": old_name,
                "newValue": new_name,
            }
        ]
    }
    self.session.patch(endpoint, json=payload)
    updated_company = self.get_by_id(id=company_id)
    return updated_company

update

update(*, company: Company) -> Company

Update a Company entity. The id of the company must be provided.

Parameters:

Name Type Description Default
company Company

The updated Company object.

required

Returns:

Type Description
Company

The updated Company object as registered in Albert.

Source code in src/albert/collections/companies.py
def update(self, *, company: Company) -> Company:
    """Update a Company entity. The id of the company must be provided.

    Parameters
    ----------
    company : Company
        The updated Company object.

    Returns
    -------
    Company
        The updated Company object as registered in Albert.
    """
    # Fetch the current object state from the server or database
    current_object = self.get_by_id(id=company.id)

    # Generate the PATCH payload
    patch_payload = self._generate_patch_payload(existing=current_object, updated=company)
    url = f"{self.base_path}/{company.id}"
    self.session.patch(url, json=patch_payload.model_dump(mode="json", by_alias=True))
    updated_company = self.get_by_id(id=company.id)
    return updated_company

TagCollection

TagCollection(*, session: AlbertSession)

Bases: BaseCollection

TagCollection is a collection class for managing Tag entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Attributes:

Name Type Description
base_path str

The base URL for tag API requests.

Methods:

Name Description
list

Lists tag entities with optional filters.

tag_exists

Checks if a tag exists by its name.

create

Creates a new tag entity.

get_by_id

Retrieves a tag by its ID.

get_by_ids

Retrieve a list of tags by their IDs.

get_by_tag

Retrieves a tag by its name.

delete

Deletes a tag by its ID.

rename

Renames an existing tag entity.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required
Source code in src/albert/collections/tags.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the TagCollection with the provided session.

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

base_path

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

tag_exists

tag_exists(*, tag: str, exact_match: bool = True) -> bool

Checks if a tag exists by its name.

Parameters:

Name Type Description Default
tag str

The name of the tag to check.

required
exact_match bool

Whether to match the name exactly, by default True.

True

Returns:

Type Description
bool

True if the tag exists, False otherwise.

Source code in src/albert/collections/tags.py
def tag_exists(self, *, tag: str, exact_match: bool = True) -> bool:
    """
    Checks if a tag exists by its name.

    Parameters
    ----------
    tag : str
        The name of the tag to check.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.

    Returns
    -------
    bool
        True if the tag exists, False otherwise.
    """

    return self.get_by_tag(tag=tag, exact_match=exact_match) is not None

create

create(*, tag: str | Tag) -> Tag

Creates a new tag entity if the given tag does not exist.

Parameters:

Name Type Description Default
tag Union[str, Tag]

The tag name or Tag object to create.

required

Returns:

Type Description
Tag

The created Tag object or the existing Tag object of it already exists.

Source code in src/albert/collections/tags.py
def create(self, *, tag: str | Tag) -> Tag:
    """
    Creates a new tag entity if the given tag does not exist.

    Parameters
    ----------
    tag : Union[str, Tag]
        The tag name or Tag object to create.

    Returns
    -------
    Tag
        The created Tag object or the existing Tag object of it already exists.
    """
    if isinstance(tag, str):
        tag = Tag(tag=tag)
    hit = self.get_by_tag(tag=tag.tag, exact_match=True)
    if hit is not None:
        logging.warning(f"Tag {hit.tag} already exists with id {hit.id}")
        return hit
    payload = {"name": tag.tag}
    response = self.session.post(self.base_path, json=payload)
    tag = Tag(**response.json())
    return tag

get_by_id

get_by_id(*, id: str) -> Tag

Get a tag by its ID.

Parameters:

Name Type Description Default
id str

The ID of the tag to get.

required

Returns:

Type Description
Tag

The Tag object.

Source code in src/albert/collections/tags.py
def get_by_id(self, *, id: str) -> Tag:
    """
    Get a tag by its ID.

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

    Returns
    -------
    Tag
        The Tag object.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return Tag(**response.json())

get_by_ids

get_by_ids(*, ids: list[str]) -> list[Tag]
Source code in src/albert/collections/tags.py
def get_by_ids(self, *, ids: list[str]) -> list[Tag]:
    url = f"{self.base_path}/ids"
    batches = [ids[i : i + 100] for i in range(0, len(ids), 100)]
    return [
        Tag(**item)
        for batch in batches
        for item in self.session.get(url, params={"id": batch}).json()
    ]

get_by_tag

get_by_tag(
    *, tag: str, exact_match: bool = True
) -> Tag | None

Retrieves a tag by its name of None if not found.

Parameters:

Name Type Description Default
tag str

The name of the tag to retrieve.

required
exact_match bool

Whether to match the name exactly, by default True.

True

Returns:

Type Description
Tag

The Tag object if found, None otherwise.

Source code in src/albert/collections/tags.py
def get_by_tag(self, *, tag: str, exact_match: bool = True) -> Tag | None:
    """
    Retrieves a tag by its name of None if not found.

    Parameters
    ----------
    tag : str
        The name of the tag to retrieve.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.

    Returns
    -------
    Tag
        The Tag object if found, None otherwise.
    """
    found = self.list(name=tag, exact_match=exact_match)
    return next(found, None)

delete

delete(*, id: str) -> None

Deletes a tag by its ID.

Parameters:

Name Type Description Default
id str

The ID of the tag to delete.

required

Returns:

Type Description
None
Source code in src/albert/collections/tags.py
def delete(self, *, id: str) -> None:
    """
    Deletes a tag by its ID.

    Parameters
    ----------
    id : str
        The ID of the tag to delete.

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

rename

rename(*, old_name: str, new_name: str) -> Tag

Renames an existing tag entity.

Parameters:

Name Type Description Default
old_name str

The current name of the tag.

required
new_name str

The new name of the tag.

required

Returns:

Type Description
Tag

The renamed Tag.

Source code in src/albert/collections/tags.py
def rename(self, *, old_name: str, new_name: str) -> Tag:
    """
    Renames an existing tag entity.

    Parameters
    ----------
    old_name : str
        The current name of the tag.
    new_name : str
        The new name of the tag.

    Returns
    -------
    Tag
        The renamed Tag.
    """
    found_tag = self.get_by_tag(tag=old_name, exact_match=True)
    if not found_tag:
        msg = f'Tag "{old_name}" not found.'
        logger.error(msg)
        raise AlbertException(msg)
    tag_id = found_tag.id
    payload = [
        {
            "data": [
                {
                    "operation": "update",
                    "attribute": "name",
                    "oldValue": old_name,
                    "newValue": new_name,
                }
            ],
            "id": tag_id,
        }
    ]
    self.session.patch(self.base_path, json=payload)
    return self.get_by_id(id=tag_id)

list

list(
    *,
    limit: int = 50,
    order_by: OrderBy = DESCENDING,
    name: str | list[str] | None = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Tag]

Lists Tag entities with optional filters.

Parameters:

Name Type Description Default
limit int

The maximum number of tags to return, by default 50.

50
order_by OrderBy

The order by which to sort the results, by default OrderBy.DESCENDING.

DESCENDING
name Union[str, None]

The name of the tag to filter by, by default None.

None
exact_match bool

Whether to match the name exactly, by default True.

True
start_key Optional[str]

The starting point for the next set of results, by default None.

None

Returns:

Type Description
Iterator[Tag]

An iterator of Tag objects.

Source code in src/albert/collections/tags.py
def list(
    self,
    *,
    limit: int = 50,
    order_by: OrderBy = OrderBy.DESCENDING,
    name: str | list[str] | None = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Tag]:
    """
    Lists Tag entities with optional filters.

    Parameters
    ----------
    limit : int, optional
        The maximum number of tags to return, by default 50.
    order_by : OrderBy, optional
        The order by which to sort the results, by default OrderBy.DESCENDING.
    name : Union[str, None], optional
        The name of the tag to filter by, by default None.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.
    start_key : Optional[str], optional
        The starting point for the next set of results, by default None.

    Returns
    -------
    Iterator[Tag]
        An iterator of Tag objects.
    """
    params = {"limit": limit, "orderBy": order_by.value, "startKey": start_key}
    if name:
        params["name"] = [name] if isinstance(name, str) else name
        params["exactMatch"] = json.dumps(exact_match)
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [Tag(**item) for item in items],
    )

FacetItem

Bases: BaseAlbertModel

name

name: str

parameter

parameter: str

type

type: FacetType

value

value: list[FacetValue] = Field(
    default_factory=list, alias="Value"
)

InventoryCategory

Bases: str, Enum

RAW_MATERIALS

RAW_MATERIALS = 'RawMaterials'

CONSUMABLES

CONSUMABLES = 'Consumables'

EQUIPMENT

EQUIPMENT = 'Equipment'

FORMULAS

FORMULAS = 'Formulas'

InventoryItem

Bases: BaseTaggedResource

An InventoryItem is a Pydantic model representing an item in the inventory. Can be a raw material, consumable, equipment, or formula. Note: Formulas should be registered via the Worksheet collection / Sheet resource.

Returns:

Type Description
InventoryItem

An InventoryItem that can be used to represent an item in the inventory. Can be a raw material, consumable, equipment, or formula.

Attributes:

Name Type Description
name str

The name of the InventoryItem.

id str | None

The Albert ID of the InventoryItem. Set when the InventoryItem is retrieved from Albert.

description str | None

The description of the InventoryItem.

category InventoryCategory

The category of the InventoryItem. Allowed values are RawMaterials, Consumables, Equipment, and Formulas.

unit_category InventoryUnitCategory

The unit category of the InventoryItem. Can be mass, volume, length, pressure, or units. By default, mass is used for RawMaterials and Formulas, and units is used for Equipment and Consumables.

security_class SecurityClass | None

The security class of the InventoryItem. Optional. Can be confidential, shared, or restricted.

company Company | str | None

The company associated with the InventoryItem. Can be a Company object or a string. If a String is provided, a Company object with the name of the provided string will be first-or-created.

minimum list[InventoryMinimum] | None

The minimum amount of the InventoryItem that must be kept in stock at a given Location. Optional.

alias str | None

An alias for the InventoryItem. Optional.

cas list[CasAmount] | None

The CAS numbers associated with the InventoryItem. This is how a compositional breakdown can be provided. Optional.

metadata dict[str, str | list[EntityLink] | EntityLink] | None

Metadata associated with the InventoryItem. Optional. Allowed metadata fields can be found in the CustomFields documentation.

project_id str | None

The project ID associated with the InventoryItem. Read Only. Required for Formulas.

formula_id str | None

The formula ID associated with the InventoryItem. Read Only.

tags list[str | Tag] | None

The tags associated with the InventoryItem. Optional. If a string is provided, a Tag object with the name of the provided string will be first-or-created.

name

name: str | None = None

id

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

description

description: str | None = None

category

unit_category

unit_category: InventoryUnitCategory | None = Field(
    default=None, alias="unitCategory"
)

security_class

security_class: SecurityClass | None = Field(
    default=None, alias="class"
)

company

company: SerializeAsEntityLink[Company] | None = Field(
    default=None, alias="Company"
)

minimum

minimum: list[InventoryMinimum] | None = Field(default=None)

alias

alias: str | None = Field(default=None)

cas

cas: list[CasAmount] | None = Field(
    default=None, alias="Cas"
)

metadata

metadata: dict[str, MetadataItem] | None = Field(
    alias="Metadata", default=None
)

project_id

project_id: str | None = Field(
    default=None, alias="parentId"
)

acls

acls: list[ACL] = Field(default_factory=list, alias='ACL')

task_config

task_config: list[dict] | None = Field(
    default=None,
    alias="TaskConfig",
    exclude=True,
    frozen=True,
)

formula_id

formula_id: str | None = Field(
    default=None,
    alias="formulaId",
    exclude=True,
    frozen=True,
)

symbols

symbols: list[dict] | None = Field(
    default=None, alias="Symbols", exclude=True, frozen=True
)

un_number

un_number: str | None = Field(
    default=None,
    alias="unNumber",
    exclude=True,
    frozen=True,
)

recent_atachment_id

recent_atachment_id: str | None = Field(
    default=None,
    alias="recentAttachmentId",
    exclude=True,
    frozen=True,
)

validate_company_string

validate_company_string(value: Any) -> Any
Source code in src/albert/resources/inventory.py
@field_validator("company", mode="before")
@classmethod
def validate_company_string(cls, value: Any) -> Any:
    if isinstance(value, str):
        value = Company(name=value)
    return value

validate_un_number

validate_un_number(value: Any) -> Any
Source code in src/albert/resources/inventory.py
@field_validator("un_number", mode="before")
@classmethod
def validate_un_number(cls, value: Any) -> Any:
    if value == "N/A":
        value = None
    return value

set_unit_category

set_unit_category() -> InventoryItem

Set unit category from category if not defined.

Source code in src/albert/resources/inventory.py
@model_validator(mode="after")
def set_unit_category(self) -> "InventoryItem":
    """Set unit category from category if not defined."""
    if self.unit_category is None:
        if self.category in [InventoryCategory.RAW_MATERIALS, InventoryCategory.FORMULAS]:
            object.__setattr__(self, "unit_category", InventoryUnitCategory.MASS)
        elif self.category in [InventoryCategory.EQUIPMENT, InventoryCategory.CONSUMABLES]:
            object.__setattr__(self, "unit_category", InventoryUnitCategory.UNITS)
    return self

validate_formula_fields

validate_formula_fields() -> InventoryItem

Ensure required fields are present for formulas.

Source code in src/albert/resources/inventory.py
@model_validator(mode="after")
def validate_formula_fields(self) -> "InventoryItem":
    """Ensure required fields are present for formulas."""
    if self.category == InventoryCategory.FORMULAS and not self.project_id and not self.id:
        # Some legacy on platform formulas don't have a project_id so check if its already on platform
        raise ValueError("A project_id must be supplied for all formulas.")
    return self

InventorySearchItem

Bases: BaseAlbertModel

id

id: str = Field(alias='albertId')

name

name: str = Field(default='')

description

description: str = Field(default='')

category

unit

unit: InventoryUnitCategory

lots

lots: list[dict[str, Any]] = Field(default_factory=list)

tags

tags: list[Tag] = Field(default_factory=list)

pictogram

pictogram: list[InventorySearchPictogramItem] = Field(
    default_factory=list
)

inventory_on_hand

inventory_on_hand: float = Field(
    default=0.0, alias="inventoryOnHand"
)

sds

sds: InventorySearchSDSItem | None = Field(
    default=None, alias="SDS"
)

InventorySpec

Bases: BaseAlbertModel

id

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

name

name: str

data_column_id

data_column_id: str = Field(..., alias='datacolumnId')

data_column_name

data_column_name: str | None = Field(
    default=None, alias="datacolumnName"
)

data_template_id

data_template_id: str | None = Field(
    default=None, alias="datatemplateId"
)

data_template_name

data_template_name: str | None = Field(
    default=None, alias="datatemplateName"
)

unit_id

unit_id: str | None = Field(default=None, alias='unitId')

unit_name

unit_name: str | None = Field(
    default=None, alias="unitName"
)

workflow_id

workflow_id: str | None = Field(
    default=None, alias="workflowId"
)

workflow_name

workflow_name: str | None = Field(
    default=None, alias="workflowName"
)

spec_config

spec_config: str | None = Field(
    default=None, alias="specConfig"
)

value

value: InventorySpecValue | None = Field(
    default=None, alias="Value"
)

InventorySpecList

Bases: BaseAlbertModel

parent_id

parent_id: str = Field(..., alias='parentId')

specs

specs: list[InventorySpec] = Field(..., alias='Specs')

MergeInventory

Bases: BaseAlbertModel

parent_id

parent_id: InventoryId = Field(alias='parentId')

child_inventories

child_inventories: list[dict[str, InventoryId]] = Field(
    alias="ChildInventories"
)

modules

modules: list[str] | None = Field(default=None)

Location

Bases: BaseResource

A location in Albert.

Attributes:

Name Type Description
name str

The name of the location.

id str | None

The Albert ID of the location. Set when the location is retrieved from Albert.

latitude float

The latitude of the location.

longitude float

The longitude of the location.

address str

The address of the location.

country str | None

The country code of the location. Must be two characters long.

name

name: str

id

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

latitude

latitude: float = Field()

longitude

longitude: float = Field()

address

address: str

country

country: str | None = Field(
    None, max_length=2, min_length=2
)

StorageLocation

Bases: BaseResource

A storage location entity. For example, a specific flammables cabinet or a storage room.

Attributes:

Name Type Description
name str

The name of the storage location.

id str | None

The Albert ID of the storage location. Set when the storage location is retrieved from Albert.

location Location

The location entity link of the storage location.

name

name: str = Field(
    alias="name", min_length=2, max_length=255
)

id

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

location

location: SerializeAsEntityLink[Location] = Field(
    alias="Location"
)

User

Bases: BaseResource

Represents a User on the Albert Platform

Attributes:

Name Type Description
name str

The name of the user.

id str | None

The Albert ID of the user. Set when the user is retrieved from Albert.

location Location | None

The location of the user.

email EmailStr | None

The email of the user.

roles list[Role]

The roles of the user.

user_class UserClass

The ACL class level of the user.

metadata dict[str, str | list[EntityLink] | EntityLink] | None

name

name: str

id

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

location

location: SerializeAsEntityLink[Location] | None = Field(
    default=None, alias="Location"
)

email

email: EmailStr = Field(default=None, alias='email')

roles

roles: list[SerializeAsEntityLink[Role]] = Field(
    max_length=1, default_factory=list, alias="Roles"
)

user_class

user_class: UserClass = Field(
    default=STANDARD, alias="userClass"
)

metadata

metadata: dict[str, MetadataItem] | None = Field(
    alias="Metadata", default=None
)

to_note_mention

to_note_mention() -> str

Convert the user to a note mention string.

Returns:

Type Description
str

The note mention string.

Source code in src/albert/resources/users.py
def to_note_mention(self) -> str:
    """Convert the user to a note mention string.

    Returns
    -------
    str
        The note mention string.
    """
    return f"@{self.name}#{self.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
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

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

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