Parameter Groups

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

EnumValidationValue

Bases: BaseAlbertModel

Represents a value for an enum type validation.

Attributes:

Name	Type	Description
text	str	The text of the enum value.
id	str | None	The ID of the enum value. If not provided, the ID will be generated upon creation.

id class-attribute instance-attribute

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

original_text class-attribute instance-attribute

original_text: str | None = Field(
    default=None,
    exclude=True,
    frozen=True,
    alias="originalText",
)

text class-attribute instance-attribute

text: str = Field()

OrderBy

Bases: str, Enum

ASCENDING class-attribute instance-attribute

ASCENDING = 'asc'

DESCENDING class-attribute instance-attribute

DESCENDING = 'desc'

PGPatchDatum

Bases: PatchDatum

rowId class-attribute instance-attribute

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

PGPatchPayload

Bases: PatchPayload

A payload for a PATCH request to update a parameter group.

Attributes:

Name	Type	Description
data	list[PGPatchDatum]	The data to be updated in the parameter group.

data class-attribute instance-attribute

data: list[PGPatchDatum | PatchDatum] = Field(
    default_factory=list
)

PGType

Bases: str, Enum

The type of a parameter group

BATCH class-attribute instance-attribute

BATCH = 'batch'

GENERAL class-attribute instance-attribute

GENERAL = 'general'

PROPERTY class-attribute instance-attribute

PROPERTY = 'property'

ParameterGroup

Bases: BaseTaggedEntity

Use 'Standards' key in metadata to store standards

acl class-attribute instance-attribute

acl: list[SerializeAsEntityLink[User]] | None = Field(
    default=None, alias="ACL"
)

description class-attribute instance-attribute

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

documents class-attribute instance-attribute

documents: list[EntityLink] = Field(
    default_factory=list, 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] = Field(
    alias="Metadata", default_factory=dict
)

name instance-attribute

name: str

parameters class-attribute instance-attribute

parameters: list[ParameterValue] = Field(
    default_factory=list, alias="Parameters"
)

security_class class-attribute instance-attribute

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

type class-attribute instance-attribute

type: PGType | None = Field(default=None)

verified class-attribute instance-attribute

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

ParameterGroupCollection

ParameterGroupCollection(*, session: AlbertSession)

Bases: BaseCollection

ParameterGroupCollection is a collection class for managing ParameterGroup entities in the Albert platform.

Parameters:

Name	Type	Description	Default
session	AlbertSession	The Albert session to use for making requests.	required

Methods:

Name	Description
create	Create a new parameter group.
delete	Delete a parameter group by its ID.
get_by_id	Get a parameter group by its ID.
get_by_ids
get_by_name	Get a parameter group by its name.
list	Search for Parameter Groups matching the given criteria.
update	Update a parameter group.

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

def __init__(self, *, session: AlbertSession):
    """A collection for interacting with Albert parameter groups.

    Parameters
    ----------
    session : AlbertSession
        The Albert session to use for making requests.
    """
    super().__init__(session=session)
    self.base_path = f"/api/{ParameterGroupCollection._api_version}/parametergroups"

base_path instance-attribute

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

create

create(
    *, parameter_group: ParameterGroup
) -> ParameterGroup

Create a new parameter group.

Parameters:

Name	Type	Description	Default
parameter_group	ParameterGroup	The parameter group to create.	required

Returns:

Type	Description
ParameterGroup	The created parameter group.

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

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

    Parameters
    ----------
    parameter_group : ParameterGroup
        The parameter group to create.

    Returns
    -------
    ParameterGroup
        The created parameter group.
    """

    response = self.session.post(
        self.base_path,
        json=parameter_group.model_dump(by_alias=True, exclude_none=True, mode="json"),
    )
    return ParameterGroup(**response.json())

delete

delete(*, id: str) -> None

Delete a parameter group by its ID.

Parameters:

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

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

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

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

get_by_id

get_by_id(*, id: str) -> ParameterGroup

Get a parameter group by its ID.

Parameters:

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

Returns:

Type	Description
ParameterGroup	The parameter group with the given ID.

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

def get_by_id(self, *, id: str) -> ParameterGroup:
    """Get a parameter group by its ID.

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

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

get_by_ids

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

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

