Parameters

albert.collections.parameters

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: _lfor (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'

Parameter

Bases: BaseResource

A parameter in Albert.

Attributes:

Name	Type	Description
name	str	The name of the parameter. Names must be unique.
id	str | None	The Albert ID of the parameter. Set when the parameter is retrieved from Albert.
category	ParameterCategory	The category of the parameter. Allowed values are Normal and Special. Read-only.
rank	int	The rank of the returned parameter. Read-only.

category class-attribute instance-attribute

category: ParameterCategory | None = Field(
    default=None, exclude=True, frozen=True
)

id class-attribute instance-attribute

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

metadata class-attribute instance-attribute

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

name instance-attribute

name: str

rank class-attribute instance-attribute

rank: int | None = Field(
    default=None, exclude=True, frozen=True
)

ParameterCollection

ParameterCollection(*, session: AlbertSession)

Bases: BaseCollection

ParameterCollection is a collection class for managing Parameter entities in the Albert platform.

Parameters:

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

Methods:

Name	Description
create	Create a new parameter.
delete	Delete a parameter by its ID.
get_all	Retrieve all Parameter items with optional filters.
get_by_id	Retrieve a parameter by its ID.
update	Update a parameter.

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

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

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

base_path instance-attribute

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

create

create(*, parameter: Parameter) -> Parameter

Create a new parameter.

Parameters:

Name	Type	Description	Default
parameter	Parameter	The parameter to create.	required

Returns:

Type	Description
Parameter	Returns the created parameter or the existing parameter if it already exists.

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

def create(self, *, parameter: Parameter) -> Parameter:
    """Create a new parameter.

    Parameters
    ----------
    parameter : Parameter
        The parameter to create.

    Returns
    -------
    Parameter
        Returns the created parameter or the existing parameter if it already exists.
    """
    match = next(self.get_all(names=parameter.name, exact_match=True), None)
    if match is not None:
        logging.warning(
            f"Parameter with name {parameter.name} already exists. Returning existing parameter."
        )
        return match
    response = self.session.post(
        self.base_path,
        json=parameter.model_dump(by_alias=True, exclude_none=True, mode="json"),
    )
    return Parameter(**response.json())

delete

delete(*, id: str) -> None

Delete a parameter by its ID.

Parameters:

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

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

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

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

get_all

get_all(
    *,
    ids: list[str] | None = None,
    names: str | list[str] = None,
    exact_match: bool = False,
    order_by: OrderBy = DESCENDING,
    start_key: str | None = None,
    limit: int = 50,
) -> Iterator[Parameter]

Retrieve all Parameter items with optional filters.

Parameters:

Name	Type	Description	Default
ids	list[str] | None	A list of parameter IDs to retrieve, by default None	None
names	str | list[str]	A list of parameter names to retrieve, by default None	None
exact_match	bool	Whether to match the name exactly, by default False	False
order_by	OrderBy	The order in which to return results, by default OrderBy.DESCENDING	DESCENDING

Yields:

Type	Description
Iterator[Parameter]	An iterator of Parameters matching the given criteria.

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

def get_all(
    self,
    *,
    ids: list[str] | None = None,
    names: str | list[str] = None,
    exact_match: bool = False,
    order_by: OrderBy = OrderBy.DESCENDING,
    start_key: str | None = None,
    limit: int = 50,
) -> Iterator[Parameter]:
    """
    Retrieve all Parameter items with optional filters.

    Parameters
    ----------
    ids : list[str] | None, optional
        A list of parameter IDs to retrieve, by default None
    names : str | list[str], optional
        A list of parameter names to retrieve, by default None
    exact_match : bool, optional
        Whether to match the name exactly, by default False
    order_by : OrderBy, optional
        The order in which to return results, by default OrderBy.DESCENDING

    Yields
    ------
    Iterator[Parameter]
        An iterator of Parameters matching the given criteria.
    """

    def deserialize(items: list[dict]) -> Iterator[Parameter]:
        yield from (Parameter(**item) for item in items)

    params = {"limit": limit, "orderBy": order_by, "parameters": ids, "startKey": start_key}
    if names:
        params["name"] = [names] if isinstance(names, str) else names
        params["exactMatch"] = json.dumps(exact_match)

    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=deserialize,
    )

get_by_id

get_by_id(*, id: str) -> Parameter

Retrieve a parameter by its ID.

Parameters:

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

Returns:

Type	Description
Parameter	The parameter with the given ID.

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

def get_by_id(self, *, id: str) -> Parameter:
    """Retrieve a parameter by its ID.

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

    Returns
    -------
    Parameter
        The parameter with the given ID.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return Parameter(**response.json())

update

update(*, parameter: Parameter) -> Parameter

Update a parameter.

Parameters:

Name	Type	Description	Default
parameter	Parameter	The updated parameter to save. The parameter must have an ID.	required

Returns:

Type	Description
Parameter	The updated parameter as returned by the server.

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

def update(self, *, parameter: Parameter) -> Parameter:
    """Update a parameter.

    Parameters
    ----------
    parameter : Parameter
        The updated parameter to save. The parameter must have an ID.

    Returns
    -------
    Parameter
        The updated parameter as returned by the server.
    """
    existing = self.get_by_id(id=parameter.id)
    payload = self._generate_patch_payload(
        existing=existing,
        updated=parameter,
    )
    payload_dump = payload.model_dump(mode="json", by_alias=True)
    for i, change in enumerate(payload_dump["data"]):
        if not self._is_metadata_item_list(
            existing_object=existing,
            updated_object=parameter,
            metadata_field=change["attribute"],
        ):
            change["operation"] = "update"
            if "newValue" in change and change["newValue"] is None:
                del change["newValue"]
            if "oldValue" in change and change["oldValue"] is None:
                del change["oldValue"]
            payload_dump["data"][i] = change
    if len(payload_dump["data"]) == 0:
        return parameter
    for e in payload_dump["data"]:
        self.session.patch(
            f"{self.base_path}/{parameter.id}",
            json={"data": [e]},
        )
    return self.get_by_id(id=parameter.id)