Attributes

albert.collections.attributes.AttributeCollection

AttributeCollection(*, session: AlbertSession)

Bases: BaseCollection

AttributeCollection manages Attribute 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 attribute API requests.

Methods:

Name	Description
get_all	Lists all attributes with optional filters.
get_by_id	Retrieves an attribute by its ID.
get_by_ids	Retrieves multiple attributes by their IDs.
create	Creates a new attribute.
update	Updates an existing attribute.
delete	Deletes an attribute by its ID.
search	Searches for attributes.
add_values	Adds or updates reference values for a parent entity.
get_values	Retrieves reference values for a parent entity.
get_by_parent_ids	Retrieves reference values for multiple parent entities.
delete_values	Deletes specific reference values from a parent entity.
clear_values	Removes all reference values from a parent entity.

Parameters:

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

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

def __init__(self, *, session: AlbertSession):
    """Initialize the AttributeCollection.

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

base_path

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

get_all

get_all(
    *,
    category: AttributeCategory | None = None,
    start_key: str | None = None,
    max_items: int | None = None,
) -> Iterator[Attribute]

List all attributes with optional filters.

Parameters:

Name	Type	Description	Default
category	AttributeCategory	Filter attributes by category.	None
start_key	str	Pagination start key from a previous request.	None
max_items	int	Maximum number of items to return.	None

Returns:

Type	Description
Iterator[Attribute]	An iterator over Attribute entities.

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

@validate_call
def get_all(
    self,
    *,
    category: AttributeCategory | None = None,
    start_key: str | None = None,
    max_items: int | None = None,
) -> Iterator[Attribute]:
    """List all attributes with optional filters.

    Parameters
    ----------
    category : AttributeCategory, optional
        Filter attributes by category.
    start_key : str, optional
        Pagination start key from a previous request.
    max_items : int, optional
        Maximum number of items to return.

    Returns
    -------
    Iterator[Attribute]
        An iterator over Attribute entities.
    """
    params: dict[str, Any] = {}
    if category is not None:
        params["category"] = category.value
    if start_key is not None:
        params["startKey"] = start_key

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

get_by_id

get_by_id(*, id: AttributeId) -> Attribute

Retrieve an attribute by its ID.

Parameters:

Name	Type	Description	Default
id	str	The attribute ID.	required

Returns:

Type	Description
Attribute	The matching Attribute.

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

@validate_call
def get_by_id(self, *, id: AttributeId) -> Attribute:
    """Retrieve an attribute by its ID.

    Parameters
    ----------
    id : str
        The attribute ID.

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

get_by_ids

get_by_ids(*, ids: list[AttributeId]) -> list[Attribute]

Retrieve multiple attributes by their IDs.

Parameters:

Name	Type	Description	Default
ids	list[str]	A list of attribute IDs.	required

Returns:

Type	Description
list[Attribute]	The matching Attribute objects.

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

@validate_call
def get_by_ids(self, *, ids: list[AttributeId]) -> list[Attribute]:
    """Retrieve multiple attributes by their IDs.

    Parameters
    ----------
    ids : list[str]
        A list of attribute IDs.

    Returns
    -------
    list[Attribute]
        The matching Attribute objects.
    """
    response = self.session.get(f"{self.base_path}/ids", params={"id": ids})
    data = response.json()
    items = data.get("Items") or data.get("items") or []
    return [Attribute(**item) for item in items]

create

create(*, attribute: Attribute) -> Attribute

Create a new attribute.

Parameters:

Name	Type	Description	Default
attribute	Attribute	The attribute to create.	required

Returns:

Type	Description
Attribute	The created Attribute.

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

@validate_call
def create(self, *, attribute: Attribute) -> Attribute:
    """Create a new attribute.

    Parameters
    ----------
    attribute : Attribute
        The attribute to create.

    Returns
    -------
    Attribute
        The created Attribute.
    """
    payload = attribute.model_dump(
        by_alias=True, exclude_unset=True, mode="json", exclude={"id"}
    )
    response = self.session.post(self.base_path, json=payload)
    return Attribute(**response.json())

update

update(*, attribute: Attribute) -> Attribute

Update an existing attribute.

Parameters:

