Skip to content

Breakthrough Insights

albert.collections.btinsight

BTInsightId module-attribute 

BTInsightId = Annotated[
    str, AfterValidator(ensure_btinsight_id)
]

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 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
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: _Ffor (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 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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 
75
76
77
78
79
80
81
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

BTInsight

Bases: BaseResource

category instance-attribute 

category: BTInsightCategory

content_edited class-attribute instance-attribute 

content_edited: bool | None = Field(
    default=None, alias="contentEdited"
)

dataset_id class-attribute instance-attribute 

dataset_id: BTDatasetId | None = Field(
    default=None, alias="datasetId"
)

end_time class-attribute instance-attribute 

end_time: str | None = Field(default=None, alias='endTime')

id class-attribute instance-attribute 

id: BTInsightId | None = Field(
    default=None, alias="albertId"
)

metadata class-attribute instance-attribute 

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

model_id class-attribute instance-attribute 

model_id: BTModelId | None = Field(
    default=None, alias="modelId"
)

model_session_id class-attribute instance-attribute 

model_session_id: BTModelSessionId | None = Field(
    default=None, alias="modelSessionId"
)

name instance-attribute 

name: str

output_key class-attribute instance-attribute 

output_key: str | None = Field(
    default=None, alias="outputKey"
)

payload_type class-attribute instance-attribute 

payload_type: BTInsightPayloadType | None = Field(
    default=None, alias="payloadType"
)

raw_payload class-attribute instance-attribute 

raw_payload: dict[str, Any] | None = Field(
    default=None, alias="RawPayload"
)

registry class-attribute instance-attribute 

registry: BTInsightRegistry | None = Field(
    default=None, alias="Registry"
)

start_time class-attribute instance-attribute 

start_time: str | None = Field(
    default=None, alias="startTime"
)

state class-attribute instance-attribute 

state: BTInsightState | None = Field(default=None)

total_time class-attribute instance-attribute 

total_time: str | None = Field(
    default=None, alias="totalTime"
)

BTInsightCategory

Bases: str, Enum

CUSTOM_OPTIMIZER class-attribute instance-attribute 

CUSTOM_OPTIMIZER = 'Custom Optimizer'

GENERATE class-attribute instance-attribute 

GENERATE = 'Generate'

IMPACT_CHART class-attribute instance-attribute 

IMPACT_CHART = 'Impact Chart'

MOLECULE class-attribute instance-attribute 

MOLECULE = 'Molecule'

OPTIMIZER class-attribute instance-attribute 

OPTIMIZER = 'Optimizer'

SMART_DOE class-attribute instance-attribute 

SMART_DOE = 'Smart DOE'

BTInsightCollection 

BTInsightCollection(*, session: AlbertSession)

Bases: BaseCollection

BTInsightCollection is a collection class for managing Breakthrough insight entities.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

 required

Attributes:

Name Type Description
base_path str

The base path for BTInsight API requests.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

 required

Methods:

Name Description
create

Create a new BTInsight.
delete

Delete a BTInsight by ID.
get_by_id

Get a BTInsight by ID.
search

Search for items in the BTInsight collection.
update

Update a BTInsight.
Source code in src/albert/collections/btinsight.py 
43
44
45
46
47
48
49
50
51
52
53
def __init__(self, *, session: AlbertSession):
    """
    Initialize the BTInsightCollection with the provided session.

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

base_path instance-attribute 

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

create 

create(*, insight: BTInsight) -> BTInsight

Create a new BTInsight.

Parameters:

Name Type Description Default
insight BTInsight

The BTInsight record to create.

 required

Returns:

Type Description
BTInsight

The created BTInsight.
Source code in src/albert/collections/btinsight.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@validate_call
def create(self, *, insight: BTInsight) -> BTInsight:
    """
    Create a new BTInsight.

    Parameters
    ----------
    insight : BTInsight
        The BTInsight record to create.

    Returns
    -------
    BTInsight
        The created BTInsight.
    """
    response = self.session.post(
        self.base_path,
        json=insight.model_dump(mode="json", by_alias=True, exclude_none=True),
    )
    return BTInsight(**response.json())

delete 

delete(*, id: BTInsightId) -> None

Delete a BTInsight by ID.

Parameters:

Name Type Description Default
id str

The ID of the BTInsight to delete.

 required

Returns:

Type Description
None
Source code in src/albert/collections/btinsight.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
@validate_call
def delete(self, *, id: BTInsightId) -> None:
    """Delete a BTInsight by ID.

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

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

get_by_id 

get_by_id(*, id: BTInsightId) -> BTInsight

Get a BTInsight by ID.

Parameters:

Name Type Description Default
id BTInsightId

The Albert ID of the insight.

 required

Returns:

Type Description
BTInsight

The retrived BTInsight.
Source code in src/albert/collections/btinsight.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@validate_call
def get_by_id(self, *, id: BTInsightId) -> BTInsight:
    """
    Get a BTInsight by ID.

    Parameters
    ----------
    id : BTInsightId
        The Albert ID of the insight.

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

search 

search(
    *,
    limit: int = 100,
    offset: int | None = None,
    order_by: OrderBy | None = None,
    sort_by: str | None = None,
    text: str | None = None,
    name: str | list[str] | None = None,
    state: BTInsightState
    | list[BTInsightState]
    | None = None,
    category: BTInsightCategory
    | list[BTInsightCategory]
    | None = None,
) -> Iterator[BTInsight]

Search for items in the BTInsight collection.

Parameters:

Name Type Description Default
limit int

Number of items to return per page, default 100

 100
offset int | None

Item offset to begin search at, default None

 None
order_by OrderBy | None

Asc/desc ordering, default None

 None
sort_by str | None

Sort field, default None

 None
text str | None

Text field in search query, default None

 None
name str | list[str] | None

BTInsight name search filter, default None

 None
state BTInsightState | list[BTInsightState] | None

BTInsight state search filter, default None

 None
category BTInsightCategory | list[BTInsightCategory] | None

BTInsight category search filter, default None

 None

Returns:

Type Description
Iterator[BTInsight]

An iterator of elements returned by the BTInsight search query.
Source code in src/albert/collections/btinsight.py 
 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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
@validate_call
def search(
    self,
    *,
    limit: int = 100,
    offset: int | None = None,
    order_by: OrderBy | None = None,
    sort_by: str | None = None,
    text: str | None = None,
    name: str | list[str] | None = None,
    state: BTInsightState | list[BTInsightState] | None = None,
    category: BTInsightCategory | list[BTInsightCategory] | None = None,
) -> Iterator[BTInsight]:
    """Search for items in the BTInsight collection.

    Parameters
    ----------
    limit : int, optional
        Number of items to return per page, default 100
    offset : int | None, optional
        Item offset to begin search at, default None
    order_by : OrderBy | None, optional
        Asc/desc ordering, default None
    sort_by : str | None
        Sort field, default None
    text : str | None
        Text field in search query, default None
    name : str | list[str] | None
        BTInsight name search filter, default None
    state : BTInsightState | list[BTInsightState] | None
        BTInsight state search filter, default None
    category : BTInsightCategory | list[BTInsightCategory] | None
        BTInsight category search filter, default None

    Returns
    -------
    Iterator[BTInsight]
        An iterator of elements returned by the BTInsight search query.
    """
    params = {
        "limit": limit,
        "offset": offset,
        "order": OrderBy(order_by).value if order_by else None,
        "sortBy": sort_by,
        "text": text,
        "name": name,
    }
    if state:
        state = state if isinstance(state, list) else [state]
        params["state"] = [BTInsightState(x).value for x in state]
    if category:
        category = category if isinstance(category, list) else [category]
        params["category"] = [BTInsightCategory(x).value for x in category]

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        params=params,
        deserialize=lambda items: [BTInsight(**item) for item in items],
    )

update 

update(*, insight: BTInsight) -> BTInsight

Update a BTInsight.

Parameters:

Name Type Description Default
insight BTInsight

The BTInsight to update.

 required

Returns:

Type Description
BTInsight

The updated BTInsight.
Source code in src/albert/collections/btinsight.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@validate_call
def update(self, *, insight: BTInsight) -> BTInsight:
    """Update a BTInsight.

    Parameters
    ----------
    insight : BTInsight
        The BTInsight to update.

    Returns
    -------
    BTInsight
        The updated BTInsight.
    """
    path = f"{self.base_path}/{insight.id}"
    payload = self._generate_patch_payload(
        existing=self.get_by_id(id=insight.id),
        updated=insight,
        generate_metadata_diff=False,
    )
    self.session.patch(path, json=payload.model_dump(mode="json", by_alias=True))
    return self.get_by_id(id=insight.id)

BTInsightState

Bases: str, Enum

BUILDING_MODELS class-attribute instance-attribute 

BUILDING_MODELS = 'Building Models'

COMPLETE class-attribute instance-attribute 

COMPLETE = 'Complete'

ERROR class-attribute instance-attribute 

ERROR = 'Error'

GENERATING_CANDIDATES class-attribute instance-attribute 

GENERATING_CANDIDATES = 'Generating Candidates'

QUEUED class-attribute instance-attribute 

QUEUED = 'Queued'

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 
21
22
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'