Parameters

albert.collections.parameters

AlbertHTTPError

AlbertHTTPError(response: Response)

Bases: AlbertException

Base class for all erors due to HTTP responses.

Source code in src/albert/exceptions.py

def __init__(self, response: requests.Response):
    message = self._format_message(response)
    super().__init__(message)
    self.response = response

response instance-attribute

response = response

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

OrderBy

Bases: str, Enum

ASCENDING class-attribute instance-attribute

ASCENDING = 'asc'

DESCENDING class-attribute instance-attribute

DESCENDING = 'desc'

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_by_id	Retrieve a parameter by its ID.
list	Lists parameters that match the provided criteria.
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.list(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_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())

list

list(
    *,
    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,
    return_full: bool = True,
) -> Iterator[Parameter]

Lists parameters that match the provided criteria.

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
return_full	bool	Whether to make additional API call to fetch the full object, by default True	True

Yields:

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

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

def list(
    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,
    return_full: bool = True,
) -> Iterator[Parameter]:
    """Lists parameters that match the provided criteria.

    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
    return_full : bool, optional
        Whether to make additional API call to fetch the full object, by default True

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

    def deserialize(items: list[dict]) -> Iterator[Parameter]:
        if return_full:
            for item in items:
                id = item["albertId"]
                try:
                    yield self.get_by_id(id=id)
                except AlbertHTTPError as e:
                    logger.warning(f"Error fetching Parameter '{id}': {e}")
        else:
            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,
    )

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)