Tags

albert.collections.tags

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

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'

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
list	Lists tag entities with optional filters.
tag_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_tag	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 of 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 of it already exists.
    """
    if isinstance(tag, str):
        tag = Tag(tag=tag)
    hit = self.get_by_tag(tag=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)

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_tag

get_by_tag(
    *, tag: str, exact_match: bool = True
) -> Tag | None

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

Parameters:

Name	Type	Description	Default
tag	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_tag(self, *, tag: str, exact_match: bool = True) -> Tag | None:
    """
    Retrieves a tag by its name of None if not found.

    Parameters
    ----------
    tag : 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.list(name=tag, exact_match=exact_match)
    return next(found, None)

list

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

Lists 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 list(
    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]:
    """
    Lists 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],
    )

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_tag(tag=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)

tag_exists

tag_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 tag_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_tag(tag=tag, exact_match=exact_match) is not None