Breakthrough Datasets

albert.collections.btdataset

BTDatasetId module-attribute

BTDatasetId = Annotated[
    str, AfterValidator(ensure_btdataset_id)
]

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

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