Data Templates

DataTemplateId module-attribute

DataTemplateId = Annotated[
    str, AfterValidator(ensure_datatemplate_id)
]

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

DataColumnValue

Bases: BaseAlbertModel

Methods:

Name	Description
check_for_id

calculation class-attribute instance-attribute

calculation: str | None = None

column_sequence class-attribute instance-attribute

column_sequence: str | None = Field(
    default=None, alias="sequence", exclude=True
)

data_column class-attribute instance-attribute

data_column: DataColumn = Field(exclude=True, default=None)

data_column_id class-attribute instance-attribute

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

hidden class-attribute instance-attribute

hidden: bool = False

unit class-attribute instance-attribute

unit: SerializeAsEntityLink[Unit] | None = Field(
    default=None, alias="Unit"
)

value class-attribute instance-attribute

value: str | None = None

check_for_id

check_for_id()

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

@model_validator(mode="after")
def check_for_id(self):
    if self.data_column_id is None and self.data_column is None:
        raise ValueError("Either data_column_id or data_column must be set")
    elif (
        self.data_column_id is not None
        and self.data_column is not None
        and self.data_column.id != self.data_column_id
    ):
        raise ValueError("If both are provided, data_column_id and data_column.id must match")
    elif self.data_column_id is None:
        self.data_column_id = self.data_column.id
    return self

DataTemplate

Bases: BaseTaggedEntity

data_column_values class-attribute instance-attribute

data_column_values: list[DataColumnValue] | None = Field(
    alias="DataColumns", default=None
)

description class-attribute instance-attribute

description: str | None = None

id class-attribute instance-attribute

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

metadata class-attribute instance-attribute

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

name instance-attribute

name: str

security_class class-attribute instance-attribute

security_class: SecurityClass | None = None

users_with_access class-attribute instance-attribute

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

verified class-attribute instance-attribute

verified: bool = False

DataTemplateCollection

DataTemplateCollection(*, session: AlbertSession)

Bases: BaseCollection

DataTemplateCollection is a collection class for managing DataTemplate entities in the Albert platform.

Methods:

Name	Description
add_data_columns	Adds data columns to a data template.
create	Creates a new data template.
delete	Deletes a data template by its ID.
get_by_id	Get a data template by its ID.
get_by_ids	Get a list of data templates by their IDs.
get_by_name	Get a data template by its name.
list	Lists data template entities with optional filters.
update	Updates a data template.

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

def __init__(self, *, session: AlbertSession):
    super().__init__(session=session)
    self.base_path = f"/api/{DataTemplateCollection._api_version}/datatemplates"

base_path instance-attribute

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

add_data_columns

add_data_columns(
    *,
    data_template_id: str,
    data_columns: list[DataColumnValue],
) -> DataTemplate

Adds data columns to a data template.

Parameters:

Name	Type	Description	Default
data_template_id	str	The ID of the data template to add the columns to.	required
data_columns	list[DataColumnValue]	The list of DataColumnValue objects to add to the data template.	required

Returns:

Type	Description
DataTemplate	The updated DataTemplate object.

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

def add_data_columns(
    self, *, data_template_id: str, data_columns: list[DataColumnValue]
) -> DataTemplate:
    """Adds data columns to a data template.

    Parameters
    ----------
    data_template_id : str
        The ID of the data template to add the columns to.
    data_columns : list[DataColumnValue]
        The list of DataColumnValue objects to add to the data template.

    Returns
    -------
    DataTemplate
        The updated DataTemplate object.
    """
    payload = {
        "DataColumns": [
            x.model_dump(mode="json", by_alias=True, exclude_none=True) for x in data_columns
        ]
    }
    self.session.put(
        f"{self.base_path}/{data_template_id}/datacolumns",
        json=payload,
    )
    return self.get_by_id(id=data_template_id)

create

create(*, data_template: DataTemplate) -> DataTemplate

Creates a new data template.

Parameters:

Name	Type	Description	Default
data_template	DataTemplate	The DataTemplate object to create.	required

Returns:

Type	Description
DataTemplate	The registered DataTemplate object with an ID.

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

