Skip to content

Inventory

albert.collections.inventory

ALL_MERGE_MODULES module-attribute

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

All modules selectable for inventory merge.

InventoryId module-attribute

InventoryId = Annotated[
    str, AfterValidator(ensure_inventory_id)
]

ProjectId module-attribute

ProjectId = Annotated[
    str, AfterValidator(ensure_project_id)
]

SearchProjectId module-attribute

SearchProjectId = Annotated[
    str, AfterValidator(ensure_project_search_id)
]

WorksheetId module-attribute

WorksheetId = Annotated[
    str, AfterValidator(ensure_worksheet_id)
]

AlbertSession

AlbertSession(
    *,
    base_url: str,
    token: str | None = None,
    client_credentials: ClientCredentials | None = None,
    retries: int | None = None,
)

Bases: Session

A session that has a base URL, which is prefixed to all request URLs.

Parameters:

Name Type Description Default
base_url str

The base URL to prefix to all requests. (e.g., "https://sandbox.albertinvent.com")

required
retries int

The number of retries for failed requests. Defaults to 3.

None
client_credentials ClientCredentials | None

The client credentials for programmatic authentication. Optional if token is provided.

None
token str | None

The JWT token for authentication. Optional if client credentials are provided.

None

Methods:

Name Description
request
Source code in src/albert/session.py
def __init__(
    self,
    *,
    base_url: str,
    token: str | None = None,
    client_credentials: ClientCredentials | None = None,
    retries: int | None = None,
):
    super().__init__()
    self.base_url = base_url
    self.headers.update(
        {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": f"albert-SDK V.{albert.__version__}",
        }
    )

    if token is None and client_credentials is None:
        raise ValueError("Either client credentials or token must be specified.")

    self._provided_token = token
    self._token_manager = (
        TokenManager(base_url, client_credentials) if client_credentials is not None else None
    )

    # Set up retry logic
    retries = retries if retries is not None else 3
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=0.3,
        status_forcelist=(500, 502, 503, 504, 403),
        raise_on_status=False,
    )
    adapter = HTTPAdapter(max_retries=retry)
    self.mount("http://", adapter)
    self.mount("https://", adapter)

base_url instance-attribute

base_url = base_url

request

request(
    method: str, path: str, *args, **kwargs
) -> Response
Source code in src/albert/session.py
def request(self, method: str, path: str, *args, **kwargs) -> requests.Response:
    self.headers["Authorization"] = f"Bearer {self._access_token}"
    full_url = urljoin(self.base_url, path) if not path.startswith("http") else path
    with handle_http_errors():
        response = super().request(method, full_url, *args, **kwargs)
        response.raise_for_status()
        return response

BaseCollection

BaseCollection(*, session: AlbertSession)

BaseCollection is the base class for all collection classes.

Parameters:

Name Type Description Default
session AlbertSession

The Albert API Session instance.

required
Source code in src/albert/collections/base.py
def __init__(self, *, session: AlbertSession):
    self.session = session

session instance-attribute

session = session

Cas pydantic-model

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:

category pydantic-field

category: CasCategory | None = None

The category of the CAS.

classification_type pydantic-field

classification_type: str | None = None

Classification type of the CAS.

description pydantic-field

description: str | None = None

The description or name of the CAS.

ec_number pydantic-field

ec_number: str | None = None

European Community (EC) number.

hazards pydantic-field

hazards: list[Hazard] | None = None

Hazards associated with the CAS.

id pydantic-field

id: str | None = None

The AlbertID of the CAS.

inchi_key pydantic-field

inchi_key: str | None = None

InChIKey of the CAS.

iupac_name pydantic-field

iupac_name: str | None = None

IUPAC name of the CAS.

name pydantic-field

name: str | None = None

Name of the CAS.

notes pydantic-field

notes: str | None = None

Notes related to the CAS.

number pydantic-field

number: str

The CAS number.

order pydantic-field

order: str | None = None

CAS order.

smiles pydantic-field

smiles: str | None = None

CAS SMILES notation.

type pydantic-field

type: str | None = None

Type of the CAS.

wgk pydantic-field

wgk: str | None = None

German Water Hazard Class (WGK) number.

from_string classmethod

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.

distance class-attribute instance-attribute

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

id class-attribute instance-attribute

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

name instance-attribute

name: str

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

Methods:

Name Description
company_exists

Checks if a company exists by its name.

create

Creates a new company entity.

delete

Deletes a company entity.

get_by_id

Get a company by its ID.

get_by_name

Retrieves a company by its name.

list

Lists company entities with optional filters.

rename

Renames an existing company entity.

update

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

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

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

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)

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)

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)

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

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