Name	Type	Description	Default
attribute	Attribute	The updated Attribute object. Must have an ID set.	required

Returns:

Type	Description
Attribute	The updated Attribute.

Notes

The following fields can be updated: reference_name, parameters, validation. unit_id can only be set once (when no unit is currently assigned); it cannot be changed afterwards.

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

@validate_call
def update(self, *, attribute: Attribute) -> Attribute:
    """Update an existing attribute.

    Parameters
    ----------
    attribute : Attribute
        The updated Attribute object. Must have an ID set.

    Returns
    -------
    Attribute
        The updated Attribute.

    Notes
    -----
    The following fields can be updated: ``reference_name``, ``parameters``,
    ``validation``. ``unit_id`` can only be set once (when no unit is currently
    assigned); it cannot be changed afterwards.
    """
    if attribute.id is None:
        raise ValueError("Attribute ID is required for update.")

    existing = self.get_by_id(id=attribute.id)

    enum_patches = self._generate_enum_patches(existing=existing, updated=attribute)
    if enum_patches:
        self.session.put(f"{self.base_path}/{attribute.id}/enums", json={"data": enum_patches})

    patch_payload = self._generate_attribute_patch_payload(
        existing=existing, updated=attribute, skip_validation=bool(enum_patches)
    )
    if len(patch_payload.data) > 0:
        self.session.patch(
            f"{self.base_path}/{attribute.id}",
            json=patch_payload.model_dump(by_alias=True, mode="json"),
        )

    return self.get_by_id(id=attribute.id)

delete

delete(*, id: AttributeId) -> None

Delete an attribute by its ID.

Parameters:

Name	Type	Description	Default
id	str	The attribute ID.	required

Returns:

Type	Description
None

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

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

    Parameters
    ----------
    id : str
        The attribute ID.

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

search

search(
    *,
    text: str | None = None,
    order_by: OrderBy = DESCENDING,
    sort_by: str | None = None,
    datacolumn_id: list[DataColumnId] | None = None,
    datacolumn_name: list[str] | None = None,
    parameter: list[str] | None = None,
    unit: list[str] | None = None,
    data_type: list[DataType] | None = None,
    max_items: int | None = None,
) -> Iterator[AttributeSearchItem]

Search for attributes with optional filters.

Parameters:

Name	Type	Description	Default
text	str	Full-text search term.	None
order_by	OrderBy	Sort order. Default is DESCENDING.	DESCENDING
sort_by	str	Field to sort results by.	None
datacolumn_id	list[str]	Filter by data column IDs.	None
datacolumn_name	list[str]	Filter by data column names.	None
parameter	list[str]	Filter by parameter name(s) (e.g., ["Temperature", "Pressure"]).	None
unit	list[str]	Filter by unit name(s) (e.g., ["cP", "MPa"]).	None
data_type	list[DataType]	Filter by data type(s).	None
max_items	int	Maximum number of items to return.	None

Returns:

Type	Description
Iterator[AttributeSearchItem]	An iterator over search results.

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

@validate_call
def search(
    self,
    *,
    text: str | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    sort_by: str | None = None,
    datacolumn_id: list[DataColumnId] | None = None,
    datacolumn_name: list[str] | None = None,
    parameter: list[str] | None = None,
    unit: list[str] | None = None,
    data_type: list[DataType] | None = None,
    max_items: int | None = None,
) -> Iterator[AttributeSearchItem]:
    """Search for attributes with optional filters.

    Parameters
    ----------
    text : str, optional
        Full-text search term.
    order_by : OrderBy, optional
        Sort order. Default is DESCENDING.
    sort_by : str, optional
        Field to sort results by.
    datacolumn_id : list[str], optional
        Filter by data column IDs.
    datacolumn_name : list[str], optional
        Filter by data column names.
    parameter : list[str], optional
        Filter by parameter name(s) (e.g., ``["Temperature", "Pressure"]``).
    unit : list[str], optional
        Filter by unit name(s) (e.g., ``["cP", "MPa"]``).
    data_type : list[DataType], optional
        Filter by data type(s).
    max_items : int, optional
        Maximum number of items to return.

    Returns
    -------
    Iterator[AttributeSearchItem]
        An iterator over search results.
    """
    body: dict[str, Any] = {"order": order_by}
    if text is not None:
        body["text"] = text
    if sort_by is not None:
        body["sortBy"] = sort_by
    if datacolumn_id is not None:
        body["datacolumnId"] = datacolumn_id
    if datacolumn_name is not None:
        body["datacolumnName"] = datacolumn_name
    if parameter is not None:
        body["parameter"] = parameter
    if unit is not None:
        body["unit"] = unit
    if data_type is not None:
        body["dataType"] = data_type

    yield from AlbertPaginator(
        path=f"{self.base_path}/search",
        mode=PaginationMode.OFFSET,
        session=self.session,
        deserialize=lambda items: [AttributeSearchItem(**item) for item in items],
        method="POST",
        json=body,
        max_items=max_items,
    )

