Breakthrough Insights

albert.collections.btinsight

BTInsightId

BTInsightId = Annotated[
    str, AfterValidator(ensure_btinsight_id)
]

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

session = session

OrderBy

Bases: str, Enum

DESCENDING

DESCENDING = 'desc'

ASCENDING

ASCENDING = 'asc'

BTInsight

Bases: BaseResource

name

name: str

category

category: BTInsightCategory

metadata

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

state

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

id

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

dataset_id

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

model_session_id

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

model_id

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

output_key

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

start_time

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

end_time

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

total_time

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

raw_payload

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

payload_type

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

registry

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

content_edited

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

BTInsightCategory

Bases: str, Enum

OPTIMIZER

OPTIMIZER = 'Optimizer'

CUSTOM_OPTIMIZER

CUSTOM_OPTIMIZER = 'Custom Optimizer'

IMPACT_CHART

IMPACT_CHART = 'Impact Chart'

MOLECULE

MOLECULE = 'Molecule'

SMART_DOE

SMART_DOE = 'Smart DOE'

GENERATE

GENERATE = 'Generate'

BTInsightState

Bases: str, Enum

QUEUED

QUEUED = 'Queued'

BUILDING_MODELS

BUILDING_MODELS = 'Building Models'

GENERATING_CANDIDATES

GENERATING_CANDIDATES = 'Generating Candidates'

COMPLETE

COMPLETE = 'Complete'

ERROR

ERROR = 'Error'

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

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

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

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

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

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

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

@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())

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

@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

@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

@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)

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

@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}")