def create(self, *, data_template: DataTemplate) -> DataTemplate:
    """Creates a new data template.

    Parameters
    ----------
    data_template : DataTemplate
        The DataTemplate object to create.

    Returns
    -------
    DataTemplate
        The registered DataTemplate object with an ID.
    """
    response = self.session.post(
        self.base_path,
        json=data_template.model_dump(mode="json", by_alias=True, exclude_none=True),
    )
    return DataTemplate(**response.json())

delete

delete(*, id: str) -> None

Deletes a data template by its ID.

Parameters:

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

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

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

    Parameters
    ----------
    id : str
        The ID of the data template to delete.
    """
    self.session.delete(f"{self.base_path}/{id}")

get_by_id

get_by_id(*, id: DataTemplateId) -> DataTemplate

Get a data template by its ID.

Parameters:

Name	Type	Description	Default
id	DataTemplateId	The ID of the data template to get.	required

Returns:

Type	Description
DataTemplate	The data template object on match or None

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

def get_by_id(self, *, id: DataTemplateId) -> DataTemplate:
    """Get a data template by its ID.

    Parameters
    ----------
    id : DataTemplateId
        The ID of the data template to get.

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

get_by_ids

get_by_ids(
    *, ids: list[DataTemplateId]
) -> list[DataTemplate]

Get a list of data templates by their IDs.

Parameters:

Name	Type	Description	Default
ids	list[DataTemplateId]	The list of DataTemplate IDs to get.	required

Returns:

Type	Description
list[DataTemplate]	A list of DataTemplate objects with the provided IDs.

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