FacetItem

Bases: BaseAlbertModel

name instance-attribute

name: str

parameter instance-attribute

parameter: str

type instance-attribute

type: FacetType

value class-attribute instance-attribute

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

InventoryCategory

Bases: str, Enum

CONSUMABLES class-attribute instance-attribute

CONSUMABLES = 'Consumables'

EQUIPMENT class-attribute instance-attribute

EQUIPMENT = 'Equipment'

FORMULAS class-attribute instance-attribute

FORMULAS = 'Formulas'

RAW_MATERIALS class-attribute instance-attribute

RAW_MATERIALS = 'RawMaterials'

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
add_specs

Add inventory specs to the inventory item.

create

Create a new 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_by_id

Retrieve an inventory item by its ID.

get_by_ids

Retrieve an set of inventory items by their IDs.

get_facet_by_name

Returns a specific facet by its name with all the filters applied to the search.

get_match_or_none

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

get_specs

Get the specs for a list of inventory items.

inventory_exists

Check if an inventory item exists.

list

List inventory items with optional filters.

merge

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

search

Get a list of inventory items that match the search criteria and

update

Update an inventory item.

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

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

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

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

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

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

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()
    ]

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)

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

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

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

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

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.

Methods:

Name Description
set_unit_category

Set unit category from category if not defined.

validate_company_string
validate_formula_fields

Ensure required fields are present for formulas.

validate_un_number

acls class-attribute instance-attribute

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

alias class-attribute instance-attribute

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

cas class-attribute instance-attribute

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

category instance-attribute

company class-attribute instance-attribute

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

description class-attribute instance-attribute

description: str | None = None

formula_id class-attribute instance-attribute

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

id class-attribute instance-attribute

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

metadata class-attribute instance-attribute

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

minimum class-attribute instance-attribute

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

name class-attribute instance-attribute

name: str | None = None

project_id class-attribute instance-attribute

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

recent_atachment_id class-attribute instance-attribute

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

security_class class-attribute instance-attribute

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

symbols class-attribute instance-attribute

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

task_config class-attribute instance-attribute

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

un_number class-attribute instance-attribute

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

unit_category class-attribute instance-attribute

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

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

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

validate_un_number classmethod

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

InventorySearchItem

Bases: BaseAlbertModel

category instance-attribute

description class-attribute instance-attribute

description: str = Field(default='')

id class-attribute instance-attribute

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

inventory_on_hand class-attribute instance-attribute

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

lots class-attribute instance-attribute

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

name class-attribute instance-attribute

name: str = Field(default='')

pictogram class-attribute instance-attribute

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

sds class-attribute instance-attribute

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

tags class-attribute instance-attribute

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

unit instance-attribute

unit: InventoryUnitCategory

InventorySpec

Bases: BaseAlbertModel

data_column_id class-attribute instance-attribute

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

data_column_name class-attribute instance-attribute

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

data_template_id class-attribute instance-attribute

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

data_template_name class-attribute instance-attribute

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

id class-attribute instance-attribute

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

name instance-attribute

name: str

spec_config class-attribute instance-attribute

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

unit_id class-attribute instance-attribute

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

unit_name class-attribute instance-attribute

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

value class-attribute instance-attribute

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

workflow_id class-attribute instance-attribute

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

workflow_name class-attribute instance-attribute

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

InventorySpecList

Bases: BaseAlbertModel

parent_id class-attribute instance-attribute

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

specs class-attribute instance-attribute

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

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.

address instance-attribute

address: str

country class-attribute instance-attribute

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

id class-attribute instance-attribute

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

latitude class-attribute instance-attribute

latitude: float = Field()

longitude class-attribute instance-attribute

longitude: float = Field()

name instance-attribute

name: str

MergeInventory

Bases: BaseAlbertModel

child_inventories class-attribute instance-attribute

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

modules class-attribute instance-attribute

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

parent_id class-attribute instance-attribute

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

OrderBy

Bases: str, Enum

ASCENDING class-attribute instance-attribute

ASCENDING = 'asc'

DESCENDING class-attribute instance-attribute

DESCENDING = 'desc'

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.

id class-attribute instance-attribute

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

location class-attribute instance-attribute

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

name class-attribute instance-attribute

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

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

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

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

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)

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)

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

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)

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

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

Methods:

Name Description
to_note_mention

Convert the user to a note mention string.

email class-attribute instance-attribute

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

id class-attribute instance-attribute

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

location class-attribute instance-attribute

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

metadata class-attribute instance-attribute

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

name instance-attribute

name: str

roles class-attribute instance-attribute

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

user_class class-attribute instance-attribute

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

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}#"