Tags

albert.collections.tags

logger module-attribute

logger = create_logger()

AlbertException

AlbertException(message: str)

Bases: Exception

Source code in src/albert/exceptions.py

def __init__(self, message: str):
    super().__init__(message)
    self.message = message

message instance-attribute

message = message

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

Tag

Bases: BaseResource

Tag is a Pydantic model representing a tag entity.

Attributes:

Name	Type	Description
tag	str	The name of the tag.
id	str | None	The Albert ID of the tag. Set when the tag is retrieved from Albert.

Methods:

Name	Description
from_string	Creates a Tag object from a string.

id class-attribute instance-attribute

id: str | None = Field(
    None,
    alias=AliasChoices("albertId", "tagId"),
    serialization_alias="albertId",
)

tag class-attribute instance-attribute

tag: str = Field(
    alias=AliasChoices("name", "tagName"),
    serialization_alias="name",
)

from_string classmethod

from_string(tag: str) -> Tag

Creates a Tag object from a string.

Parameters:

Name	Type	Description	Default
tag	str	The name of the tag.	required

Returns:

Type	Description
Tag	The Tag object created from the string.

Source code in src/albert/resources/tags.py

@classmethod
def from_string(cls, tag: str) -> "Tag":
    """
    Creates a Tag object from a string.

    Parameters
    ----------
    tag : str
        The name of the tag.

    Returns
    -------
    Tag
        The Tag object created from the string.
    """
    return cls(tag=tag)

TagCollection

TagCollection(*, session: AlbertSession)

Bases: BaseCollection

TagCollection is a collection class for managing Tag entities in the Albert platform.

Parameters:

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

Attributes:

Name	Type	Description
base_path	str	The base URL for tag API requests.

Methods:

Name	Description
get_all	Lists tag entities with optional filters.
exists	Checks if a tag exists by its name.
create	Creates a new tag entity.
get_by_id	Retrieves a tag by its ID.
get_by_ids	Retrieve a list of tags by their IDs.
get_by_name	Retrieves a tag by its name.
delete	Deletes a tag by its ID.
rename	Renames an existing tag entity.

Parameters:

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

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

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

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

base_path instance-attribute

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

create

create(*, tag: str | Tag) -> Tag

Creates a new tag entity if the given tag does not exist.

Parameters:

Name	Type	Description	Default
tag	Union[str, Tag]	The tag name or Tag object to create.	required

Returns:

Type	Description
Tag	The created Tag object or the existing Tag object if it already exists.

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

def create(self, *, tag: str | Tag) -> Tag:
    """
    Creates a new tag entity if the given tag does not exist.

    Parameters
    ----------
    tag : Union[str, Tag]
        The tag name or Tag object to create.

    Returns
    -------
    Tag
        The created Tag object or the existing Tag object if it already exists.
    """
    if isinstance(tag, str):
        tag = Tag(tag=tag)
    hit = self.get_by_name(name=tag.tag, exact_match=True)
    if hit is not None:
        logging.warning(f"Tag {hit.tag} already exists with id {hit.id}")
        return hit
    payload = {"name": tag.tag}
    response = self.session.post(self.base_path, json=payload)
    tag = Tag(**response.json())
    return tag

delete

delete(*, id: str) -> None

Deletes a tag by its ID.

Parameters:

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

Returns:

Type	Description
None

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

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

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

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

exists

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

Checks if a tag exists by its name.

Parameters:

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

Returns:

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

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

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

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

    Returns
    -------
    bool
        True if the tag exists, False otherwise.
    """

    return self.get_by_name(name=tag, exact_match=exact_match) is not None

get_all

get_all(
    *,
    limit: int = 50,
    order_by: OrderBy = DESCENDING,
    name: str | list[str] | None = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Tag]

Get all Tag entities with optional filters.

Parameters:

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

Returns:

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

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

def get_all(
    self,
    *,
    limit: int = 50,
    order_by: OrderBy = OrderBy.DESCENDING,
    name: str | list[str] | None = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Tag]:
    """
    Get all Tag entities with optional filters.

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

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

get_by_id

get_by_id(*, id: str) -> Tag

Get a tag by its ID.

Parameters:

Name	Type	Description	Default
id	str	The ID of the tag to get.	required

Returns:

Type	Description
Tag	The Tag object.

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

def get_by_id(self, *, id: str) -> Tag:
    """
    Get a tag by its ID.

    Parameters
    ----------
    id : str
        The ID of the tag to get.

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

get_by_ids

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

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

def get_by_ids(self, *, ids: list[str]) -> list[Tag]:
    url = f"{self.base_path}/ids"
    batches = [ids[i : i + 100] for i in range(0, len(ids), 100)]
    return [
        Tag(**item)
        for batch in batches
        for item in self.session.get(url, params={"id": batch}).json()
    ]

get_by_name

get_by_name(
    *, name: str, exact_match: bool = True
) -> Tag | None

Retrieves a tag by its name or None if not found.

Parameters:

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

Returns:

Type	Description
Tag	The Tag object if found, None otherwise.

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

def get_by_name(self, *, name: str, exact_match: bool = True) -> Tag | None:
    """
    Retrieves a tag by its name or None if not found.

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

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

rename

rename(*, old_name: str, new_name: str) -> Tag

Renames an existing tag entity.

Parameters:

Name	Type	Description	Default
old_name	str	The current name of the tag.	required
new_name	str	The new name of the tag.	required

Returns:

Type	Description
Tag	The renamed Tag.

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

def rename(self, *, old_name: str, new_name: str) -> Tag:
    """
    Renames an existing tag entity.

    Parameters
    ----------
    old_name : str
        The current name of the tag.
    new_name : str
        The new name of the tag.

    Returns
    -------
    Tag
        The renamed Tag.
    """
    found_tag = self.get_by_name(name=old_name, exact_match=True)
    if not found_tag:
        msg = f'Tag "{old_name}" not found.'
        logger.error(msg)
        raise AlbertException(msg)
    tag_id = found_tag.id
    payload = [
        {
            "data": [
                {
                    "operation": "update",
                    "attribute": "name",
                    "oldValue": old_name,
                    "newValue": new_name,
                }
            ],
            "id": tag_id,
        }
    ]
    self.session.patch(self.base_path, json=payload)
    return self.get_by_id(id=tag_id)