Parameter Groups

albert.collections.parameter_groups

Functions:

Name	Description
generate_parameter_group_patches

logger module-attribute

logger = create_logger()

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

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

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'

PaginationMode

Bases: str, Enum

KEY class-attribute instance-attribute

KEY = 'key'

OFFSET class-attribute instance-attribute

OFFSET = 'offset'

ParameterGroup

Bases: BaseTaggedResource

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_all	Search and hydrate all Parameter Groups matching the given criteria.
get_by_id	Get a parameter group by its ID.
get_by_ids
get_by_name	Get a parameter group by its name.
search	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_all

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

Search and hydrate all 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	Filter 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
limit	int	Page size for each search request, by default 25.	25
offset	int | None	Offset to start from, by default None.	None

Yields:

Type	Description
Iterator[ParameterGroup]	An iterator of fully hydrated Parameter Groups.

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

def get_all(
    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 and hydrate all Parameter Groups matching the given criteria.

    Parameters
    ----------
    text : str | None, optional
        Text to search for, by default None.
    types : PGType | list[PGType] | None, optional
        Filter the returned Parameter Groups by Type, by default None.
    order_by : OrderBy, optional
        The order in which to return results, by default OrderBy.DESCENDING.
    limit : int, optional
        Page size for each search request, by default 25.
    offset : int | None, optional
        Offset to start from, by default None.

    Yields
    ------
    Iterator[ParameterGroup]
        An iterator of fully hydrated Parameter Groups.
    """
    for item in self.search(
        text=text,
        types=types,
        order_by=order_by,
        limit=limit,
        offset=offset,
    ):
        try:
            # Currently, the API is not returning Metadata, Tags, Documents, and ACL for the get_by_ids endpoint, so we need to fetch individually until that is fixed
            yield self.get_by_id(id=item.id)
        except AlbertHTTPError as e:  # pragma: no cover
            logger.warning(f"Error fetching parameter group {item.id}: {e}")

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.get_all(text=name)
    # TODO: optimize with explicit hydrate() after self.search()
    for m in matches:
        if m.name.lower() == name.lower():
            return m
    return None

search

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

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 search(
    self,
    *,
    text: str | None = None,
    types: PGType | list[PGType] | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    limit: int = 25,
    offset: int | None = None,
) -> Iterator[ParameterGroupSearchItem]:
    """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.
    """

    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=lambda items: [
            ParameterGroupSearchItem(**item)._bind_collection(self) for item in items
        ],
    )

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}"

    base_payload = self._generate_patch_payload(
        existing=existing, updated=parameter_group, generate_metadata_diff=True
    )

    general_patches, new_parameter_values, enum_patches = generate_parameter_group_patches(
        initial_patches=base_payload,
        updated_parameter_group=parameter_group,
        existing_parameter_group=existing,
    )

    # add new parameters
    new_param_url = f"{self.base_path}/{parameter_group.id}/parameters"
    if len(new_parameter_values) > 0:
        self.session.put(
            url=new_param_url,
            json=[
                x.model_dump(mode="json", by_alias=True, exclude_none=True)
                for x in new_parameter_values
            ],
        )
    new_param_sequences = [x.sequence for x in new_parameter_values]
    # handle enum updates
    for sequence, ep in enum_patches.items():
        if sequence in new_param_sequences:
            # we don't need to handle enum updates for new parameters
            continue
        if len(ep) > 0:
            enum_url = f"{self.base_path}/{parameter_group.id}/parameters/{sequence}/enums"
            self.session.put(
                url=enum_url,
                json=ep,
            )
    if len(general_patches.data) > 0:
        # patch the general patches
        self.session.patch(
            url=path,
            json=general_patches.model_dump(mode="json", by_alias=True, exclude_none=True),
        )

    return self.get_by_id(id=parameter_group.id)

ParameterGroupSearchItem pydantic-model

Bases: BaseAlbertModel, HydrationMixin[ParameterGroup]

Lightweight representation of a ParameterGroup returned from unhydrated search().

Show JSON schema:

{
  "$defs": {
    "LocalizedNames": {
      "properties": {
        "de": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "De"
        },
        "ja": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Ja"
        },
        "zh": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Zh"
        },
        "es": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Es"
        }
      },
      "title": "LocalizedNames",
      "type": "object"
    },
    "PGType": {
      "description": "The type of a parameter group",
      "enum": [
        "general",
        "batch",
        "property"
      ],
      "title": "PGType",
      "type": "string"
    },
    "ParameterSearchItemParameter": {
      "properties": {
        "name": {
          "title": "Name",
          "type": "string"
        },
        "id": {
          "title": "Id",
          "type": "string"
        },
        "localizedNames": {
          "$ref": "#/$defs/LocalizedNames"
        }
      },
      "required": [
        "name",
        "id",
        "localizedNames"
      ],
      "title": "ParameterSearchItemParameter",
      "type": "object"
    }
  },
  "description": "Lightweight representation of a ParameterGroup returned from unhydrated search().",
  "properties": {
    "name": {
      "title": "Name",
      "type": "string"
    },
    "type": {
      "anyOf": [
        {
          "$ref": "#/$defs/PGType"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "albertId": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Albertid"
    },
    "description": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Description"
    },
    "parameters": {
      "items": {
        "$ref": "#/$defs/ParameterSearchItemParameter"
      },
      "title": "Parameters",
      "type": "array"
    }
  },
  "required": [
    "name"
  ],
  "title": "ParameterGroupSearchItem",
  "type": "object"
}

Fields:

• _collection (BaseCollection | None)
• name (str)
• type (PGType | None)
• id (str | None)
• description (str | None)
• parameters (list[ParameterSearchItemParameter])

description pydantic-field

description: str | None = None

id pydantic-field

id: str | None = None

name pydantic-field

name: str

parameters pydantic-field

parameters: list[ParameterSearchItemParameter]

type pydantic-field

type: PGType | None = None

generate_parameter_group_patches

generate_parameter_group_patches(
    initial_patches: PatchPayload,
    updated_parameter_group: ParameterGroup,
    existing_parameter_group: ParameterGroup,
)

Source code in src/albert/utils/_patch.py

def generate_parameter_group_patches(
    initial_patches: PatchPayload,
    updated_parameter_group: ParameterGroup,
    existing_parameter_group: ParameterGroup,
):
    # convert to PGPatchPayload to be able to add PGPatchDatum
    general_patches = PGPatchPayload(data=initial_patches.data)
    parameter_patches, new_parameters, parameter_enum_patches = generate_parameter_patches(
        initial_parameters=existing_parameter_group.parameters,
        updated_parameters=updated_parameter_group.parameters,
        parameter_attribute_name="parameter",
    )
    tag_patches = handle_tags(
        existing_tags=existing_parameter_group.tags,
        updated_tags=updated_parameter_group.tags,
        attribute_name="tagId",
    )
    # add to the general patches
    general_patches.data.extend(parameter_patches)
    general_patches.data.extend(tag_patches)

    return general_patches, new_parameters, parameter_enum_patches