Data Columns

albert.collections.data_columns

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: _3for (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

DataColumn

Bases: BaseResource

defalt class-attribute instance-attribute

defalt: bool = False

id class-attribute instance-attribute

id: str = Field(default=None, alias='albertId')

metadata class-attribute instance-attribute

metadata: dict[str, MetadataItem] | None = Field(
    alias="Metadata", default=None
)

name instance-attribute

name: str

DataColumnCollection

DataColumnCollection(*, session: AlbertSession)

Bases: BaseCollection

DataColumnCollection is a collection class for managing DataColumn entities in the Albert platform.

Methods:

Name	Description
create	Create a new data column entity.
delete	Delete a data column entity.
get_all	Get all data column entities with optional filters.
get_by_id	Get a data column by its ID.
get_by_name	Get a data column by its name.
update	Update a data column entity.

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

def __init__(self, *, session: AlbertSession):
    """Initialize the DataColumnCollection with the provided session."""
    super().__init__(session=session)
    self.base_path = f"/api/{DataColumnCollection._api_version}/datacolumns"

base_path instance-attribute

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

create

create(*, data_column: DataColumn) -> DataColumn

Create a new data column entity.

Parameters:

Name	Type	Description	Default
data_column	DataColumn	The data column object to create.	required

Returns:

Type	Description
DataColumn	The created data column object.

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

def create(self, *, data_column: DataColumn) -> DataColumn:
    """
    Create a new data column entity.

    Parameters
    ----------
    data_column : DataColumn
        The data column object to create.

    Returns
    -------
    DataColumn
        The created data column object.
    """
    payload = [data_column.model_dump(by_alias=True, exclude_unset=True, mode="json")]
    response = self.session.post(self.base_path, json=payload)

    return DataColumn(**response.json()[0])

delete

delete(*, id: str) -> None

Delete a data column entity.

Parameters:

Name	Type	Description	Default
id	str	The ID of the data column object to delete.	required

Returns:

Type	Description
None

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

def delete(self, *, id: str) -> None:
    """
    Delete a data column entity.

    Parameters
    ----------
    id : str
        The ID of the data column object to delete.

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

get_all

get_all(
    *,
    order_by: OrderBy = DESCENDING,
    ids: str | list[str] | None = None,
    name: str | list[str] | None = None,
    exact_match: bool | None = None,
    default: bool | None = None,
    start_key: str | None = None,
    limit: int = 100,
) -> Iterator[DataColumn]

Get all data column entities with optional filters.

Parameters:

Name	Type	Description	Default
order_by	OrderBy	The order by which to sort the results, by default OrderBy.DESCENDING.	DESCENDING
ids	str | list[str] | None	Data column IDs to filter the search by, default None.	None
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.	None
default	bool	Whether to return only default columns, by default None.	None

Returns:

Type	Description
Iterator[DataColumn]	An iterator of DataColumns matching the provided criteria.

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

def get_all(
    self,
    *,
    order_by: OrderBy = OrderBy.DESCENDING,
    ids: str | list[str] | None = None,
    name: str | list[str] | None = None,
    exact_match: bool | None = None,
    default: bool | None = None,
    start_key: str | None = None,
    limit: int = 100,
) -> Iterator[DataColumn]:
    """
    Get all data column entities with optional filters.

    Parameters
    ----------
    order_by : OrderBy, optional
        The order by which to sort the results, by default OrderBy.DESCENDING.
    ids: str | list[str] | None, optional
        Data column IDs to filter the search by, default None.
    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.
    default : bool, optional
        Whether to return only default columns, by default None.

    Returns
    -------
    Iterator[DataColumn]
        An iterator of DataColumns matching the provided criteria.
    """

    def deserialize(items: list[dict]) -> Iterator[DataColumn]:
        yield from (DataColumn(**item) for item in items)

    params = {
        "limit": limit,
        "orderBy": order_by.value,
        "startKey": start_key,
        "name": [name] if isinstance(name, str) else name,
        "exactMatch": json.dumps(exact_match) if exact_match is not None else None,
        "default": json.dumps(default) if default is not None else None,
        "dataColumns": [ids] if isinstance(ids, str) else ids,
    }
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=deserialize,
    )

get_by_id

get_by_id(*, id) -> DataColumn

Get a data column by its ID.

Parameters:

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

Returns:

Type	Description
DataColumn | None	The data column object on match or None

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

def get_by_id(self, *, id) -> DataColumn:
    """
    Get a data column by its ID.

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

    Returns
    -------
    DataColumn | None
        The data column object on match or None
    """
    response = self.session.get(f"{self.base_path}/{id}")
    dc = DataColumn(**response.json())
    return dc

get_by_name

get_by_name(*, name) -> DataColumn | None

Get a data column by its name.

Parameters:

Name	Type	Description	Default
name	str	The name of the data column to get.	required

Returns:

Type	Description
DataColumn | None	The data column object on match or None

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

def get_by_name(self, *, name) -> DataColumn | None:
    """
    Get a data column by its name.

    Parameters
    ----------
    name : str
        The name of the data column to get.

    Returns
    -------
    DataColumn | None
        The data column object on match or None
    """
    for dc in self.get_all(name=name):
        if dc.name.lower() == name.lower():
            return dc
    return None

update

update(*, data_column: DataColumn) -> DataColumn

Update a data column entity.

Parameters:

Name	Type	Description	Default
data_column	DataColumn	The updated data column object. The ID must be set and match an existing data column.	required

Returns:

Type	Description
DataColumn	The updated data column object as registered in Albert.

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

def update(self, *, data_column: DataColumn) -> DataColumn:
    """Update a data column entity.

    Parameters
    ----------
    data_column : DataColumn
        The updated data column object. The ID must be set and match an existing data column.

    Returns
    -------
    DataColumn
        The updated data column object as registered in Albert.
    """
    existing = self.get_by_id(id=data_column.id)
    payload = self._generate_patch_payload(
        existing=existing,
        updated=data_column,
    )
    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=data_column,
            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 data_column
    for e in payload_dump["data"]:
        self.session.patch(
            f"{self.base_path}/{data_column.id}",
            json={"data": [e]},
        )
    return self.get_by_id(id=data_column.id)

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'