def get_by_ids(self, *, ids: list[str]) -> ParameterGroup:
    url = f"{self.base_path}/ids"
    batches = [ids[i : i + 100] for i in range(0, len(ids), 100)]
    return [
        ParameterGroup(**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) -> ParameterGroup | None

Get a parameter group by its name.

Parameters:

Name	Type	Description	Default
name	str	The name of the parameter group to retrieve.	required

Returns:

Type	Description
ParameterGroup | None	The parameter group with the given name, or None if not found.

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

def get_by_name(self, *, name: str) -> ParameterGroup | None:
    """Get a parameter group by its name.

    Parameters
    ----------
    name : str
        The name of the parameter group to retrieve.

    Returns
    -------
    ParameterGroup | None
        The parameter group with the given name, or None if not found.
    """
    matches = self.list(text=name)
    for m in matches:
        if m.name.lower() == name.lower():
            return m
    return None

list

list(
    *,
    text: str | None = None,
    types: PGType | list[PGType] | None = None,
    order_by: OrderBy = DESCENDING,
    limit: int = 25,
    offset: int | None = None,
) -> Iterator[ParameterGroup]

Search for Parameter Groups matching the given criteria.

Parameters:

Name	Type	Description	Default
text	str | None	Text to search for, by default None	None
types	PGType | list[PGType] | None	Filer the returned Parameter Groups by Type, by default None	None
order_by	OrderBy	The order in which to return results, by default OrderBy.DESCENDING	DESCENDING

Yields:

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

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

def list(
    self,
    *,
    text: str | None = None,
    types: PGType | list[PGType] | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    limit: int = 25,
    offset: int | None = None,
) -> Iterator[ParameterGroup]:
    """Search for Parameter Groups matching the given criteria.

    Parameters
    ----------
    text : str | None, optional
        Text to search for, by default None
    types : PGType | list[PGType] | None, optional
        Filer the returned Parameter Groups by Type, by default None
    order_by : OrderBy, optional
        The order in which to return results, by default OrderBy.DESCENDING

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

    def deserialize(items: list[dict]) -> Iterator[ParameterGroup]:
        for item in items:
            id = item["albertId"]
            try:
                yield self.get_by_id(id=id)
            except AlbertHTTPError as e:  # pragma: no cover
                logger.warning(f"Error fetching parameter group {id}: {e}")
        # Currently, the API is not returning metadata for the list_by_ids endpoint, so we need to fetch individually until that is fixed
        # return self.get_by_ids(ids=[x["albertId"] for x in items])

    params = {
        "limit": limit,
        "offset": offset,
        "order": order_by.value,
        "text": text,
        "types": [types] if isinstance(types, PGType) else types,
    }

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        params=params,
        deserialize=deserialize,
    )

update

update(
    *, parameter_group: ParameterGroup
) -> ParameterGroup

Update a parameter group.

Parameters:

Name	Type	Description	Default
parameter_group	ParameterGroup	The updated ParameterGroup. The ParameterGroup must have an ID.	required

Returns:

Type	Description
ParameterGroup	The updated ParameterGroup as returned by the server.

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

def update(self, *, parameter_group: ParameterGroup) -> ParameterGroup:
    """Update a parameter group.

    Parameters
    ----------
    parameter_group : ParameterGroup
        The updated ParameterGroup. The ParameterGroup must have an ID.

    Returns
    -------
    ParameterGroup
        The updated ParameterGroup as returned by the server.
    """

    existing = self.get_by_id(id=parameter_group.id)
    path = f"{self.base_path}/{existing.id}"

    payload = self._generate_patch_payload(existing=existing, updated=parameter_group)
    # need to use a different payload for the special update parameters
    payload = PGPatchPayload(
        data=payload.data,
    )

    # Handle special update parameters
    special_patches, special_enum_patches, new_param_patches = (
        self._handle_special_update_parameters(existing=existing, updated=parameter_group)
    )

    payload.data.extend(special_patches)
    if len(payload.data) > 0:
        self.session.patch(
            path, json=payload.model_dump(mode="json", by_alias=True, exclude_none=True)
        )

    # handle adding new parameters
    if len(new_param_patches) > 0:
        self.session.put(
            f"{self.base_path}/{existing.id}/parameters",
            json={"Parameters": new_param_patches},
        )
    # Handle special enum update parameters
    for sequence, enum_patches in special_enum_patches.items():
        if len(enum_patches) == 0:
            continue
        enum_path = f"{self.base_path}/{existing.id}/parameters/{sequence}/enums"
        self.session.put(enum_path, json=enum_patches)
    return self.get_by_id(id=parameter_group.id)