Units

albert.collections.units

AlbertPaginator

AlbertPaginator(
    *,
    path: str,
    mode: PaginationMode,
    session: AlbertSession,
    deserialize: Callable[
        [Iterable[dict]], Iterable[ItemType]
    ],
    params: dict[str, str] | None = None,
)

Bases: Iterator[ItemType]

Helper class for pagination through Albert endpoints.

Two pagination modes are possible: - Offset-based via by the offset query parameter - Key-based via by the startKey query parameter and 'lastKey' response field

A custom deserialize function is provided when additional logic is required to load the raw items returned by the search listing, e.g., making additional Albert API calls.

Source code in src/albert/core/pagination.py

def __init__(
    self,
    *,
    path: str,
    mode: PaginationMode,
    session: AlbertSession,
    deserialize: Callable[[Iterable[dict]], Iterable[ItemType]],
    params: dict[str, str] | None = None,
):
    self.path = path
    self.mode = mode
    self.session = session
    self.deserialize = deserialize

    params = params or {}
    self.params = {k: v for k, v in params.items() if v is not None}

    if "startKey" in self.params:
        self.params["startKey"] = quote_plus(self.params["startKey"])

    self._iterator = self._create_iterator()

deserialize instance-attribute

deserialize = deserialize

mode instance-attribute

mode = mode

params instance-attribute

params = {k: _0for (k, v) in items() if v is not None}

path instance-attribute

path = path

session instance-attribute

session = session

AlbertSession