add_values

add_values(
    *, parent_id: str, values: list[AttributeValue]
) -> AttributeValuesResponse

Add or update reference values for a parent entity.

If a value already exists for any of the provided attributes it is replaced. Attributes not mentioned in values are left unchanged.

Parameters:

Name	Type	Description	Default
parent_id	str	The ID of the parent entity (inventory item, lot, etc.).	required
values	list[AttributeValue]	The attribute values to add or update.	required

Returns:

Type	Description
AttributeValuesResponse	The saved attribute values with full attribute definitions.

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

@validate_call
def add_values(
    self,
    *,
    parent_id: str,
    values: list[AttributeValue],
) -> AttributeValuesResponse:
    """Add or update reference values for a parent entity.

    If a value already exists for any of the provided attributes it is
    replaced. Attributes not mentioned in ``values`` are left unchanged.

    Parameters
    ----------
    parent_id : str
        The ID of the parent entity (inventory item, lot, etc.).
    values : list[AttributeValue]
        The attribute values to add or update.

    Returns
    -------
    AttributeValuesResponse
        The saved attribute values with full attribute definitions.
    """
    attribute_ids = [v.attribute_id for v in values]
    with suppress(NotFoundError):
        self.delete_values(parent_id=parent_id, attribute_ids=attribute_ids)
    payload = [v.model_dump(by_alias=True, mode="json", exclude_none=True) for v in values]
    response = self.session.put(f"{self.base_path}/values/{parent_id}", json=payload)
    return AttributeValuesResponse(**response.json())

get_values

get_values(
    *,
    parent_id: str,
    scope: AttributeScope | None = None,
    start_key: str | None = None,
    max_items: int | None = None,
) -> Iterator[AttributeValuesResponse]

Retrieve reference values for a parent entity.

Parameters:

Name	Type	Description	Default
parent_id	str	The ID of the parent entity.	required
scope	AttributeScope	Defines which entities to fetch values for. SELF (default) — the parent entity only. LOT — lot entities under the parent (inventory parents only). ALL — parent and all child entities.	None
start_key	str	Pagination start key from a previous request.	None
max_items	int	Maximum number of items to return.	None

Returns:

Type	Description
Iterator[AttributeValuesResponse]	An iterator over attribute value responses, one per entity.

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

@validate_call
def get_values(
    self,
    *,
    parent_id: str,
    scope: AttributeScope | None = None,
    start_key: str | None = None,
    max_items: int | None = None,
) -> Iterator[AttributeValuesResponse]:
    """Retrieve reference values for a parent entity.

    Parameters
    ----------
    parent_id : str
        The ID of the parent entity.
    scope : AttributeScope, optional
        Defines which entities to fetch values for.
        ``SELF`` (default) — the parent entity only.
        ``LOT`` — lot entities under the parent (inventory parents only).
        ``ALL`` — parent and all child entities.
    start_key : str, optional
        Pagination start key from a previous request.
    max_items : int, optional
        Maximum number of items to return.

    Returns
    -------
    Iterator[AttributeValuesResponse]
        An iterator over attribute value responses, one per entity.
    """
    params: dict[str, Any] = {}
    if scope is not None:
        params["scope"] = scope.value
    if start_key is not None:
        params["startKey"] = start_key

    yield from AlbertPaginator(
        path=f"{self.base_path}/values/{parent_id}",
        mode=PaginationMode.KEY,
        session=self.session,
        deserialize=lambda items: [AttributeValuesResponse(**item) for item in items],
        params=params,
        max_items=max_items,
    )

