Skip to content

Data Columns

albert.collections.data_columns.DataColumnCollection 

DataColumnCollection(*, session: AlbertSession)

Bases: BaseCollection

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

Methods:

Name Description
get_by_name

Get a data column by its name.
get_by_id

Get a data column by its ID.
list

Lists data column entities with optional filters.
create

Create a new data column entity.
delete

Delete a data column entity.
update

Update a data column entity.

Attributes:

Name Type Description
base_path
Source code in src/albert/collections/data_columns.py 
18
19
20
21
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 

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

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 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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.list(name=name):
        if dc.name.lower() == name.lower():
            return dc
    return None

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 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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

list 

list(
    *,
    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,
    return_full: bool = True,
) -> Iterator[DataColumn]

Lists 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
return_full bool

Whether to make additional API call to fetch the full object, by default True

 True

Returns:

Type Description
Iterator[DataColumn]

An iterator of DataColumns matching the provided criteria.
Source code in src/albert/collections/data_columns.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def list(
    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,
    return_full: bool = True,
) -> Iterator[DataColumn]:
    """
    Lists 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.
    return_full : bool, optional
        Whether to make additional API call to fetch the full object, by default True

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

    def deserialize(items: list[dict]) -> Iterator[DataColumn]:
        if return_full:
            for item in items:
                id = item["albertId"]
                try:
                    yield self.get_by_id(id=id)
                except AlbertHTTPError as e:
                    logger.warning(f"Error fetching Data Column '{id}': {e}")
        else:
            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,
    )

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 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
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 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
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}")

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 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
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)