AlbertSession(
    *,
    base_url: str,
    token: str | None = None,
    auth_manager: AlbertClientCredentials
    | AlbertSSOClient
    | 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 relative request paths (e.g., "https://app.albertinvent.com").	required
token	str | None	A static JWT token for authentication. Ignored if auth_manager is provided.	None
auth_manager	AlbertClientCredentials | AlbertSSOClient	An authentication manager used to dynamically fetch and refresh tokens. If provided, it overrides token.	None
retries	int	The number of automatic retries on failed requests (default is 3).	None

Methods:

Name	Description
request

Source code in src/albert/core/session.py

def __init__(
    self,
    *,
    base_url: str,
    token: str | None = None,
    auth_manager: AlbertClientCredentials | AlbertSSOClient | 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 auth_manager is None:
        raise ValueError("Either `token` or `auth_manager` must be specified.")

    self._auth_manager = auth_manager
    self._provided_token = token

    # 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/core/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

OrderBy

Bases: str, Enum

ASCENDING class-attribute instance-attribute

ASCENDING = 'asc'

DESCENDING class-attribute instance-attribute

DESCENDING = 'desc'

PaginationMode

Bases: str, Enum

KEY class-attribute instance-attribute

KEY = 'key'

OFFSET class-attribute instance-attribute

OFFSET = 'offset'

Unit

Bases: BaseResource

Unit is a Pydantic model representing a unit entity.

Attributes:

Name	Type	Description
id	str | None	The Albert ID of the unit. Set when the unit is retrieved from Albert.
name	str	The name of the unit.
symbol	str | None	The symbol of the unit.
synonyms	List[str] | None	The list of synonyms for the unit.
category	UnitCategory	The category of the unit.
verified	bool | None	Whether the unit is verified.
status	Status | None	The status of the unit. Allowed values are active, and inactive

category class-attribute instance-attribute

category: UnitCategory | None = Field(None)

id class-attribute instance-attribute

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

name instance-attribute

name: str

symbol class-attribute instance-attribute

symbol: str | None = Field(None)

synonyms class-attribute instance-attribute

synonyms: list[str] | None = Field(
    default_factory=list, alias="Synonyms"
)

verified class-attribute instance-attribute

verified: bool | None = Field(
    default=False, exclude=True, frozen=True
)

UnitCategory

Bases: str, Enum

UnitCategory is an enumeration of possible unit categories.

Attributes:

Name	Type	Description
LENGTH	str	Represents length units.
VOLUME	str	Represents volume units.
LIQUID_VOLUME	str	Represents liquid volume units.
ANGLES	str	Represents angle units.
TIME	str	Represents time units.
FREQUENCY	str	Represents frequency units.
MASS	str	Represents mass units.
CURRENT	str	Represents electric current units.
TEMPERATURE	str	Represents temperature units.
AMOUNT	str	Represents amount of substance units.
LUMINOSITY	str	Represents luminous intensity units.
FORCE	str	Represents force units.
ENERGY	str	Represents energy units.
POWER	str	Represents power units.
PRESSURE	str	Represents pressure units.
ELECTRICITY_AND_MAGNETISM	str	Represents electricity and magnetism units.
OTHER	str	Represents other units.
WEIGHT	str	Represents weight units.

AMOUNT class-attribute instance-attribute

AMOUNT = 'Amount of substance'

ANGLES class-attribute instance-attribute

ANGLES = 'Angles'

AREA class-attribute instance-attribute

AREA = 'Area'

BINARY class-attribute instance-attribute

BINARY = 'Binary'

CAPACITANCE class-attribute instance-attribute

CAPACITANCE = 'Capacitance'

CURRENT class-attribute instance-attribute

CURRENT = 'Electric current'

DENSITY class-attribute instance-attribute

DENSITY = 'Density'

ELECTRICAL_CONDUCTIVITY class-attribute instance-attribute

ELECTRICAL_CONDUCTIVITY = 'Electrical conductivity'

ELECTRICAL_PERMITTIVITY class-attribute instance-attribute

ELECTRICAL_PERMITTIVITY = 'Electrical permitivitty'

ELECTRICITY_AND_MAGNETISM class-attribute instance-attribute

ELECTRICITY_AND_MAGNETISM = 'Electricity and magnetism'

ENERGY class-attribute instance-attribute

ENERGY = 'Energy'

FORCE class-attribute instance-attribute

FORCE = 'Force'

FREQUENCY class-attribute instance-attribute

FREQUENCY = 'Frequency'

LENGTH class-attribute instance-attribute

LENGTH = 'Length'

LIQUID_VOLUME class-attribute instance-attribute

LIQUID_VOLUME = 'Liquid volume'

LUMINOSITY class-attribute instance-attribute

LUMINOSITY = 'Luminous intensity'

MASS class-attribute instance-attribute

MASS = 'Mass'

OTHER class-attribute instance-attribute

OTHER = 'Other'

POWER class-attribute instance-attribute

POWER = 'Power'

PRESSURE class-attribute instance-attribute

PRESSURE = 'Pressure'

RESISTANCE class-attribute instance-attribute

RESISTANCE = 'Resistance'

SPEED class-attribute instance-attribute

SPEED = 'Speed'

SURFACE_AREA class-attribute instance-attribute

SURFACE_AREA = 'Surface Area'

TEMPERATURE class-attribute instance-attribute

TEMPERATURE = 'Temperature'

TIME class-attribute instance-attribute

TIME = 'Time'

VOLUME class-attribute instance-attribute

VOLUME = 'Volume'

WEIGHT class-attribute instance-attribute

WEIGHT = 'Weight'

UnitCollection

UnitCollection(*, session: AlbertSession)

Bases: BaseCollection

UnitCollection is a collection class for managing Unit entities in the Albert platform.

Parameters:

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

Methods:

Name	Description
create	Creates a new unit entity.
delete	Deletes a unit by its ID.
exists	Checks if a unit exists by its name.
get_all	Get all unit entities with optional filters.
get_by_id	Retrieves a unit by its ID.
get_by_ids	Retrieves a set of units by their IDs
get_by_name	Retrieves a unit by its name.
update	Updates a unit entity by its ID.

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

def __init__(self, *, session: AlbertSession):
    """
    Initializes the UnitCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, unit: Unit) -> Unit

Creates a new unit entity.

Parameters:

Name	Type	Description	Default
unit	Unit	The unit object to create.	required

Returns:

Type	Description
Unit	The created Unit object.

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

def create(self, *, unit: Unit) -> Unit:
    """
    Creates a new unit entity.

    Parameters
    ----------
    unit : Unit
        The unit object to create.

    Returns
    -------
    Unit
        The created Unit object.
    """
    hit = self.get_by_name(name=unit.name, exact_match=True)
    if hit is not None:
        logging.warning(
            f"Unit with the name {hit.name} already exists. Returning the existing unit."
        )
        return hit
    response = self.session.post(
        self.base_path, json=unit.model_dump(by_alias=True, exclude_unset=True, mode="json")
    )
    this_unit = Unit(**response.json())
    return this_unit

delete

delete(*, id: str) -> None

Deletes a unit by its ID.

Parameters:

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

Returns:

Type	Description
None

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

def delete(self, *, id: str) -> None:
    """
    Deletes a unit by its ID.

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

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

exists

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

Checks if a unit exists by its name.

Parameters:

Name	Type	Description	Default
name	str	The name of the unit to check.	required
exact_match	bool	Whether to match the name exactly, by default True.	True

Returns:

Type	Description
bool	True if the unit exists, False otherwise.

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

def exists(self, *, name: str, exact_match: bool = True) -> bool:
    """
    Checks if a unit exists by its name.

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

    Returns
    -------
    bool
        True if the unit exists, False otherwise.
    """
    return self.get_by_name(name=name, exact_match=exact_match) is not None

get_all

get_all(
    *,
    limit: int = 100,
    name: str | list[str] | None = None,
    category: UnitCategory | None = None,
    order_by: OrderBy = DESCENDING,
    exact_match: bool = False,
    start_key: str | None = None,
    verified: bool | None = None,
) -> Iterator[Unit]

Get all unit entities with optional filters.

Parameters:

Name	Type	Description	Default
limit	int	The maximum number of units to return, by default 50.	100
name	Optional[str]	The name of the unit to filter by, by default None.	None
category	Optional[UnitCategory]	The category of the unit to filter by, by default None.	None
order_by	OrderBy	The order by which to sort the results, by default OrderBy.DESCENDING.	DESCENDING
exact_match	bool	Whether to match the name exactly, by default False.	False
start_key	Optional[str]	The starting point for the next set of results, by default None.	None

Returns:

Type	Description
Iterator[Unit]	An iterator of Unit objects.

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

def get_all(
    self,
    *,
    limit: int = 100,
    name: str | list[str] | None = None,
    category: UnitCategory | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    exact_match: bool = False,
    start_key: str | None = None,
    verified: bool | None = None,
) -> Iterator[Unit]:
    """
    Get all unit entities with optional filters.

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

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

get_by_id

get_by_id(*, id: str) -> Unit

Retrieves a unit by its ID.

Parameters:

Name	Type	Description	Default
id	str	The ID of the unit to retrieve.	required

Returns:

Type	Description
Unit	The Unit object if found.

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

def get_by_id(self, *, id: str) -> Unit:
    """
    Retrieves a unit by its ID.

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

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

get_by_ids

get_by_ids(*, ids: list[str]) -> list[Unit]

Retrieves a set of units by their IDs

Parameters:

Name	Type	Description	Default
ids	list[str]	The IDs of the units to retrieve.	required

Returns:

Type	Description
list[Unit]	The Unit objects

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

def get_by_ids(self, *, ids: list[str]) -> list[Unit]:
    """
    Retrieves a set of units by their IDs

    Parameters
    ----------
    ids : list[str]
        The IDs of the units to retrieve.

    Returns
    -------
    list[Unit]
        The Unit objects
    """
    url = f"{self.base_path}/ids"
    batches = [ids[i : i + 500] for i in range(0, len(ids), 500)]
    return [
        Unit(**item)
        for batch in batches
        for item in self.session.get(url, params={"id": batch}).json()["Items"]
    ]

get_by_name

get_by_name(
    *, name: str, exact_match: bool = False
) -> Unit | None

Retrieves a unit by its name.

Parameters:

Name	Type	Description	Default
name	str	The name of the unit to retrieve.	required
exact_match	bool	Whether to match the name exactly, by default False.	False

Returns:

Type	Description
Optional[Unit]	The Unit object if found, None otherwise.

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

def get_by_name(self, *, name: str, exact_match: bool = False) -> Unit | None:
    """
    Retrieves a unit by its name.

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

    Returns
    -------
    Optional[Unit]
        The Unit object if found, None otherwise.
    """
    found = self.get_all(name=name, exact_match=exact_match)
    return next(found, None)

update

update(*, unit: Unit) -> Unit

Updates a unit entity by its ID.

Parameters:

Name	Type	Description	Default
unit	Unit	The updated Unit object.	required

Returns:

Type	Description
Unit	The updated Unit

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

def update(self, *, unit: Unit) -> Unit:
    """
    Updates a unit entity by its ID.

    Parameters
    ----------
    unit : Unit
        The updated Unit object.

    Returns
    -------
    Unit
        The updated Unit
    """
    unit_id = unit.id
    original_unit = self.get_by_id(id=unit_id)
    payload = self._generate_patch_payload(existing=original_unit, updated=unit)
    url = f"{self.base_path}/{unit_id}"
    self.session.patch(url, json=payload.model_dump(mode="json", by_alias=True))
    unit = self.get_by_id(id=unit_id)
    return unit