def get_by_ids(self, *, ids: list[DataTemplateId]) -> list[DataTemplate]:
    """Get a list of data templates by their IDs.

    Parameters
    ----------
    ids : list[DataTemplateId]
        The list of DataTemplate IDs to get.

    Returns
    -------
    list[DataTemplate]
        A list of DataTemplate objects with the provided IDs.
    """
    url = f"{self.base_path}/ids"
    batches = [ids[i : i + 250] for i in range(0, len(ids), 250)]
    return [
        DataTemplate(**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) -> DataTemplate | None

Get a data template by its name.

Parameters:

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

Returns:

Type	Description
DataTemplate | None	The matching data template object or None if not found.

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

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

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

    Returns
    -------
    DataTemplate | None
        The matching data template object or None if not found.
    """
    hits = list(self.list(name=name))
    for h in hits:
        if h.name.lower() == name.lower():
            return h
    return None

list

list(
    *,
    name: str | None = None,
    user_id: str | None = None,
    order_by: OrderBy = DESCENDING,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[DataTemplate]

Lists data template entities with optional filters.

Parameters:

Name	Type	Description	Default
name	Union[str, None]	The name of the data template to filter by, by default None.	None
user_id	str	user_id to filter by, by default None.	None
order_by	OrderBy	The order by which to sort the results, by default OrderBy.DESCENDING.	DESCENDING

Returns:

Type	Description
Iterator[DataTemplate]	An iterator of DataTemplate objects matching the provided criteria.

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

def list(
    self,
    *,
    name: str | None = None,
    user_id: str | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[DataTemplate]:
    """
    Lists data template entities with optional filters.

    Parameters
    ----------
    name : Union[str, None], optional
        The name of the data template to filter by, by default None.
    user_id : str, optional
        user_id to filter by, by default None.
    order_by : OrderBy, optional
        The order by which to sort the results, by default OrderBy.DESCENDING.

    Returns
    -------
    Iterator[DataTemplate]
        An iterator of DataTemplate objects matching the provided criteria.
    """

    def deserialize(items: list[dict]) -> Iterator[DataTemplate]:
        for item in items:
            id = item["albertId"]
            try:
                yield self.get_by_id(id=id)
            except AlbertHTTPError as e:
                logger.warning(f"Error fetching parameter group {id}: {e}")
        # get by ids is not currently returning metadata correctly, so temp fixing this
        # return self.get_by_ids(ids=[x["albertId"] for x in items])

    params = {
        "limit": limit,
        "offset": offset,
        "order": OrderBy(order_by).value if order_by else None,
        "text": name,
        "userId": user_id,
    }

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

update

update(*, data_template: DataTemplate) -> DataTemplate

Updates a data template.

Parameters:

Name	Type	Description	Default
data_template	DataTemplate	The DataTemplate object to update. The ID must be set and matching the ID of the DataTemplate to update.	required

Returns:

Type	Description
DataTemplate	The Updated DataTemplate object.

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

def update(self, *, data_template: DataTemplate) -> DataTemplate:
    """Updates a data template.

    Parameters
    ----------
    data_template : DataTemplate
        The DataTemplate object to update. The ID must be set and matching the ID of the DataTemplate to update.

    Returns
    -------
    DataTemplate
        The Updated DataTemplate object.
    """
    existing = self.get_by_id(id=data_template.id)
    base_payload = self._generate_patch_payload(existing=existing, updated=data_template)
    payload = base_payload.model_dump(mode="json", by_alias=True)
    _updatable_attributes_special = {"tags", "data_column_values"}
    for attribute in _updatable_attributes_special:
        old_value = getattr(existing, attribute)
        new_value = getattr(data_template, attribute)
        if attribute == "tags":
            if (old_value is None or old_value == []) and new_value is not None:
                for t in new_value:
                    payload["data"].append(
                        {
                            "operation": "add",
                            "attribute": "tagId",
                            "newValue": t.id,  # This will be a CasAmount Object,
                            "entityId": t.id,
                        }
                    )
            else:
                if old_value is None:  # pragma: no cover
                    old_value = []
                if new_value is None:  # pragma: no cover
                    new_value = []
                old_set = {obj.id for obj in old_value}
                new_set = {obj.id for obj in new_value}

                # Find what's in set 1 but not in set 2
                to_del = old_set - new_set

                # Find what's in set 2 but not in set 1
                to_add = new_set - old_set

                for id in to_add:
                    payload["data"].append(
                        {
                            "operation": "add",
                            "attribute": "tagId",
                            "newValue": id,
                        }
                    )
                for id in to_del:
                    payload["data"].append(
                        {
                            "operation": "delete",
                            "attribute": "tagId",
                            "oldValue": id,
                        }
                    )
        elif attribute == "data_column_values":
            # Do the update by column
            to_remove = set([x.data_column_id for x in old_value]) - set(
                [x.data_column_id for x in new_value]
            )
            to_add = set([x.data_column_id for x in new_value]) - set(
                [x.data_column_id for x in old_value]
            )
            to_update = set([x.data_column_id for x in new_value]) & set(
                [x.data_column_id for x in old_value]
            )
            if len(to_remove) > 0:
                logger.error(
                    "Data Columns cannot be Removed from a Data Template. Set to hidden instead and retry."
                )
            if len(to_add) > 0:
                new_dcs = [x for x in new_value if x.data_column_id in to_add]
                self.add_data_columns(data_template_id=data_template.id, data_columns=new_dcs)
            for dc_id in to_update:
                actions = []
                old_dc_val = next(x for x in old_value if x.data_column_id == dc_id)
                new_dc_val = next(x for x in new_value if x.data_column_id == dc_id)
                # do hidden last because it can change the column sequence.
                if old_dc_val.unit != new_dc_val.unit:
                    payload["data"].append(
                        {
                            "operation": "update",
                            "attribute": "unit",
                            "newValue": new_dc_val.unit.id,
                            "oldValue": old_dc_val.unit.id,
                            "colId": old_dc_val.column_sequence,
                        }
                    )
                if old_dc_val.hidden != new_dc_val.hidden:
                    actions.append(
                        {
                            "operation": "update",
                            "attribute": "hidden",
                            "newValue": new_dc_val.hidden,
                            "oldValue": old_dc_val.hidden,
                        }
                    )
                if len(actions) > 0:
                    payload["data"].append(
                        {
                            "actions": actions,
                            "attribute": "datacolumn",
                            "colId": old_dc_val.column_sequence,
                        }
                    )
    if len(payload["data"]) > 0:
        url = f"{self.base_path}/{existing.id}"
        self.session.patch(url, json=payload)
    return self.get_by_id(id=existing.id)  # always do this in case columns were added

OrderBy

Bases: str, Enum

ASCENDING class-attribute instance-attribute

ASCENDING = 'asc'

DESCENDING class-attribute instance-attribute

DESCENDING = 'desc'