Breakthrough Datasets

albert.collections.btdataset

BTDatasetId module-attribute

BTDatasetId = Annotated[
    str, AfterValidator(ensure_btdataset_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

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

BTDataset

Bases: BaseResource

file_name class-attribute instance-attribute

file_name: str | None = Field(
    default=None, alias="fileName"
)

id class-attribute instance-attribute

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

key class-attribute instance-attribute

key: str | None = Field(default=None)

name instance-attribute

name: str

references class-attribute instance-attribute

references: BTDatasetReferences | None = Field(
    default=None, alias="References"
)

report class-attribute instance-attribute

report: EntityLink | None = Field(
    default=None, alias="Report"
)

BTDatasetCollection

BTDatasetCollection(*, session: AlbertSession)

Bases: BaseCollection

BTDatasetCollection is a collection class for managing Breakthrough dataset entities.

Parameters:

Name	Type	Description	Default
session	AlbertSession	The Albert session instance.	required

Attributes:

Name	Type	Description
base_path	str	The base path for btdataset API requests.

Parameters:

Name	Type	Description	Default
session	AlbertSession	The Albert session instance.	required

Methods:

Name	Description
create	Create a new BTDataset.
delete	Delete a BTDataset by ID.
get_all	Get all items from the BTDataset collection.
get_by_id	Get a BTDataset by ID.
update	Update a BTDataset.

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

def __init__(self, *, session: AlbertSession):
    """
    Initialize the BTDatasetCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, dataset: BTDataset) -> BTDataset

Create a new BTDataset.

Parameters:

Name	Type	Description	Default
dataset	BTDataset	The BTDataset record to create.	required

Returns:

Type	Description
BTDataset	The created BTDataset.

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

@validate_call
def create(self, *, dataset: BTDataset) -> BTDataset:
    """
    Create a new BTDataset.

    Parameters
    ----------
    dataset : BTDataset
        The BTDataset record to create.

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

delete

delete(*, id: BTDatasetId) -> None

Delete a BTDataset by ID.

Parameters:

Name	Type	Description	Default
id	BTDatasetId	The ID of the BTDataset to delete.	required

Returns:

Type	Description
None

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

@validate_call
def delete(self, *, id: BTDatasetId) -> None:
    """Delete a BTDataset by ID.

    Parameters
    ----------
    id : BTDatasetId
        The ID of the BTDataset to delete.

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

get_all

get_all(
    *,
    limit: int = 100,
    name: str | None = None,
    start_key: str | None = None,
    created_by: str | None = None,
) -> Iterator[BTDataset]

Get all items from the BTDataset collection.

Parameters:

Name	Type	Description	Default
limit	int	Number of items to return per page, default 100	100
name	str	Name of the dataset to filter by, default None	None
start_key	str	The starting key for pagination, default None	None
created_by	str	The user who created the dataset, default None	None

Returns:

Type	Description
Iterator[BTDataset]	An iterator of elements returned by the BTDataset listing.

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

@validate_call
def get_all(
    self,
    *,
    limit: int = 100,
    name: str | None = None,
    start_key: str | None = None,
    created_by: str | None = None,
) -> Iterator[BTDataset]:
    """Get all items from the BTDataset collection.

    Parameters
    ----------
    limit : int, optional
        Number of items to return per page, default 100
    name : str, optional
        Name of the dataset to filter by, default None
    start_key : str, optional
        The starting key for pagination, default None
    created_by : str, optional
        The user who created the dataset, default None

    Returns
    -------
    Iterator[BTDataset]
        An iterator of elements returned by the BTDataset listing.
    """
    params = {
        "limit": limit,
        "startKey": start_key,
        "createdBy": created_by,
        "name": name,
    }
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [BTDataset(**item) for item in items],
    )

get_by_id

get_by_id(*, id: BTDatasetId) -> BTDataset

Get a BTDataset by ID.

Parameters:

Name	Type	Description	Default
id	BTDatasetId	The Albert ID of the BTDataset.	required

Returns:

Type	Description
BTDataset	The retrived BTDataset.

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

@validate_call
def get_by_id(self, *, id: BTDatasetId) -> BTDataset:
    """
    Get a BTDataset by ID.

    Parameters
    ----------
    id : BTDatasetId
        The Albert ID of the BTDataset.

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

update

update(*, dataset: BTDataset) -> BTDataset

Update a BTDataset.

The provided dataset must be registered with an Albert ID.

Parameters:

Name	Type	Description	Default
dataset	BTDataset	The BTDataset with updated fields.	required

Returns:

Type	Description
BTDataset	The updated BTDataset object.

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

@validate_call
def update(self, *, dataset: BTDataset) -> BTDataset:
    """
    Update a BTDataset.

    The provided dataset must be registered with an Albert ID.

    Parameters
    ----------
    dataset : BTDataset
        The BTDataset with updated fields.

    Returns
    -------
    BTDataset
        The updated BTDataset object.
    """
    path = f"{self.base_path}/{dataset.id}"
    payload = self._generate_patch_payload(
        existing=self.get_by_id(id=dataset.id),
        updated=dataset,
    )
    self.session.patch(path, json=payload.model_dump(mode="json", by_alias=True))
    return self.get_by_id(id=dataset.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 instance-attribute

session = session

PaginationMode

Bases: str, Enum

KEY class-attribute instance-attribute

KEY = 'key'

OFFSET class-attribute instance-attribute

OFFSET = 'offset'