get_by_parent_ids

get_by_parent_ids(
    *, parent_ids: list[str], max_items: int | None = None
) -> list[AttributeValuesResponse]

Retrieve reference values for multiple parent entities.

Parameters:

Name	Type	Description	Default
parent_ids	list[str]	The IDs of the parent entities to fetch values for.	required
max_items	int | None	Maximum total number of attribute value items to return across all parents. When None (default), all pages are fetched for every parent.	None

Returns:

Type	Description
list[AttributeValuesResponse]	Attribute values for each parent entity that has values set.

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

@validate_call
def get_by_parent_ids(
    self,
    *,
    parent_ids: list[str],
    max_items: int | None = None,
) -> list[AttributeValuesResponse]:
    """Retrieve reference values for multiple parent entities.

    Parameters
    ----------
    parent_ids : list[str]
        The IDs of the parent entities to fetch values for.
    max_items : int | None, optional
        Maximum total number of attribute value items to return across all
        parents. When None (default), all pages are fetched for every parent.

    Returns
    -------
    list[AttributeValuesResponse]
        Attribute values for each parent entity that has values set.
    """
    pending: list[dict] = [{"parentId": pid} for pid in parent_ids]
    accumulated: dict[str, list[AttributeValuesResponseItem]] = {}
    total = 0

    while pending:
        response = self.session.post(f"{self.base_path}/values", json=pending)
        next_pending: list[dict] = []
        for item in response.json().get("items") or []:
            parsed = AttributeValuesResponse.model_validate(item)
            accumulated.setdefault(parsed.parent_id, []).extend(parsed.attributes)
            total += len(parsed.attributes)
            if last_key := item.get("lastKey"):
                next_pending.append({"parentId": parsed.parent_id, "startKey": last_key})
        pending = next_pending
        if max_items is not None and total >= max_items:
            break

    return [
        AttributeValuesResponse.model_validate({"parentId": pid, "attributes": attrs})
        for pid, attrs in accumulated.items()
    ]

delete_values

delete_values(
    *,
    parent_id: str,
    attribute_ids: list[AttributeId],
    scope: AttributeScope | None = None,
) -> None

Delete specific reference values from a parent entity.

Parameters:

Name	Type	Description	Default
parent_id	str	The ID of the parent entity.	required
attribute_ids	list[str]	The attribute IDs whose values should be removed.	required
scope	AttributeScope	Scope of deletion. Defaults to SELF.	None

Returns:

Type	Description
None

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

@validate_call
def delete_values(
    self,
    *,
    parent_id: str,
    attribute_ids: list[AttributeId],
    scope: AttributeScope | None = None,
) -> None:
    """Delete specific reference values from a parent entity.

    Parameters
    ----------
    parent_id : str
        The ID of the parent entity.
    attribute_ids : list[str]
        The attribute IDs whose values should be removed.
    scope : AttributeScope, optional
        Scope of deletion. Defaults to ``SELF``.

    Returns
    -------
    None
    """
    params: dict[str, Any] = {"attributeId": attribute_ids}
    if scope is not None:
        params["scope"] = scope.value
    self.session.delete(f"{self.base_path}/values/{parent_id}", params=params)

clear_values

clear_values(
    *, parent_id: str, scope: AttributeScope | None = None
) -> None

Remove all reference values from a parent entity.

Parameters:

Name	Type	Description	Default
parent_id	str	The ID of the parent entity.	required
scope	AttributeScope	Scope of deletion. Defaults to SELF.	None

Returns:

Type	Description
None

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

@validate_call
def clear_values(
    self,
    *,
    parent_id: str,
    scope: AttributeScope | None = None,
) -> None:
    """Remove all reference values from a parent entity.

    Parameters
    ----------
    parent_id : str
        The ID of the parent entity.
    scope : AttributeScope, optional
        Scope of deletion. Defaults to ``SELF``.

    Returns
    -------
    None
    """
    params: dict[str, Any] = {}
    if scope is not None:
        params["scope"] = scope.value
    self.session.delete(f"{self.base_path}/values/{parent_id}", params=params)