Skip to content

Albert Client

albert.albert

Albert

Albert(
    *,
    base_url: str | None = None,
    token: str | None = None,
    client_credentials: ClientCredentials | None = None,
    retries: int | None = None,
    session: AlbertSession | None = None,
)

Albert is the main client class for interacting with the Albert API.

Parameters:

Name Type Description Default
base_url str

The base URL of the Albert API (default is "https://app.albertinvent.com").

None
token str

The token for authentication (default is read from environment variable "ALBERT_TOKEN").

None
client_credentials ClientCredentials | None

The client credentials for programmatic authentication. Client credentials can be read from the environment by ClientCredentials.from_env().

None
retries int

The maximum number of retries for failed requests (default is None).

None
session AlbertSession

An optional preconfigured session to use for API requests. If not provided, a default session is created using the other parameters or environment variables. When supplied, base_url, token and client_credentials are ignored.

None

Attributes:

Name Type Description
session AlbertSession

The session for API requests, with a base URL set.

projects ProjectCollection

The project collection instance.

tags TagCollection

The tag collection instance.

inventory InventoryCollection

The inventory collection instance.

companies CompanyCollection

The company collection instance.

Source code in src/albert/albert.py
def __init__(
    self,
    *,
    base_url: str | None = None,
    token: str | None = None,
    client_credentials: ClientCredentials | None = None,
    retries: int | None = None,
    session: AlbertSession | None = None,
):
    self.session = session or AlbertSession(
        base_url=base_url or os.getenv("ALBERT_BASE_URL") or "https://app.albertinvent.com",
        token=token or os.getenv("ALBERT_TOKEN"),
        client_credentials=client_credentials or ClientCredentials.from_env(),
        retries=retries,
    )

attachments property

attachments: AttachmentCollection

batch_data property

batch_data: BatchDataCollection

btdatasets property

btdatasets: BTDatasetCollection

btinsights property

btinsights: BTInsightCollection

btmodels property

btmodelsessions property

btmodelsessions: BTModelSessionCollection

cas_numbers property

cas_numbers: CasCollection

companies property

companies: CompanyCollection

custom_fields property

custom_fields: CustomFieldCollection

data_columns property

data_columns: DataColumnCollection

data_templates property

data_templates: DataTemplateCollection

files property

inventory property

lists property

locations property

locations: LocationCollection

lots property

notebooks property

notebooks: NotebookCollection

notes property

parameter_groups property

parameter_groups: ParameterGroupCollection

parameters property

parameters: ParameterCollection

pricings property

product_design property

product_design: ProductDesignCollection

projects property

property_data property

property_data: PropertyDataCollection

reports property

roles property

session instance-attribute

session = session or AlbertSession(
    base_url=base_url
    or getenv("ALBERT_BASE_URL")
    or "https://app.albertinvent.com",
    token=token or getenv("ALBERT_TOKEN"),
    client_credentials=client_credentials or from_env(),
    retries=retries,
)

storage_locations property

storage_locations: StorageLocationsCollection

substances property

substances: SubstanceCollection

tags property

tasks property

templates property

un_numbers property

un_numbers: UnNumberCollection

units property

users property

workflows property

workflows: WorkflowCollection

worksheets property

worksheets: WorksheetCollection

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

AttachmentCollection

AttachmentCollection(*, session)

Bases: BaseCollection

AttachmentCollection is a collection class for managing Attachment entities in the Albert platform.

Methods:

Name Description
attach_file_to_note

Attaches an already uploaded file to a note.

delete

Deletes an attachment by ID.

upload_and_attach_file_as_note

Uploads a file and attaches it to a new note. A user can be tagged in the note_text string by using f-string and the User.to_note_mention() method.

Source code in src/albert/collections/attachments.py
def __init__(self, *, session):
    super().__init__(session=session)
    self.base_path = f"/api/{AttachmentCollection._api_version}/attachments"

base_path instance-attribute

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

attach_file_to_note

attach_file_to_note(
    *,
    note_id: str,
    file_name: str,
    file_key: str,
    category: FileCategory = OTHER,
) -> Attachment

Attaches an already uploaded file to a note.

Parameters:

Name Type Description Default
note_id str

The ID of the note to attach the file to.

required
file_name str

The name of the file to attach.

required
file_key str

The unique key of the file to attach (the returned upload name).

required
category FileCategory

The type of file, by default FileCategory.OTHER

OTHER

Returns:

Type Description
Attachment

The related attachment object.

Source code in src/albert/collections/attachments.py
def attach_file_to_note(
    self,
    *,
    note_id: str,
    file_name: str,
    file_key: str,
    category: FileCategory = FileCategory.OTHER,
) -> Attachment:
    """Attaches an already uploaded file to a note.

    Parameters
    ----------
    note_id : str
        The ID of the note to attach the file to.
    file_name : str
        The name of the file to attach.
    file_key : str
        The unique key of the file to attach (the returned upload name).
    category : FileCategory, optional
        The type of file, by default FileCategory.OTHER

    Returns
    -------
    Attachment
        The related attachment object.
    """
    attachment = Attachment(
        parent_id=note_id, name=file_name, key=file_key, namespace="result", category=category
    )
    response = self.session.post(
        url=self.base_path,
        json=attachment.model_dump(by_alias=True, mode="json", exclude_unset=True),
    )
    return Attachment(**response.json())

delete

delete(*, id: str) -> None

Deletes an attachment by ID.

Parameters:

Name Type Description Default
id str

The ID of the attachment to delete.

required
Source code in src/albert/collections/attachments.py
def delete(self, *, id: str) -> None:
    """Deletes an attachment by ID.

    Parameters
    ----------
    id : str
        The ID of the attachment to delete.
    """
    self.session.delete(f"{self.base_path}/{id}")

upload_and_attach_file_as_note

upload_and_attach_file_as_note(
    parent_id: str,
    file_data: IO,
    note_text: str = "",
    file_name: str = "",
) -> Note

Uploads a file and attaches it to a new note. A user can be tagged in the note_text string by using f-string and the User.to_note_mention() method. This allows for easy tagging and referencing of users within notes. example: f"Hello {tagged_user.to_note_mention()}!"

Parameters:

Name Type Description Default
parent_id str

The ID of the parent entity onto which the note will be attached.

required
file_data IO

The file data to upload.

required
note_text str

Any additional text to add to the note, by default ""

''
file_name str

The name of the file, by default ""

''

Returns:

Type Description
Note

The created note.

Source code in src/albert/collections/attachments.py
def upload_and_attach_file_as_note(
    self, parent_id: str, file_data: IO, note_text: str = "", file_name: str = ""
) -> Note:
    """Uploads a file and attaches it to a new note. A user can be tagged in the note_text string by using f-string and the User.to_note_mention() method.
    This allows for easy tagging and referencing of users within notes. example: f"Hello {tagged_user.to_note_mention()}!"

    Parameters
    ----------
    parent_id : str
        The ID of the parent entity onto which the note will be attached.
    file_data : IO
        The file data to upload.
    note_text : str, optional
        Any additional text to add to the note, by default ""
    file_name : str, optional
        The name of the file, by default ""

    Returns
    -------
    Note
        The created note.
    """
    file_type = mimetypes.guess_type(file_name)[0]
    file_collection = self._get_file_collection()
    note_collection = self._get_note_collection()

    file_collection.sign_and_upload_file(
        data=file_data,
        name=file_name,
        namespace=FileNamespace.RESULT.value,
        content_type=file_type,
    )
    file_info = file_collection.get_by_name(
        name=file_name, namespace=FileNamespace.RESULT.value
    )
    note = Note(
        parent_id=parent_id,
        note=note_text,
    )
    registered_note = note_collection.create(note=note)
    self.attach_file_to_note(
        note_id=registered_note.id,
        file_name=file_name,
        file_key=file_info.name,
    )
    return note_collection.get_by_id(id=registered_note.id)

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)

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

BTModelCollection

BTModelCollection(*, session: AlbertSession)

Bases: BaseCollection

BTModelCollection is a collection class for managing Breakthrough model entities.

Breakthrough models can be associated with a parent Breakthrough model session, or a detached without a parent.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Create a new BTModel instance.

delete

Delete a BTModel by ID.

get_by_id

Retrieve a BTModel by its ID.

update

Update an existing BTModel.

Source code in src/albert/collections/btmodel.py
def __init__(self, *, session: AlbertSession):
    super().__init__(session=session)

create

create(
    *,
    model: BTModel,
    parent_id: BTModelSessionId | None = None,
) -> BTModel

Create a new BTModel instance.

Parameters:

Name Type Description Default
model BTModel

The BTModel instance to create.

required
parent_id BTModelSessionId | None

The optional ID of the parent BTModelSession.

None

Returns:

Type Description
BTModel

The created BTModel instance.

Source code in src/albert/collections/btmodel.py
@validate_call
def create(self, *, model: BTModel, parent_id: BTModelSessionId | None = None) -> BTModel:
    """
    Create a new BTModel instance.

    Parameters
    ----------
    model : BTModel
        The BTModel instance to create.
    parent_id : BTModelSessionId | None
        The optional ID of the parent BTModelSession.

    Returns
    -------
    BTModel
        The created BTModel instance.
    """
    base_path = self._get_base_path(parent_id)
    response = self.session.post(
        base_path,
        json=model.model_dump(mode="json", by_alias=True, exclude_none=True),
    )
    return BTModel(**response.json())

delete

delete(
    *,
    id: BTModelId,
    parent_id: BTModelSessionId | None = None,
) -> None

Delete a BTModel by ID.

Parameters:

Name Type Description Default
id BTModelId

The ID of the BTModel to delete.

required
parent_id BTModelSessionId | None

The optional ID of the parent BTModelSession.

None

Returns:

Type Description
None
Source code in src/albert/collections/btmodel.py
@validate_call
def delete(self, *, id: BTModelId, parent_id: BTModelSessionId | None = None) -> None:
    """Delete a BTModel by ID.

    Parameters
    ----------
    id : BTModelId
        The ID of the BTModel to delete.
    parent_id : BTModelSessionId | None
        The optional ID of the parent BTModelSession.

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

get_by_id

get_by_id(
    *,
    id: BTModelId,
    parent_id: BTModelSessionId | None = None,
) -> BTModel

Retrieve a BTModel by its ID.

Parameters:

Name Type Description Default
id BTModelId

The ID of the BTModel to retrieve.

required
parent_id BTModelSessionId | None

The optional ID of the parent BTModelSession.

None

Returns:

Type Description
BTModel

The retrieved BTModel instance.

Source code in src/albert/collections/btmodel.py
@validate_call
def get_by_id(self, *, id: BTModelId, parent_id: BTModelSessionId | None = None) -> BTModel:
    """
    Retrieve a BTModel by its ID.

    Parameters
    ----------
    id : BTModelId
        The ID of the BTModel to retrieve.
    parent_id : BTModelSessionId | None
        The optional ID of the parent BTModelSession.

    Returns
    -------
    BTModel
        The retrieved BTModel instance.
    """
    base_path = self._get_base_path(parent_id)
    response = self.session.get(f"{base_path}/{id}")
    return BTModel(**response.json())

update

update(
    *,
    model: BTModel,
    parent_id: BTModelSessionId | None = None,
) -> BTModel

Update an existing BTModel.

Parameters:

Name Type Description Default
model BTModel

The BTModel instance with updated data.

required
parent_id BTModelSessionId | None

The optional ID of the parent BTModelSession.

None

Returns:

Type Description
BTModel

The updated BTModel instance.

Source code in src/albert/collections/btmodel.py
@validate_call
def update(self, *, model: BTModel, parent_id: BTModelSessionId | None = None) -> BTModel:
    """
    Update an existing BTModel.

    Parameters
    ----------
    model : BTModel
        The BTModel instance with updated data.
    parent_id : BTModelSessionId | None
        The optional ID of the parent BTModelSession.

    Returns
    -------
    BTModel
        The updated BTModel instance.
    """
    base_path = self._get_base_path(parent_id)
    payload = self._generate_patch_payload(
        existing=self.get_by_id(id=model.id, parent_id=parent_id),
        updated=model,
        generate_metadata_diff=False,
    )
    self.session.patch(
        f"{base_path}/{model.id}",
        json=payload.model_dump(mode="json", by_alias=True),
    )
    return self.get_by_id(id=model.id, parent_id=parent_id)

BTModelSessionCollection

BTModelSessionCollection(*, session: AlbertSession)

Bases: BaseCollection

BTModelSessionCollection is a collection class for managing Breakthrough model session entities.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Attributes:

Name Type Description
base_path str

The base path for BTModelSession API requests.

Methods:

Name Description
create
delete

Delete a BTModelSession by ID.

get_by_id
update
Source code in src/albert/collections/btmodel.py
def __init__(self, *, session: AlbertSession):
    super().__init__(session=session)
    self.base_path = f"/api/{BTModelSessionCollection._api_version}/btmodel"

base_path instance-attribute

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

create

create(*, model_session: BTModelSession) -> BTModelSession
Source code in src/albert/collections/btmodel.py
@validate_call
def create(self, *, model_session: BTModelSession) -> BTModelSession:
    response = self.session.post(
        self.base_path,
        json=model_session.model_dump(mode="json", by_alias=True, exclude_none=True),
    )
    return BTModelSession(**response.json())

delete

delete(*, id: BTModelSessionId) -> None

Delete a BTModelSession by ID.

Parameters:

Name Type Description Default
id BTModelSessionId

The ID of the BTModelSession to delete.

required

Returns:

Type Description
None
Source code in src/albert/collections/btmodel.py
@validate_call
def delete(self, *, id: BTModelSessionId) -> None:
    """Delete a BTModelSession by ID.

    Parameters
    ----------
    id : BTModelSessionId
        The ID of the BTModelSession to delete.

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

get_by_id

get_by_id(*, id: BTModelSessionId) -> BTModelSession
Source code in src/albert/collections/btmodel.py
@validate_call
def get_by_id(self, *, id: BTModelSessionId) -> BTModelSession:
    response = self.session.get(f"{self.base_path}/{id}")
    return BTModelSession(**response.json())

update

update(*, model_session: BTModelSession) -> BTModelSession
Source code in src/albert/collections/btmodel.py
@validate_call
def update(self, *, model_session: BTModelSession) -> BTModelSession:
    path = f"{self.base_path}/{model_session.id}"
    payload = self._generate_patch_payload(
        existing=self.get_by_id(id=model_session.id),
        updated=model_session,
    )
    self.session.patch(path, json=payload.model_dump(mode="json", by_alias=True))
    return self.get_by_id(id=model_session.id)

BatchDataCollection

BatchDataCollection(*, session: AlbertSession)

Bases: BaseCollection

BatchDataCollection is a collection class for managing BatchData entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create_batch_data

Create a new batch data entry.

get

Retrieve BatchData by ID.

update_used_batch_amounts

Update the used batch amounts for a given task ID.

Source code in src/albert/collections/batch_data.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the BatchDataCollection with the provided session.

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

base_path instance-attribute

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

create_batch_data

create_batch_data(*, task_id: TaskId)

Create a new batch data entry.

Parameters:

Name Type Description Default
task_id TaskId

The ID of the task for which the batch data is being created.

required

Returns:

Type Description
BatchData

The created BatchData object.

Source code in src/albert/collections/batch_data.py
def create_batch_data(self, *, task_id: TaskId):
    """
    Create a new batch data entry.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task for which the batch data is being created.

    Returns
    -------
    BatchData
        The created BatchData object.
    """
    url = f"{self.base_path}"
    response = self.session.post(url, json={"parentId": task_id})
    return BatchData(**response.json())

get

get(
    *,
    id: TaskId,
    type: BatchDataType = TASK_ID,
    limit: int = 100,
    start_key: str | None = None,
    order_by: OrderBy = DESCENDING,
) -> BatchData

Retrieve BatchData by ID.

Parameters:

Name Type Description Default
id TaskId

Unique Id of the selected type.

required
type BatchDataType

Type of Id for which BatchData will be fetched.

TASK_ID
limit int

The maximum number of list entities to return.

100
start_key str

The primary key of the first item that this operation will evaluate.

None
order_by OrderBy

The order by which to sort the results, by default OrderBy.DESCENDING

DESCENDING

Returns:

Type Description
BatchData

The BatchData object.

Source code in src/albert/collections/batch_data.py
def get(
    self,
    *,
    id: TaskId,
    type: BatchDataType = BatchDataType.TASK_ID,
    limit: int = 100,
    start_key: str | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
) -> BatchData:
    """
    Retrieve BatchData by ID.

    Parameters
    ----------
    id : TaskId
        Unique Id of the selected type.
    type : BatchDataType
        Type of Id for which BatchData will be fetched.
    limit : int, optional
        The maximum number of list entities to return.
    start_key : str, optional
        The primary key of the first item that this operation will evaluate.
    order_by : OrderBy, optional
        The order by which to sort the results, by default OrderBy.DESCENDING
    Returns
    ------
    BatchData
        The BatchData object.
    """
    params = {
        "id": id,
        "limit": limit,
        "type": type,
        "startKey": start_key,
        "orderBy": order_by,
    }
    response = self.session.get(self.base_path, params=params)
    return BatchData(**response.json())

update_used_batch_amounts

update_used_batch_amounts(
    *, task_id: str, patches=list[BatchValuePatchPayload]
) -> None

Update the used batch amounts for a given task ID.

Parameters:

Name Type Description Default
task_id str

The ID of the task to update.

required
patches list[BatchValuePatchPayload]

The patch payloads containing the data to update.

list[BatchValuePatchPayload]

Returns:

Type Description
None

This method does not return anything.

Source code in src/albert/collections/batch_data.py
def update_used_batch_amounts(
    self, *, task_id: str, patches=list[BatchValuePatchPayload]
) -> None:
    """
    Update the used batch amounts for a given task ID.

    Parameters
    ----------
    task_id : str
        The ID of the task to update.
    patches : list[BatchValuePatchPayload]
        The patch payloads containing the data to update.

    Returns
    -------
    None
        This method does not return anything.
    """
    url = f"{self.base_path}/{task_id}/values"
    self.session.patch(
        url,
        json=[
            patch.model_dump(exclude_none=True, by_alias=True, mode="json")
            for patch in patches
        ],
    )

CasCollection

CasCollection(*, session: AlbertSession)

Bases: BaseCollection

CasCollection is a collection class for managing Cas entities on the Albert Platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
cas_exists

Checks if a CAS exists by its number.

create

Creates a new CAS entity.

delete

Deletes a CAS by its ID.

get_by_id

Retrieves a CAS by its ID.

get_by_number

Retrieves a CAS by its number.

list

Lists CAS entities with optional filters.

update

Updates a CAS entity. The updated object must have the same ID as the object you want to update.

Source code in src/albert/collections/cas.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the CasCollection with the provided session.

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

base_path instance-attribute

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

cas_exists

cas_exists(
    *, number: str, exact_match: bool = True
) -> bool

Checks if a CAS exists by its number.

Parameters:

Name Type Description Default
number str

The number of the CAS to check.

required
exact_match bool

Whether to match the number exactly, by default True.

True

Returns:

Type Description
bool

True if the CAS exists, False otherwise.

Source code in src/albert/collections/cas.py
def cas_exists(self, *, number: str, exact_match: bool = True) -> bool:
    """
    Checks if a CAS exists by its number.

    Parameters
    ----------
    number : str
        The number of the CAS to check.
    exact_match : bool, optional
        Whether to match the number exactly, by default True.

    Returns
    -------
    bool
        True if the CAS exists, False otherwise.
    """
    cas_list = self.get_by_number(number=number, exact_match=exact_match)
    return cas_list is not None

create

create(*, cas: str | Cas) -> Cas

Creates a new CAS entity.

Parameters:

Name Type Description Default
cas Union[str, Cas]

The CAS number or Cas object to create.

required

Returns:

Type Description
Cas

The created Cas object.

Source code in src/albert/collections/cas.py
def create(self, *, cas: str | Cas) -> Cas:
    """
    Creates a new CAS entity.

    Parameters
    ----------
    cas : Union[str, Cas]
        The CAS number or Cas object to create.

    Returns
    -------
    Cas
        The created Cas object.
    """
    if isinstance(cas, str):
        cas = Cas(number=cas)
    hit = self.get_by_number(number=cas.number, exact_match=True)
    if hit:
        return hit
    else:
        payload = cas.model_dump(by_alias=True, exclude_unset=True, mode="json")
        response = self.session.post(self.base_path, json=payload)
        cas = Cas(**response.json())
        return cas

delete

delete(*, id: str) -> None

Deletes a CAS by its ID.

Parameters:

Name Type Description Default
id str

The ID of the CAS to delete.

required

Returns:

Type Description
None
Source code in src/albert/collections/cas.py
def delete(self, *, id: str) -> None:
    """
    Deletes a CAS by its ID.

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

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

get_by_id

get_by_id(*, id: str) -> Cas

Retrieves a CAS by its ID.

Parameters:

Name Type Description Default
id str

The ID of the CAS to retrieve.

required

Returns:

Type Description
Cas

The Cas object if found, None otherwise.

Source code in src/albert/collections/cas.py
def get_by_id(self, *, id: str) -> Cas:
    """
    Retrieves a CAS by its ID.

    Parameters
    ----------
    id : str
        The ID of the CAS to retrieve.

    Returns
    -------
    Cas
        The Cas object if found, None otherwise.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    cas = Cas(**response.json())
    return cas

get_by_number

get_by_number(
    *, number: str, exact_match: bool = True
) -> Cas | None

Retrieves a CAS by its number.

Parameters:

Name Type Description Default
number str

The number of the CAS to retrieve.

required
exact_match bool

Whether to match the number exactly, by default True.

True

Returns:

Type Description
Optional[Cas]

The Cas object if found, None otherwise.

Source code in src/albert/collections/cas.py
def get_by_number(self, *, number: str, exact_match: bool = True) -> Cas | None:
    """
    Retrieves a CAS by its number.

    Parameters
    ----------
    number : str
        The number of the CAS to retrieve.
    exact_match : bool, optional
        Whether to match the number exactly, by default True.

    Returns
    -------
    Optional[Cas]
        The Cas object if found, None otherwise.
    """
    found = self.list(number=number)
    if exact_match:
        for f in found:
            if self._clean_cas_number(f.number) == self._clean_cas_number(number):
                return f
    return next(found, None)

list

list(
    *,
    limit: int = 50,
    start_key: str | None = None,
    number: str | None = None,
    id: str | None = None,
    order_by: OrderBy = DESCENDING,
) -> Iterator[Cas]

Lists CAS entities with optional filters.

Parameters:

Name Type Description Default
limit int | None

The maximum number of CAS entities to return, by default 50.

50
start_key str | None

The primary key of the first item that this operation will evaluate.

None
number str | None

Fetches list of CAS by CAS number.

None
id str | None

Fetches list of CAS using the CAS Albert ID.

None
order_by OrderBy

The order by which to sort the results, by default OrderBy.DESCENDING.

DESCENDING

Returns:

Type Description
Iterator[Cas]

An iterator of Cas objects.

Source code in src/albert/collections/cas.py
def list(
    self,
    *,
    limit: int = 50,
    start_key: str | None = None,
    number: str | None = None,
    id: str | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
) -> Iterator[Cas]:
    """
    Lists CAS entities with optional filters.

    Parameters
    ----------
    limit : int | None, optional
        The maximum number of CAS entities to return, by default 50.
    start_key : str | None, optional
        The primary key of the first item that this operation will evaluate.
    number : str | None, optional
        Fetches list of CAS by CAS number.
    id : str | None, optional
        Fetches list of CAS using the CAS Albert ID.
    order_by : OrderBy, optional
        The order by which to sort the results, by default OrderBy.DESCENDING.

    Returns
    -------
    Iterator[Cas]
        An iterator of Cas objects.
    """
    params = {
        "limit": limit,
        "orderBy": order_by.value,
        "startKey": start_key,
        "number": number,
        "albertId": id,
    }
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [Cas(**item) for item in items],
    )

update

update(*, updated_object: Cas) -> Cas

Updates a CAS entity. The updated object must have the same ID as the object you want to update.

Parameters:

Name Type Description Default
updated_object Cas

The Updated Cas object.

required

Returns:

Type Description
Cas

The updated Cas object as it appears in Albert

Source code in src/albert/collections/cas.py
def update(self, *, updated_object: Cas) -> Cas:
    """Updates a CAS entity. The updated object must have the same ID as the object you want to update.

    Parameters
    ----------
    updated_object : Cas
        The Updated Cas object.

    Returns
    -------
    Cas
        The updated Cas object as it appears in Albert
    """
    # Fetch the current object state from the server or database
    existing_cas = self.get_by_id(id=updated_object.id)

    # Generate the PATCH payload
    patch_payload = self._generate_patch_payload(existing=existing_cas, updated=updated_object)
    url = f"{self.base_path}/{updated_object.id}"
    self.session.patch(url, json=patch_payload.model_dump(mode="json", by_alias=True))

    updated_cas = self.get_by_id(id=updated_object.id)
    return updated_cas

CompanyCollection

CompanyCollection(*, session: AlbertSession)

Bases: BaseCollection

CompanyCollection is a collection class for managing Company entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
company_exists

Checks if a company exists by its name.

create

Creates a new company entity.

delete

Deletes a company entity.

get_by_id

Get a company by its ID.

get_by_name

Retrieves a company by its name.

list

Lists company entities with optional filters.

rename

Renames an existing company entity.

update

Update a Company entity. The id of the company must be provided.

Source code in src/albert/collections/companies.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the CompanyCollection with the provided session.

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

base_path instance-attribute

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

company_exists

company_exists(
    *, name: str, exact_match: bool = True
) -> bool

Checks if a company exists by its name.

Parameters:

Name Type Description Default
name str

The name of the company to check.

required
exact_match bool

Whether to match the name exactly, by default True.

True

Returns:

Type Description
bool

True if the company exists, False otherwise.

Source code in src/albert/collections/companies.py
def company_exists(self, *, name: str, exact_match: bool = True) -> bool:
    """
    Checks if a company exists by its name.

    Parameters
    ----------
    name : str
        The name of the company to check.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.

    Returns
    -------
    bool
        True if the company exists, False otherwise.
    """
    companies = self.get_by_name(name=name, exact_match=exact_match)
    return bool(companies)

create

create(
    *, company: str | Company, check_if_exists: bool = True
) -> Company

Creates a new company entity.

Parameters:

Name Type Description Default
company Union[str, Company]

The company name or Company object to create.

required
check_if_exists bool

Whether to check if the company already exists, by default True.

True

Returns:

Type Description
Company

The created Company object.

Source code in src/albert/collections/companies.py
def create(self, *, company: str | Company, check_if_exists: bool = True) -> Company:
    """
    Creates a new company entity.

    Parameters
    ----------
    company : Union[str, Company]
        The company name or Company object to create.
    check_if_exists : bool, optional
        Whether to check if the company already exists, by default True.

    Returns
    -------
    Company
        The created Company object.
    """
    if isinstance(company, str):
        company = Company(name=company)
    hit = self.get_by_name(name=company.name, exact_match=True)
    if check_if_exists and hit:
        logging.warning(f"Company {company.name} already exists with id {hit.id}.")
        return hit

    payload = company.model_dump(by_alias=True, exclude_unset=True, mode="json")
    response = self.session.post(self.base_path, json=payload)
    this_company = Company(**response.json())
    return this_company

delete

delete(*, id: str) -> None

Deletes a company entity.

Parameters:

Name Type Description Default
id str

The ID of the company to delete.

required
Source code in src/albert/collections/companies.py
def delete(self, *, id: str) -> None:
    """Deletes a company entity.

    Parameters
    ----------
    id : str
        The ID of the company to delete.
    """
    url = f"{self.base_path}/{id}"
    self.session.delete(url)

get_by_id

get_by_id(*, id: str) -> Company

Get a company by its ID.

Parameters:

Name Type Description Default
id str

The ID of the company to retrieve.

required

Returns:

Type Description
Company

The Company object.

Source code in src/albert/collections/companies.py
def get_by_id(self, *, id: str) -> Company:
    """
    Get a company by its ID.

    Parameters
    ----------
    id : str
        The ID of the company to retrieve.

    Returns
    -------
    Company
        The Company object.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    company = response.json()
    found_company = Company(**company)
    return found_company

get_by_name

get_by_name(
    *, name: str, exact_match: bool = True
) -> Company | None

Retrieves a company by its name.

Parameters:

Name Type Description Default
name str

The name of the company to retrieve.

required
exact_match bool

Whether to match the name exactly, by default True.

True

Returns:

Type Description
Company

The Company object if found, None otherwise.

Source code in src/albert/collections/companies.py
def get_by_name(self, *, name: str, exact_match: bool = True) -> Company | None:
    """
    Retrieves a company by its name.

    Parameters
    ----------
    name : str
        The name of the company to retrieve.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.

    Returns
    -------
    Company
        The Company object if found, None otherwise.
    """
    found = self.list(name=name, exact_match=exact_match)
    return next(found, None)

list

list(
    *,
    limit: int = 50,
    name: str | list[str] = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Company]

Lists company entities with optional filters.

Parameters:

Name Type Description Default
limit int

The maximum number of companies to return, by default 50.

50
name Union[str, None]

The name of the company to filter by, by default None.

None
exact_match bool

Whether to match the name exactly, by default True.

True

Returns:

Type Description
Iterator

An iterator of Company objects.

Source code in src/albert/collections/companies.py
def list(
    self,
    *,
    limit: int = 50,
    name: str | list[str] = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Company]:
    """
    Lists company entities with optional filters.

    Parameters
    ----------
    limit : int, optional
        The maximum number of companies to return, by default 50.
    name : Union[str, None], optional
        The name of the company to filter by, by default None.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.

    Returns
    -------
    Iterator
        An iterator of Company objects.
    """
    params = {"limit": limit, "dupDetection": "false", "startKey": start_key}
    if name:
        params["name"] = name if isinstance(name, list) else [name]
        params["exactMatch"] = str(exact_match).lower()
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [Company(**item) for item in items],
    )

rename

rename(*, old_name: str, new_name: str) -> Company

Renames an existing company entity.

Parameters:

Name Type Description Default
old_name str

The current name of the company.

required
new_name str

The new name of the company.

required

Returns:

Type Description
Company

The renamed Company object

Source code in src/albert/collections/companies.py
def rename(self, *, old_name: str, new_name: str) -> Company:
    """
    Renames an existing company entity.

    Parameters
    ----------
    old_name : str
        The current name of the company.
    new_name : str
        The new name of the company.

    Returns
    -------
    Company
        The renamed Company object
    """
    company = self.get_by_name(name=old_name, exact_match=True)
    if not company:
        msg = f'Company "{old_name}" not found.'
        logger.error(msg)
        raise AlbertException(msg)
    company_id = company.id
    endpoint = f"{self.base_path}/{company_id}"
    payload = {
        "data": [
            {
                "operation": "update",
                "attribute": "name",
                "oldValue": old_name,
                "newValue": new_name,
            }
        ]
    }
    self.session.patch(endpoint, json=payload)
    updated_company = self.get_by_id(id=company_id)
    return updated_company

update

update(*, company: Company) -> Company

Update a Company entity. The id of the company must be provided.

Parameters:

Name Type Description Default
company Company

The updated Company object.

required

Returns:

Type Description
Company

The updated Company object as registered in Albert.

Source code in src/albert/collections/companies.py
def update(self, *, company: Company) -> Company:
    """Update a Company entity. The id of the company must be provided.

    Parameters
    ----------
    company : Company
        The updated Company object.

    Returns
    -------
    Company
        The updated Company object as registered in Albert.
    """
    # Fetch the current object state from the server or database
    current_object = self.get_by_id(id=company.id)

    # Generate the PATCH payload
    patch_payload = self._generate_patch_payload(existing=current_object, updated=company)
    url = f"{self.base_path}/{company.id}"
    self.session.patch(url, json=patch_payload.model_dump(mode="json", by_alias=True))
    updated_company = self.get_by_id(id=company.id)
    return updated_company

CustomFieldCollection

CustomFieldCollection(*, session: AlbertSession)

Bases: BaseCollection

CustomFieldCollection is a collection class for managing CustomField entities in the Albert platform.

This collection provides methods to create, update, retrieve, and list custom fields. CustomFields allow you to store custom metadata on a Project, InventoryItem, User, BaseTask (Tasks), and Lot.

The FieldType used determines the shape of the metadata field's value. If the FieldType is LIST, then the FieldCategory defines the ACL needed to add new allowed items to the given list:

  • FieldCategory.USER_DEFINED: allows general users to add items
  • FieldCategory.BUSINESS_DEFINED: only admins can add new items to the list
Example
# Creating some custom fields
from albert import Albert
from albert.resources.custom_fields import CustomField, FieldCategory, FieldType, ServiceType
from albert.resources.lists import ListItem
from albert.resources.project import Project

# Initialize the Albert client
client = Albert()

# Define the custom fields
stage_gate_field = CustomField(
    name="stage_gate_status",
    display_name="Stage Gate",
    field_type=FieldType.LIST,
    service=ServiceType.PROJECTS,
    min=1,
    max=1,
    category=FieldCategory.BUSINESS_DEFINED  # Defined by the business
)
justification_field = CustomField(
    name="justification",
    display_name="Project Justification",
    field_type=FieldType.STRING,
    service=ServiceType.PROJECTS,
)

# Create the custom fields
client.custom_fields.create(custom_field=stage_gate_field)
client.custom_fields.create(custom_field=justification_field)

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Create a new CustomField item.

get_by_id

Get a CustomField item by its ID.

get_by_name

Get a CustomField item by its name.

list

Searches for CustomField items based on the provided parameters.

update

Update a CustomField item.

Source code in src/albert/collections/custom_fields.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the CasCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, custom_field: CustomField) -> CustomField

Create a new CustomField item.

Parameters:

Name Type Description Default
custom_field CustomField

The CustomField item to create.

required

Returns:

Type Description
CustomField

The created CustomField item with its ID.

Source code in src/albert/collections/custom_fields.py
def create(self, *, custom_field: CustomField) -> CustomField:
    """Create a new CustomField item.

    Parameters
    ----------
    custom_field : CustomField
        The CustomField item to create.

    Returns
    -------
    CustomField
        The created CustomField item with its ID.
    """
    response = self.session.post(
        self.base_path,
        json=custom_field.model_dump(by_alias=True, exclude_none=True, mode="json"),
    )
    return CustomField(**response.json())

get_by_id

get_by_id(*, id: str) -> CustomField

Get a CustomField item by its ID.

Parameters:

Name Type Description Default
id str

The ID of the CustomField item.

required

Returns:

Type Description
CustomField

The CustomField item.

Source code in src/albert/collections/custom_fields.py
def get_by_id(self, *, id: str) -> CustomField:
    """Get a CustomField item by its ID.

    Parameters
    ----------
    id : str
        The ID of the CustomField item.

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

get_by_name

get_by_name(
    *, name: str, service: ServiceType | None = None
) -> CustomField | None

Get a CustomField item by its name.

Parameters:

Name Type Description Default
name str

The name of the CustomField item.

required
service ServiceType | None

The service the field relates to, by default None

None

Returns:

Type Description
CustomField | None

The CustomField item, or None if not found.

Source code in src/albert/collections/custom_fields.py
def get_by_name(self, *, name: str, service: ServiceType | None = None) -> CustomField | None:
    """Get a CustomField item by its name.

    Parameters
    ----------
    name : str
        The name of the CustomField item.
    service : ServiceType | None, optional
        The service the field relates to, by default None

    Returns
    -------
    CustomField | None
        The CustomField item, or None if not found.
    """
    for custom_field in self.list(name=name, service=service):
        if custom_field.name.lower() == name.lower():
            return custom_field
    return None

list

list(
    *,
    name: str | None = None,
    service: ServiceType | None = None,
    lookup_column: bool | None = None,
    lookup_row: bool | None = None,
) -> Iterator[CustomField]

Searches for CustomField items based on the provided parameters.

Parameters:

Name Type Description Default
name str | None

The name of the field, by default None

None
service ServiceType | None

The related service the field is in, by default None

None
lookup_column bool | None

Whether the field relates to a lookup column, by default None

None
lookup_row bool | None

Whether the field relates to a lookup row, by default None

None

Yields:

Type Description
Iterator[CustomField]

Returns an iterator of CustomField items matching the search criteria.

Source code in src/albert/collections/custom_fields.py
def list(
    self,
    *,
    name: str | None = None,
    service: ServiceType | None = None,
    lookup_column: bool | None = None,
    lookup_row: bool | None = None,
) -> Iterator[CustomField]:
    """Searches for CustomField items based on the provided parameters.

    Parameters
    ----------
    name : str | None, optional
        The name of the field, by default None
    service : ServiceType | None, optional
        The related service the field is in, by default None
    lookup_column : bool | None, optional
        Whether the field relates to a lookup column, by default None
    lookup_row : bool | None, optional
        Whether the field relates to a lookup row, by default None

    Yields
    ------
    Iterator[CustomField]
        Returns an iterator of CustomField items matching the search criteria.
    """
    params = {
        "name": name,
        "service": service if service else None,
        "lookupColumn": json.dumps(lookup_column) if lookup_column is not None else None,
        "lookupRow": json.dumps(lookup_row) if lookup_row is not None else None,
    }
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        params=params,
        session=self.session,
        deserialize=lambda items: [CustomField(**item) for item in items],
    )

update

update(*, custom_field: CustomField) -> CustomField

Update a CustomField item.

Parameters:

Name Type Description Default
custom_field CustomField

The updated CustomField item. The ID must be set and match the Field you want to update.

required

Returns:

Type Description
CustomField

The updated CustomField item as registered in Albert.

Source code in src/albert/collections/custom_fields.py
def update(self, *, custom_field: CustomField) -> CustomField:
    """Update a CustomField item.

    Parameters
    ----------
    custom_field : CustomField
        The updated CustomField item. The ID must be set and match the Field you want to update.

    Returns
    -------
    CustomField
        The updated CustomField item as registered in Albert.
    """
    # fetch current object state
    current_object = self.get_by_id(id=custom_field.id)

    # generate the patch payload
    payload = self._generate_patch_payload(
        existing=current_object,
        updated=custom_field,
        generate_metadata_diff=False,
        stringify_values=False,
    )

    for patch in payload.data:
        if (
            patch.attribute in ("hidden", "search", "lkpColumn", "lkpRow")
            and patch.operation == "add"
        ):
            patch.operation = "update"
            patch.old_value = False
        if (
            patch.attribute in ("entityCategory")
            and patch.operation == "add"
            and isinstance(patch.new_value, list)
        ):
            patch.new_value = patch.new_value[0]

    # run patch
    url = f"{self.base_path}/{custom_field.id}"
    self.session.patch(url, json=payload.model_dump(mode="json", by_alias=True))
    updated_ctf = self.get_by_id(id=custom_field.id)
    return updated_ctf

CustomTemplatesCollection

CustomTemplatesCollection(*, session: AlbertSession)

Bases: BaseCollection

CustomTemplatesCollection is a collection class for managing CustomTemplate entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
get_by_id

Get a Custom Template by ID

list

Searches for custom templates matching the provided criteria.

Source code in src/albert/collections/custom_templates.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the CustomTemplatesCollection with the provided session.

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

base_path instance-attribute

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

get_by_id

get_by_id(*, id) -> CustomTemplate

Get a Custom Template by ID

Parameters:

Name Type Description Default
id str

id of the custom template

required

Returns:

Type Description
CustomTemplate

The CutomTemplate with the provided ID (or None if not found)

Source code in src/albert/collections/custom_templates.py
def get_by_id(self, *, id) -> CustomTemplate:
    """Get a Custom Template by ID

    Parameters
    ----------
    id : str
        id of the custom template

    Returns
    -------
    CustomTemplate
        The CutomTemplate with the provided ID (or None if not found)
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return CustomTemplate(**response.json())

list

list(
    *,
    text: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> Iterator[CustomTemplate]

Searches for custom templates matching the provided criteria.

Parameters:

Name Type Description Default
text str | None

The text to search for, by default None

None

Yields:

Type Description
Iterator[CustomTemplate]

An iterator of CustomTemplate items matching the search criteria.

Source code in src/albert/collections/custom_templates.py
def list(
    self,
    *,
    text: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> Iterator[CustomTemplate]:
    """Searches for custom templates matching the provided criteria.

    Parameters
    ----------
    text : str | None, optional
        The text to search for, by default None


    Yields
    ------
    Iterator[CustomTemplate]
        An iterator of CustomTemplate items matching the search criteria.
    """

    def deserialize(items: list[dict]) -> Iterator[CustomTemplate]:
        for item in items:
            id = item["albertId"]
            try:
                yield self.get_by_id(id=id)
            except AlbertHTTPError as e:
                logger.warning(f"Error fetching custom template {id}: {e}")

    params = {"limit": limit, "offset": offset, "text": text}

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        params=params,
        deserialize=deserialize,
    )

DataColumnCollection

DataColumnCollection(*, session: AlbertSession)

Bases: BaseCollection

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

Methods:

Name Description
create

Create a new data column entity.

delete

Delete a data column entity.

get_by_id

Get a data column by its ID.

get_by_name

Get a data column by its name.

list

Lists data column entities with optional filters.

update

Update a data column entity.

Source code in src/albert/collections/data_columns.py
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 instance-attribute

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

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

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

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

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

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

DataTemplateCollection

DataTemplateCollection(*, session: AlbertSession)

Bases: BaseCollection

DataTemplateCollection is a collection class for managing DataTemplate entities in the Albert platform.

Methods:

Name Description
add_data_columns

Adds data columns to a data template.

add_parameters

Adds parameters to a data template.

create

Creates a new data template.

delete

Deletes a data template by its ID.

get_by_id

Get a data template by its ID.

get_by_ids

Get a list of data templates by their IDs.

get_by_name

Get a data template by its name.

list

Lists data template entities with optional filters.

update

Updates a data template.

Source code in src/albert/collections/data_templates.py
def __init__(self, *, session: AlbertSession):
    super().__init__(session=session)
    self.base_path = f"/api/{DataTemplateCollection._api_version}/datatemplates"

base_path instance-attribute

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

add_data_columns

add_data_columns(
    *,
    data_template_id: DataTemplateId,
    data_columns: list[DataColumnValue],
) -> DataTemplate

Adds data columns to a data template.

Parameters:

Name Type Description Default
data_template_id str

The ID of the data template to add the columns to.

required
data_columns list[DataColumnValue]

The list of DataColumnValue objects to add to the data template.

required

Returns:

Type Description
DataTemplate

The updated DataTemplate object.

Source code in src/albert/collections/data_templates.py
def add_data_columns(
    self, *, data_template_id: DataTemplateId, data_columns: list[DataColumnValue]
) -> DataTemplate:
    """Adds data columns to a data template.

    Parameters
    ----------
    data_template_id : str
        The ID of the data template to add the columns to.
    data_columns : list[DataColumnValue]
        The list of DataColumnValue objects to add to the data template.

    Returns
    -------
    DataTemplate
        The updated DataTemplate object.
    """
    # if there are enum values, we need to add them as an allowed enum
    for column in data_columns:
        if (
            column.validation
            and len(column.validation) > 0
            and isinstance(column.validation[0].value, list)
        ):
            for enum_value in column.validation[0].value:
                self.session.put(
                    f"{self.base_path}/{data_template_id}/datacolumns/{column.sequence}/enums",
                    json=[
                        enum_value.model_dump(mode="json", by_alias=True, exclude_none=True)
                    ],
                )

    payload = {
        "DataColumns": [
            x.model_dump(mode="json", by_alias=True, exclude_none=True) for x in data_columns
        ]
    }
    self.session.put(
        f"{self.base_path}/{data_template_id}/datacolumns",
        json=payload,
    )
    return self.get_by_id(id=data_template_id)

add_parameters

add_parameters(
    *,
    data_template_id: DataTemplateId,
    parameters: list[ParameterValue],
) -> DataTemplate

Adds parameters to a data template.

Parameters:

Name Type Description Default
data_template_id str

The ID of the data template to add the columns to.

required
parameters list[ParameterValue]

The list of ParameterValue objects to add to the data template.

required

Returns:

Type Description
DataTemplate

The updated DataTemplate object.

Source code in src/albert/collections/data_templates.py
def add_parameters(
    self, *, data_template_id: DataTemplateId, parameters: list[ParameterValue]
) -> DataTemplate:
    """Adds parameters to a data template.

    Parameters
    ----------
    data_template_id : str
        The ID of the data template to add the columns to.
    parameters : list[ParameterValue]
        The list of ParameterValue objects to add to the data template.

    Returns
    -------
    DataTemplate
        The updated DataTemplate object.
    """
    # make sure the parameter values have a default validaion of string type.
    initial_enum_values = {}  # use index to track the enum values
    if parameters is None or len(parameters) == 0:
        return self.get_by_id(id=data_template_id)
    for i, param in enumerate(parameters):
        if (
            param.validation
            and len(param.validation) > 0
            and param.validation[0].datatype == DataType.ENUM
        ):
            initial_enum_values[i] = param.validation[0].value
            param.validation[0].value = None
            param.validation[0].datatype = DataType.STRING

    payload = {
        "Parameters": [
            x.model_dump(mode="json", by_alias=True, exclude_none=True) for x in parameters
        ]
    }
    # if there are enum values, we need to add them as an allowed enum
    response = self.session.put(
        f"{self.base_path}/{data_template_id}/parameters",
        json=payload,
    )
    returned_parameters = [ParameterValue(**x) for x in response.json()["Parameters"]]
    for i, param in enumerate(returned_parameters):
        if i in initial_enum_values:
            param.validation[0].value = initial_enum_values[i]
            param.validation[0].datatype = DataType.ENUM
    self._add_param_enums(
        data_template_id=data_template_id,
        new_parameters=returned_parameters,
    )
    dt_with_params = self.get_by_id(id=data_template_id)
    for i, param in enumerate(dt_with_params.parameter_values):
        if i in initial_enum_values:
            param.validation[0].value = initial_enum_values[i]
            param.validation[0].datatype = DataType.ENUM

    return self.update(data_template=dt_with_params)

create

create(*, data_template: DataTemplate) -> DataTemplate

Creates a new data template.

Parameters:

Name Type Description Default
data_template DataTemplate

The DataTemplate object to create.

required

Returns:

Type Description
DataTemplate

The registered DataTemplate object with an ID.

Source code in src/albert/collections/data_templates.py
def create(self, *, data_template: DataTemplate) -> DataTemplate:
    """Creates a new data template.

    Parameters
    ----------
    data_template : DataTemplate
        The DataTemplate object to create.

    Returns
    -------
    DataTemplate
        The registered DataTemplate object with an ID.
    """
    # Preprocess data_column_values to set validation to None if it is an empty list
    # Handle a bug in the API where validation is an empty list
    # https://support.albertinvent.com/hc/en-us/requests/9177
    for column_value in data_template.data_column_values:
        if isinstance(column_value.validation, list) and len(column_value.validation) == 0:
            column_value.validation = None
    # remove them on the initial post
    parameter_values = data_template.parameter_values
    data_template.parameter_values = None
    response = self.session.post(
        self.base_path,
        json=data_template.model_dump(mode="json", by_alias=True, exclude_none=True),
    )
    dt = DataTemplate(**response.json())
    dt.parameter_values = parameter_values
    if data_template.parameter_values is None or len(data_template.parameter_values) == 0:
        return dt
    else:
        return self.add_parameters(
            data_template_id=dt.id, parameters=data_template.parameter_values
        )

delete

delete(*, id: DataTemplateId) -> None

Deletes a data template by its ID.

Parameters:

Name Type Description Default
id str

The ID of the data template to delete.

required
Source code in src/albert/collections/data_templates.py
def delete(self, *, id: DataTemplateId) -> None:
    """Deletes a data template by its ID.

    Parameters
    ----------
    id : str
        The ID of the data template to delete.
    """
    self.session.delete(f"{self.base_path}/{id}")

get_by_id

get_by_id(*, id: DataTemplateId) -> DataTemplate

Get a data template by its ID.

Parameters:

Name Type Description Default
id DataTemplateId

The ID of the data template to get.

required

Returns:

Type Description
DataTemplate

The data template object on match or None

Source code in src/albert/collections/data_templates.py
def get_by_id(self, *, id: DataTemplateId) -> DataTemplate:
    """Get a data template by its ID.

    Parameters
    ----------
    id : DataTemplateId
        The ID of the data template to get.

    Returns
    -------
    DataTemplate
        The data template object on match or None
    """
    response = self.session.get(f"{self.base_path}/{id}")
    return DataTemplate(**response.json())

get_by_ids

get_by_ids(
    *, ids: list[DataTemplateId]
) -> list[DataTemplate]

Get a list of data templates by their IDs.

Parameters:

Name Type Description Default
ids list[DataTemplateId]

The list of DataTemplate IDs to get.

required

Returns:

Type Description
list[DataTemplate]

A list of DataTemplate objects with the provided IDs.

Source code in src/albert/collections/data_templates.py
def get_by_ids(self, *, ids: list[DataTemplateId]) -> list[DataTemplate]:
    """Get a list of data templates by their IDs.

    Parameters
    ----------
    ids : list[DataTemplateId]
        The list of DataTemplate IDs to get.

    Returns
    -------
    list[DataTemplate]
        A list of DataTemplate objects with the provided IDs.
    """
    url = f"{self.base_path}/ids"
    batches = [ids[i : i + 250] for i in range(0, len(ids), 250)]
    return [
        DataTemplate(**item)
        for batch in batches
        for item in self.session.get(url, params={"id": batch}).json()["Items"]
    ]

get_by_name

get_by_name(*, name: str) -> DataTemplate | None

Get a data template by its name.

Parameters:

Name Type Description Default
name str

The name of the data template to get.

required

Returns:

Type Description
DataTemplate | None

The matching data template object or None if not found.

Source code in src/albert/collections/data_templates.py
def get_by_name(self, *, name: str) -> DataTemplate | None:
    """Get a data template by its name.

    Parameters
    ----------
    name : str
        The name of the data template to get.

    Returns
    -------
    DataTemplate | None
        The matching data template object or None if not found.
    """
    hits = list(self.list(name=name))
    for h in hits:
        if h.name.lower() == name.lower():
            return h
    return None

list

list(
    *,
    name: str | None = None,
    user_id: str | None = None,
    order_by: OrderBy = DESCENDING,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[DataTemplate]

Lists data template entities with optional filters.

Parameters:

Name Type Description Default
name Union[str, None]

The name of the data template to filter by, by default None.

None
user_id str

user_id to filter by, by default None.

None
order_by OrderBy

The order by which to sort the results, by default OrderBy.DESCENDING.

DESCENDING

Returns:

Type Description
Iterator[DataTemplate]

An iterator of DataTemplate objects matching the provided criteria.

Source code in src/albert/collections/data_templates.py
def list(
    self,
    *,
    name: str | None = None,
    user_id: str | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[DataTemplate]:
    """
    Lists data template entities with optional filters.

    Parameters
    ----------
    name : Union[str, None], optional
        The name of the data template to filter by, by default None.
    user_id : str, optional
        user_id to filter by, by default None.
    order_by : OrderBy, optional
        The order by which to sort the results, by default OrderBy.DESCENDING.

    Returns
    -------
    Iterator[DataTemplate]
        An iterator of DataTemplate objects matching the provided criteria.
    """

    def deserialize(items: list[dict]) -> Iterator[DataTemplate]:
        for item in items:
            id = item["albertId"]
            try:
                yield self.get_by_id(id=id)
            except AlbertHTTPError as e:
                logger.warning(f"Error fetching parameter group {id}: {e}")
        # get by ids is not currently returning metadata correctly, so temp fixing this
        # return self.get_by_ids(ids=[x["albertId"] for x in items])

    params = {
        "limit": limit,
        "offset": offset,
        "order": OrderBy(order_by).value if order_by else None,
        "text": name,
        "userId": user_id,
    }

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        deserialize=deserialize,
        params=params,
    )

update

update(*, data_template: DataTemplate) -> DataTemplate

Updates a data template.

Parameters:

Name Type Description Default
data_template DataTemplate

The DataTemplate object to update. The ID must be set and matching the ID of the DataTemplate to update.

required

Returns:

Type Description
DataTemplate

The Updated DataTemplate object.

Source code in src/albert/collections/data_templates.py
def update(self, *, data_template: DataTemplate) -> DataTemplate:
    """Updates a data template.

    Parameters
    ----------
    data_template : DataTemplate
        The DataTemplate object to update. The ID must be set and matching the ID of the DataTemplate to update.

    Returns
    -------
    DataTemplate
        The Updated DataTemplate object.
    """

    existing = self.get_by_id(id=data_template.id)

    base_payload = self._generate_patch_payload(existing=existing, updated=data_template)

    path = f"{self.base_path}/{existing.id}"
    (
        general_patches,
        new_data_columns,
        data_column_enum_patches,
        new_parameters,
        parameter_enum_patches,
        parameter_patches,
    ) = generate_data_template_patches(
        initial_patches=base_payload,
        updated_data_template=data_template,
        existing_data_template=existing,
    )

    if len(new_data_columns) > 0:
        self.session.put(
            f"{self.base_path}/{existing.id}/datacolumns",
            json={
                "DataColumns": [
                    x.model_dump(mode="json", by_alias=True, exclude_none=True)
                    for x in new_data_columns
                ],
            },
        )
    if len(data_column_enum_patches) > 0:
        for sequence, enum_patches in data_column_enum_patches.items():
            if len(enum_patches) == 0:
                continue
            self.session.put(
                f"{self.base_path}/{existing.id}/datacolumns/{sequence}/enums",
                json=enum_patches,  # these are simple dicts for now
            )
    if len(new_parameters) > 0:
        self.session.put(
            f"{self.base_path}/{existing.id}/parameters",
            json={
                "Parameters": [
                    x.model_dump(mode="json", by_alias=True, exclude_none=True)
                    for x in new_parameters
                ],
            },
        )
    if len(parameter_enum_patches) > 0:
        for sequence, enum_patches in parameter_enum_patches.items():
            if len(enum_patches) == 0:
                continue
            self.session.put(
                f"{self.base_path}/{existing.id}/parameters/{sequence}/enums",
                json=enum_patches,  # these are simple dicts for now
            )
    if len(parameter_patches) > 0:
        payload = PGPatchPayload(data=parameter_patches)
        self.session.patch(
            path + "/parameters",
            json=payload.model_dump(mode="json", by_alias=True, exclude_none=True),
        )
    if len(general_patches.data) > 0:
        payload = GeneralPatchPayload(data=general_patches.data)
        self.session.patch(
            path,
            json=payload.model_dump(mode="json", by_alias=True, exclude_none=True),
        )
    return self.get_by_id(id=data_template.id)

FileCollection

FileCollection(*, session: AlbertSession)

Bases: BaseCollection

FileCollection is a collection class for managing File entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
get_by_name

Gets a file by name and namespace.

get_signed_download_url

Get a signed download URL for a file.

get_signed_upload_url

Get a signed upload URL for a file.

sign_and_upload_file

Sign and upload a file to Albert.

Source code in src/albert/collections/files.py
def __init__(self, *, session: AlbertSession):
    """
    Initialize the FileCllection with the provided session.

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

base_path instance-attribute

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

get_by_name

get_by_name(
    *,
    name: str,
    namespace: FileNamespace,
    generic: bool = False,
) -> FileInfo

Gets a file by name and namespace.

Parameters:

Name Type Description Default
name str

The Name of the file

required
namespace FileNamespace

The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)

required
generic bool

description, by default False

False

Returns:

Type Description
FileInfo

The file information related to the matching file.

Source code in src/albert/collections/files.py
def get_by_name(
    self,
    *,
    name: str,
    namespace: FileNamespace,
    generic: bool = False,
) -> FileInfo:
    """Gets a file by name and namespace.

    Parameters
    ----------
    name : str
        The Name of the file
    namespace : FileNamespace
        The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)
    generic : bool, optional
        _description_, by default False

    Returns
    -------
    FileInfo
        The file information related to the matching file.
    """
    params = {
        "name": name,
        "namespace": namespace,
        "generic": json.dumps(generic),
    }
    response = self.session.get(f"{self.base_path}/info", params=params)
    return FileInfo(**response.json())

get_signed_download_url

get_signed_download_url(
    *,
    name: str,
    namespace: FileNamespace,
    version_id: str | None = None,
    generic: bool = False,
    category: FileCategory | None = None,
) -> str

Get a signed download URL for a file.

Parameters:

Name Type Description Default
name str

The Name of the file

required
namespace FileNamespace

The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)

required
version_id str | None

The version of the file, by default None

None
category FileCategory | None

The file category (E.g., SDS, OTHER), by default None

None

Returns:

Type Description
str

description

Source code in src/albert/collections/files.py
def get_signed_download_url(
    self,
    *,
    name: str,
    namespace: FileNamespace,
    version_id: str | None = None,
    generic: bool = False,
    category: FileCategory | None = None,
) -> str:
    """Get a signed download URL for a file.

    Parameters
    ----------
    name : str
        The Name of the file
    namespace : FileNamespace
        The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)
    version_id : str | None, optional
        The version of the file, by default None
    category : FileCategory | None, optional
        The file category (E.g., SDS, OTHER), by default None

    Returns
    -------
    str
        _description_
    """
    params = {
        "name": name,
        "namespace": namespace,
        "versionId": version_id,
        "generic": json.dumps(generic),
        "category": category,
    }
    response = self.session.get(
        f"{self.base_path}/sign",
        params={k: v for k, v in params.items() if v is not None},
    )
    return response.json()["URL"]

get_signed_upload_url

get_signed_upload_url(
    *,
    name: str,
    namespace: FileNamespace,
    content_type: str,
    generic: bool = False,
    category: FileCategory | None = None,
) -> str

Get a signed upload URL for a file.

Parameters:

Name Type Description Default
name str

The Name of the file

required
namespace FileNamespace

The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)

required
content_type str

The content type of the file

required
category FileCategory | None

The File category (E.g., SDS, OTHER), by default None

None

Returns:

Type Description
str

description

Source code in src/albert/collections/files.py
def get_signed_upload_url(
    self,
    *,
    name: str,
    namespace: FileNamespace,
    content_type: str,
    generic: bool = False,
    category: FileCategory | None = None,
) -> str:
    """Get a signed upload URL for a file.

    Parameters
    ----------
    name : str
        The Name of the file
    namespace : FileNamespace
        The namespace of the file (e.g. AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)
    content_type : str
        The content type of the file
    category : FileCategory | None, optional
        The File category (E.g., SDS, OTHER), by default None

    Returns
    -------
    str
        _description_
    """
    params = {"generic": json.dumps(generic)}

    post_body = SignURLPOST(
        files=[
            SignURLPOSTFile(
                name=name,
                namespace=namespace,
                content_type=content_type,
                category=category,
            )
        ]
    )

    response = self.session.post(
        f"{self.base_path}/sign",
        json=post_body.model_dump(by_alias=True, exclude_unset=True, mode="json"),
        params=params,
    )
    return response.json()[0]["URL"]

sign_and_upload_file

sign_and_upload_file(
    data: IO,
    name: str,
    namespace: FileNamespace,
    content_type: str,
    generic: bool = False,
    category: FileCategory | None = None,
) -> None

Sign and upload a file to Albert.

Parameters:

Name Type Description Default
data IO

The file data

required
name str

The name of the file

required
namespace FileNamespace

The File Namespace (e.g., AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)

required
content_type str

The content type of the file

required
category FileCategory | None

The category of the file (E.g., SDS, OTHER), by default None

None
Source code in src/albert/collections/files.py
def sign_and_upload_file(
    self,
    data: IO,
    name: str,
    namespace: FileNamespace,
    content_type: str,
    generic: bool = False,
    category: FileCategory | None = None,
) -> None:
    """Sign and upload a file to Albert.

    Parameters
    ----------
    data : IO
        The file data
    name : str
        The name of the file
    namespace : FileNamespace
        The File Namespace (e.g., AGENT, BREAKTHROUGH, PIPELINE, PUBLIC, RESULT, SDS)
    content_type : str
        The content type of the file
    category : FileCategory | None, optional
        The category of the file (E.g., SDS, OTHER), by default None
    """
    upload_url = self.get_signed_upload_url(
        name=name,
        namespace=namespace,
        content_type=content_type,
        generic=generic,
        category=category,
    )
    requests.put(upload_url, data=data, headers={"Content-Type": content_type})

InventoryCollection

InventoryCollection(*, session: AlbertSession)

Bases: BaseCollection

InventoryCollection is a collection class for managing Inventory Item entities in the Albert platform.

Parameters:

Name Type Description Default
session Albert

The Albert session instance.

required

Methods:

Name Description
add_specs

Add inventory specs to the inventory item.

create

Create a new inventory item.

delete

Delete an inventory item by its ID.

get_all_facets

Get available facets for inventory items based on the provided filters.

get_by_id

Retrieve an inventory item by its ID.

get_by_ids

Retrieve an set of inventory items by their IDs.

get_facet_by_name

Returns a specific facet by its name with all the filters applied to the search.

get_match_or_none

Get a matching inventory item or return None if not found.

get_specs

Get the specs for a list of inventory items.

inventory_exists

Check if an inventory item exists.

list

List inventory items with optional filters.

merge

Merge one or multiple child inventory into a parent inventory item.

search

Get a list of inventory items that match the search criteria and

update

Update an inventory item.

Source code in src/albert/collections/inventory.py
def __init__(self, *, session: AlbertSession):
    """
    InventoryCollection is a collection class for managing inventory items.

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

base_path instance-attribute

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

add_specs

add_specs(
    *,
    inventory_id: InventoryId,
    specs: InventorySpec | list[InventorySpec],
) -> InventorySpecList

Add inventory specs to the inventory item.

An InventorySpec is a property that was not directly measured via a task, but is a generic property of that inentory item.

Parameters:

Name Type Description Default
inventory_id InventoryId

The Albert ID of the inventory item to add the specs to

required
specs list[InventorySpec]

List of InventorySpec objects to add to the inventory item, which described the value and, optionally, the conditions associated with the value (via workflow).

required

Returns:

Type Description
InventorySpecList

The list of InventorySpecs attached to the InventoryItem.

Source code in src/albert/collections/inventory.py
@validate_call
def add_specs(
    self,
    *,
    inventory_id: InventoryId,
    specs: InventorySpec | list[InventorySpec],
) -> InventorySpecList:
    """Add inventory specs to the inventory item.

    An `InventorySpec` is a property that was not directly measured via a task,
    but is a generic property of that inentory item.

    Parameters
    ----------
    inventory_id : InventoryId
        The Albert ID of the inventory item to add the specs to
    specs : list[InventorySpec]
        List of InventorySpec objects to add to the inventory item,
        which described the value and, optionally,
        the conditions associated with the value (via workflow).

    Returns
    -------
    InventorySpecList
        The list of InventorySpecs attached to the InventoryItem.
    """
    if isinstance(specs, InventorySpec):
        specs = [specs]
    response = self.session.put(
        url=f"{self.base_path}/{inventory_id}/specs",
        json=[x.model_dump(exclude_unset=True, by_alias=True, mode="json") for x in specs],
    )
    return InventorySpecList(**response.json())

create

create(
    *,
    inventory_item: InventoryItem,
    avoid_duplicates: bool = True,
) -> InventoryItem

Create a new inventory item.

Parameters:

Name Type Description Default
inventory_item InventoryItem

The inventory item to create.

required
avoid_duplicates bool

Whether to avoid creating duplicate items (default is True).

True

Returns:

Type Description
InventoryItem

The created inventory item.

Source code in src/albert/collections/inventory.py
def create(
    self,
    *,
    inventory_item: InventoryItem,
    avoid_duplicates: bool = True,
) -> InventoryItem:
    """
    Create a new inventory item.

    Parameters
    ----------
    inventory_item : InventoryItem
        The inventory item to create.
    avoid_duplicates : bool, optional
        Whether to avoid creating duplicate items (default is True).

    Returns
    -------
    InventoryItem
        The created inventory item.
    """
    category = (
        inventory_item.category
        if isinstance(inventory_item.category, str)
        else inventory_item.category.value
    )
    if category == InventoryCategory.FORMULAS.value:
        # This will need to interact with worksheets
        raise NotImplementedError("Registrations of formulas not yet implemented")
    tag_collection = TagCollection(session=self.session)
    if inventory_item.tags is not None and inventory_item.tags != []:
        all_tags = [
            tag_collection.create(tag=t) if t.id is None else t for t in inventory_item.tags
        ]
        inventory_item.tags = all_tags
    if inventory_item.company and inventory_item.company.id is None:
        company_collection = CompanyCollection(session=self.session)
        inventory_item.company = company_collection.create(company=inventory_item.company)
    # Check to see if there is a match on name + Company already
    if avoid_duplicates:
        existing = self.get_match_or_none(inventory_item=inventory_item)
        if isinstance(existing, InventoryItem):
            logging.warning(
                f"Inventory item already exists with name {existing.name} and company {existing.company.name}, returning existing item."
            )
            return existing
    response = self.session.post(
        self.base_path,
        json=inventory_item.model_dump(by_alias=True, exclude_none=True, mode="json"),
    )
    return InventoryItem(**response.json())

delete

delete(*, id: InventoryId) -> None

Delete an inventory item by its ID.

Parameters:

Name Type Description Default
id InventoryId

The ID of the inventory item.

required

Returns:

Type Description
None
Source code in src/albert/collections/inventory.py
@validate_call
def delete(self, *, id: InventoryId) -> None:
    """
    Delete an inventory item by its ID.

    Parameters
    ----------
    id : InventoryId
        The ID of the inventory item.

    Returns
    -------
    None
    """

    url = f"{self.base_path}/{id}"
    self.session.delete(url)

get_all_facets

get_all_facets(
    *,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory]
    | InventoryCategory
    | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation]
    | StorageLocation
    | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> list[FacetItem]

Get available facets for inventory items based on the provided filters.

Source code in src/albert/collections/inventory.py
@validate_call
def get_all_facets(
    self,
    *,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory] | InventoryCategory | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation] | StorageLocation | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> list[FacetItem]:
    """
    Get available facets for inventory items based on the provided filters.
    """

    params = self._prepare_parameters(
        limit=1,
        text=text,
        cas=cas,
        category=category,
        company=company,
        location=location,
        storage_location=storage_location,
        project_id=project_id,
        sheet_id=sheet_id,
        created_by=created_by,
        lot_owner=lot_owner,
        tags=tags,
    )
    response = self.session.get(
        url=f"{self.base_path}/llmsearch"
        if match_all_conditions
        else f"{self.base_path}/search",
        params=params,
    )
    return [FacetItem.model_validate(x) for x in response.json()["Facets"]]

get_by_id

get_by_id(*, id: InventoryId) -> InventoryItem

Retrieve an inventory item by its ID.

Parameters:

Name Type Description Default
id InventoryId

The ID of the inventory item.

required

Returns:

Type Description
InventoryItem

The retrieved inventory item.

Source code in src/albert/collections/inventory.py
@validate_call
def get_by_id(self, *, id: InventoryId) -> InventoryItem:
    """
    Retrieve an inventory item by its ID.

    Parameters
    ----------
    id : InventoryId
        The ID of the inventory item.

    Returns
    -------
    InventoryItem
        The retrieved inventory item.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return InventoryItem(**response.json())

get_by_ids

get_by_ids(
    *, ids: list[InventoryId]
) -> list[InventoryItem]

Retrieve an set of inventory items by their IDs.

Parameters:

Name Type Description Default
ids list[InventoryId]

The list of IDs of the inventory items.

required

Returns:

Type Description
list[InventoryItem]

The retrieved inventory items.

Source code in src/albert/collections/inventory.py
@validate_call
def get_by_ids(self, *, ids: list[InventoryId]) -> list[InventoryItem]:
    """
    Retrieve an set of inventory items by their IDs.

    Parameters
    ----------
    ids : list[InventoryId]
        The list of IDs of the inventory items.

    Returns
    -------
    list[InventoryItem]
        The retrieved inventory items.
    """
    batch_size = 250
    batches = [ids[i : i + batch_size] for i in range(0, len(ids), batch_size)]
    inventory = []
    for batch in batches:
        response = self.session.get(f"{self.base_path}/ids", params={"id": batch})
        inventory.extend([InventoryItem(**item) for item in response.json()["Items"]])
    return inventory

get_facet_by_name

get_facet_by_name(
    name: str | list[str],
    *,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory]
    | InventoryCategory
    | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation]
    | StorageLocation
    | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> list[FacetItem]

Returns a specific facet by its name with all the filters applied to the search. This can be used for example to fetch all remaining tags as part of an iterative refinement of a search.

Source code in src/albert/collections/inventory.py
@validate_call
def get_facet_by_name(
    self,
    name: str | list[str],
    *,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory] | InventoryCategory | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation] | StorageLocation | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> list[FacetItem]:
    """
    Returns a specific facet by its name with all the filters applied to the search.
    This can be used for example to fetch all remaining tags as part of an iterative
    refinement of a search.
    """
    if isinstance(name, str):
        name = [name]

    facets = self.get_all_facets(
        text=text,
        cas=cas,
        category=category,
        company=company,
        location=location,
        storage_location=storage_location,
        project_id=project_id,
        sheet_id=sheet_id,
        created_by=created_by,
        lot_owner=lot_owner,
        tags=tags,
        match_all_conditions=match_all_conditions,
    )
    filtered_facets = []
    for facet in facets:
        if facet.name in name or facet.name.lower() in name:
            filtered_facets.append(facet)

    return filtered_facets

get_match_or_none

get_match_or_none(
    *, inventory_item: InventoryItem
) -> InventoryItem | None

Get a matching inventory item or return None if not found.

Parameters:

Name Type Description Default
inventory_item InventoryItem

The inventory item to match.

required

Returns:

Type Description
Union[InventoryItem, None]

The matching inventory item or None if not found.

Source code in src/albert/collections/inventory.py
def get_match_or_none(self, *, inventory_item: InventoryItem) -> InventoryItem | None:
    """
    Get a matching inventory item or return None if not found.

    Parameters
    ----------
    inventory_item : InventoryItem
        The inventory item to match.

    Returns
    -------
    Union[InventoryItem, None]
        The matching inventory item or None if not found.
    """
    hits = self.list(text=inventory_item.name, company=[inventory_item.company])
    inv_company = (
        inventory_item.company.name
        if isinstance(inventory_item.company, Company)
        else inventory_item.company
    )
    for inv in hits:
        if inv and inv.name == inventory_item.name and inv.company.name == inv_company:
            return inv
    else:
        return None

get_specs

get_specs(
    *, ids: list[InventoryId]
) -> list[InventorySpecList]

Get the specs for a list of inventory items.

Parameters:

Name Type Description Default
ids list[InventoryId]

List of Inventory IDs to get the specs for.

required

Returns:

Type Description
list[InventorySpecList]

A list of InventorySpecList objects, each containing the specs for an inventory item.

Source code in src/albert/collections/inventory.py
@validate_call
def get_specs(self, *, ids: list[InventoryId]) -> list[InventorySpecList]:
    """Get the specs for a list of inventory items.

    Parameters
    ----------
    ids : list[InventoryId]
        List of Inventory IDs to get the specs for.

    Returns
    -------
    list[InventorySpecList]
        A list of InventorySpecList objects, each containing the specs for an inventory item.
    """
    url = f"{self.base_path}/specs"
    batches = [ids[i : i + 250] for i in range(0, len(ids), 250)]
    ta = TypeAdapter(InventorySpecList)
    return [
        ta.validate_python(item)
        for batch in batches
        for item in self.session.get(url, params={"id": batch}).json()
    ]

inventory_exists

inventory_exists(*, inventory_item: InventoryItem) -> bool

Check if an inventory item exists.

Parameters:

Name Type Description Default
inventory_item InventoryItem

The inventory item to check.

required

Returns:

Type Description
bool

True if the inventory item exists, False otherwise.

Source code in src/albert/collections/inventory.py
def inventory_exists(self, *, inventory_item: InventoryItem) -> bool:
    """
    Check if an inventory item exists.

    Parameters
    ----------
    inventory_item : InventoryItem
        The inventory item to check.

    Returns
    -------
    bool
        True if the inventory item exists, False otherwise.
    """
    hit = self.get_match_or_none(inventory_item=inventory_item)
    return bool(hit)

list

list(
    *,
    limit: int = 100,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory]
    | InventoryCategory
    | None = None,
    company: list[Company] | Company | None = None,
    order: OrderBy = DESCENDING,
    sort_by: str | None = "createdAt",
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation]
    | StorageLocation
    | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> Iterator[InventoryItem]

List inventory items with optional filters.

Parameters:

Name Type Description Default
limit int

Maximum number of items to return (default is 100)

100
text str

Text to search for in inventory names and descriptions

None
cas list[Cas] | Cas | None

Filter by CAS number(s)

None
category list[InventoryCategory] | InventoryCategory | None

Filter by inventory category/categories

None
company list[Company] | Company | None

Filter by manufacturing company/companies

None
order OrderBy

Sort order, either ASCENDING or DESCENDING (default is DESCENDING)

DESCENDING
sort_by str

Field to sort by (default is "createdAt")

'createdAt'
location list[Location] | None

Filter by location(s)

None
storage_location list[StorageLocation] | None

Filter by storage location(s)

None
project_id str

Filter by project ID

None
sheet_id str

Filter by sheet ID

None
created_by list[User]

Filter by creator(s)

None
lot_owner list[User]

Filter by lot owner(s)

None
tags list[str]

Filter by tag(s)

None
match_all_conditions bool

Whether to match all conditions (default is False -- e.g. OR between conditions)

False

Returns:

Type Description
Iterator[InventoryItem]

An iterator over the matching inventory items

Source code in src/albert/collections/inventory.py
@validate_call
def list(
    self,
    *,
    limit: int = 100,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory] | InventoryCategory | None = None,
    company: list[Company] | Company | None = None,
    order: OrderBy = OrderBy.DESCENDING,
    sort_by: str | None = "createdAt",
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation] | StorageLocation | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> Iterator[InventoryItem]:
    """
    List inventory items with optional filters.

    Parameters
    ----------
    limit : int, optional
        Maximum number of items to return (default is 100)
    text : str, optional
        Text to search for in inventory names and descriptions
    cas : list[Cas] | Cas | None, optional
        Filter by CAS number(s)
    category : list[InventoryCategory] | InventoryCategory | None, optional
        Filter by inventory category/categories
    company : list[Company] | Company | None, optional
        Filter by manufacturing company/companies
    order : OrderBy, optional
        Sort order, either ASCENDING or DESCENDING (default is DESCENDING)
    sort_by : str, optional
        Field to sort by (default is "createdAt")
    location : list[Location] | None, optional
        Filter by location(s)
    storage_location : list[StorageLocation] | None, optional
        Filter by storage location(s)
    project_id : str, optional
        Filter by project ID
    sheet_id : str, optional
        Filter by sheet ID
    created_by : list[User], optional
        Filter by creator(s)
    lot_owner : list[User], optional
        Filter by lot owner(s)
    tags : list[str], optional
        Filter by tag(s)
    match_all_conditions : bool, optional
        Whether to match all conditions (default is False -- e.g. OR between conditions)

    Returns
    -------
    Iterator[InventoryItem]
        An iterator over the matching inventory items
    """

    def deserialize(items: list[dict]) -> list[InventoryItem]:
        return self.get_by_ids(ids=[x["albertId"] for x in items])

    search_text = text if (text is None or len(text) < 50) else text[0:50]
    params = self._prepare_parameters(
        limit=limit,
        text=search_text,
        cas=cas,
        category=category,
        company=company,
        order=order,
        sort_by=sort_by,
        location=location,
        storage_location=storage_location,
        project_id=project_id,
        sheet_id=sheet_id,
        created_by=created_by,
        lot_owner=lot_owner,
        tags=tags,
    )
    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/llmsearch"
        if match_all_conditions
        else f"{self.base_path}/search",
        params=params,
        session=self.session,
        deserialize=deserialize,
    )

merge

merge(
    *,
    parent_id: InventoryId,
    child_id: InventoryId | list[InventoryId],
    modules: list[str] | None = None,
) -> None

Merge one or multiple child inventory into a parent inventory item.

Parameters:

Name Type Description Default
parent_id InventoryId

The ID of the parent inventory item.

required
child_id InventoryId | list[InventoryId]

The ID(s) of the child inventory item(s).

required
modules list[str]

The merge modules to use (default is all).

None

Returns:

Type Description
None
Source code in src/albert/collections/inventory.py
def merge(
    self,
    *,
    parent_id: InventoryId,
    child_id: InventoryId | list[InventoryId],
    modules: list[str] | None = None,
) -> None:
    """
    Merge one or multiple child inventory into a parent inventory item.

    Parameters
    ----------
    parent_id : InventoryId
        The ID of the parent inventory item.
    child_id : InventoryId | list[InventoryId]
        The ID(s) of the child inventory item(s).
    modules : list[str], optional
        The merge modules to use (default is all).

    Returns
    -------
    None
    """

    # assume "all" modules if not specified explicitly
    modules = modules if modules is not None else ALL_MERGE_MODULES

    # define merge endpoint
    url = f"{self.base_path}/merge"

    if isinstance(child_id, list):
        child_inventories = [{"id": i} for i in child_id]
    else:
        child_inventories = [{"id": child_id}]

    # define payload using the class
    payload = MergeInventory(
        parent_id=parent_id,
        child_inventories=child_inventories,
        modules=modules,
    )

    # post request
    self.session.post(url, json=payload.model_dump(mode="json", by_alias=True))

search

search(
    *,
    limit: int = 100,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory]
    | InventoryCategory
    | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation]
    | StorageLocation
    | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> Iterator[InventorySearchItem]

Get a list of inventory items that match the search criteria and return the raw search records. These are not full inventory item objects, but are special short documents intended for fast summary results

Source code in src/albert/collections/inventory.py
@validate_call
def search(
    self,
    *,
    limit: int = 100,
    text: str | None = None,
    cas: list[Cas] | Cas | None = None,
    category: list[InventoryCategory] | InventoryCategory | None = None,
    company: list[Company] | Company | None = None,
    location: list[Location] | Location | None = None,
    storage_location: list[StorageLocation] | StorageLocation | None = None,
    project_id: ProjectId | None = None,
    sheet_id: WorksheetId | None = None,
    created_by: list[User] | User | None = None,
    lot_owner: list[User] | User | None = None,
    tags: list[str] | None = None,
    match_all_conditions: bool = False,
) -> Iterator[InventorySearchItem]:
    """
    Get a list of inventory items that match the search criteria and
    return the raw search records. These are not full inventory item
    objects, but are special short documents intended for fast summary results
    """

    def deserialize(items: list[dict]):
        return [InventorySearchItem.model_validate(x) for x in items]

    params = self._prepare_parameters(
        limit=limit,
        text=text,
        cas=cas,
        category=category,
        company=company,
        location=location,
        storage_location=storage_location,
        project_id=project_id,
        sheet_id=sheet_id,
        created_by=created_by,
        lot_owner=lot_owner,
        tags=tags,
    )
    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/llmsearch"
        if match_all_conditions
        else f"{self.base_path}/search",
        params=params,
        session=self.session,
        deserialize=deserialize,
    )

update

update(*, inventory_item: InventoryItem) -> InventoryItem

Update an inventory item.

Parameters:

Name Type Description Default
inventory_item InventoryItem

The updated inventory item object.

required

Returns:

Type Description
InventoryItem

The updated inventory item retrieved from the server.

Source code in src/albert/collections/inventory.py
def update(self, *, inventory_item: InventoryItem) -> InventoryItem:
    """
    Update an inventory item.

    Parameters
    ----------
    inventory_item : InventoryItem
        The updated inventory item object.

    Returns
    -------
    InventoryItem
        The updated inventory item retrieved from the server.
    """
    # Fetch the current object state from the server or database
    current_object = self.get_by_id(id=inventory_item.id)

    # Generate the PATCH payload
    patch_payload = self._generate_inventory_patch_payload(
        existing=current_object, updated=inventory_item
    )

    # Complex patching is not working, so I'm going to do this in a loop :(
    # https://teams.microsoft.com/l/message/19:de4a48c366664ce1bafcdbea02298810@thread.tacv2/1724856117312?tenantId=98aab90e-764b-48f1-afaa-02e3c7300653&groupId=35a36a3d-fc25-4899-a1dd-ad9c7d77b5b3&parentMessageId=1724856117312&teamName=Product%20%2B%20Engineering&channelName=General%20-%20API&createdTime=1724856117312
    url = f"{self.base_path}/{inventory_item.id}"
    for change in patch_payload["data"]:
        change_payload = {"data": [change]}
        self.session.patch(url, json=change_payload)
    updated_inv = self.get_by_id(id=inventory_item.id)
    return updated_inv

LinksCollection

LinksCollection(*, session: AlbertSession)

Bases: BaseCollection

LinksCollection is a collection class for managing Link entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Creates a new link entity.

delete

Deletes a link entity by its ID.

get_by_id

Retrieves a link entity by its ID.

list

Generates a list of link entities with optional filters.

Source code in src/albert/collections/links.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the LinksCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, links: list[Link]) -> list[Link]

Creates a new link entity.

Parameters:

Name Type Description Default
links list[Link]

List of Link entities to create.

required

Returns:

Type Description
Link

The created link entity.

Source code in src/albert/collections/links.py
def create(self, *, links: list[Link]) -> list[Link]:
    """
    Creates a new link entity.

    Parameters
    ----------
    links : list[Link]
        List of Link entities to create.

    Returns
    -------
    Link
        The created link entity.
    """
    response = self.session.post(
        self.base_path,
        json=[l.model_dump(by_alias=True, exclude_none=True, mode="json") for l in links],
    )
    return [Link(**l) for l in response.json()]

delete

delete(*, id: str) -> None

Deletes a link entity by its ID.

Parameters:

Name Type Description Default
id str

The ID of the link entity to delete.

required
Source code in src/albert/collections/links.py
def delete(self, *, id: str) -> None:
    """
    Deletes a link entity by its ID.

    Parameters
    ----------
    id : str
        The ID of the link entity to delete.
    """
    path = f"{self.base_path}/{id}"
    self.session.delete(path)

get_by_id

get_by_id(*, id: str) -> Link

Retrieves a link entity by its ID.

Parameters:

Name Type Description Default
id str

The ID of the link entity to retrieve.

required

Returns:

Type Description
Link

The retrieved link entity.

Source code in src/albert/collections/links.py
def get_by_id(self, *, id: str) -> Link:
    """
    Retrieves a link entity by its ID.

    Parameters
    ----------
    id : str
        The ID of the link entity to retrieve.

    Returns
    -------
    Link
        The retrieved link entity.
    """
    path = f"{self.base_path}/{id}"
    response = self.session.get(path)
    return Link(**response.json())

list

list(
    *,
    limit: int = 100,
    type: str | None = None,
    category: LinkCategory | None = None,
    id: str | None = None,
) -> Iterator[Link]

Generates a list of link entities with optional filters.

Parameters:

Name Type Description Default
limit int

The maximum number of link entities to return.

100
type str

The type of the link entities to return. Allowed values are parent, child, and all. If type is "all" then it will fetch both parent/child record for mentioned id.

None
category LinkCategory

The category of the link entities to return. Allowed values are mention, linkedTask, and synthesis.

None
id str

The ID of the link entity to return. (Use with type parameter)

None

Returns:

Type Description
Iterator[Link]

An iterator of Links.

Source code in src/albert/collections/links.py
def list(
    self,
    *,
    limit: int = 100,
    type: str | None = None,
    category: LinkCategory | None = None,
    id: str | None = None,
) -> Iterator[Link]:
    """
    Generates a list of link entities with optional filters.

    Parameters
    ----------
    limit : int, optional
        The maximum number of link entities to return.
    type : str, optional
        The type of the link entities to return. Allowed values are `parent`, `child`, and `all`. If type is "all" then it will fetch both parent/child record for mentioned id.
    category : LinkCategory, optional
        The category of the link entities to return. Allowed values are `mention`, `linkedTask`, and `synthesis`.
    id : str
        The ID of the link entity to return. (Use with `type` parameter)

    Returns
    ------
    Iterator[Link]
        An iterator of Links.
    """
    params = {"limit": limit, "type": type, "category": category, "id": id}
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        params=params,
        session=self.session,
        deserialize=lambda items: [Link(**item) for item in items],
    )

ListsCollection

ListsCollection(*, session: AlbertSession)

Bases: BaseCollection

ListsCollection is a collection class for managing ListItem entities in the Albert platform.

Example
stages = [
    "1. Discovery",
    "2. Concept Validation",
    "3. Proof of Concept",
    "4. Prototype Development",
    "5. Preliminary Evaluation",
    "6. Feasibility Study",
    "7. Optimization",
    "8. Scale-Up",
    "9. Regulatory Assessment",
]
# Initialize the Albert client
client = Albert()

# Get the custom field this list is associated with
stage_gate_field = client.custom_fields.get_by_id(id="CF123")

# Create the list items
for s in stages:
    item = ListItem(
        name=s,
        category=stage_gate_field.category,
        list_type=stage_gate_field.name,
    )

    client.lists.create(list_item=item)

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Creates a list entity.

delete

Delete a lists entry item by its ID.

get_by_id

Retrieves a list entity by its ID.

get_matching_item

Get a list item by name and list type.

list

Generates a list of list entities with optional filters.

update

Update a list item.

Source code in src/albert/collections/lists.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the TagCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, list_item: ListItem) -> ListItem

Creates a list entity.

Parameters:

Name Type Description Default
list_item ListItem

The list entity to create.

required

Returns:

Type Description
List

The created list entity.

Source code in src/albert/collections/lists.py
def create(self, *, list_item: ListItem) -> ListItem:
    """
    Creates a list entity.

    Parameters
    ----------
    list_item : ListItem
        The list entity to create.

    Returns
    -------
    List
        The created list entity.
    """
    response = self.session.post(
        self.base_path,
        json=list_item.model_dump(by_alias=True, exclude_none=True, mode="json"),
    )
    return ListItem(**response.json())

delete

delete(*, id: str) -> None

Delete a lists entry item by its ID.

Parameters:

Name Type Description Default
id str

The ID of the lists item.

required

Returns:

Type Description
None
Source code in src/albert/collections/lists.py
def delete(self, *, id: str) -> None:
    """
    Delete a lists entry item by its ID.

    Parameters
    ----------
    id : str
        The ID of the lists item.

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

get_by_id

get_by_id(*, id: str) -> ListItem

Retrieves a list entity by its ID.

Parameters:

Name Type Description Default
id str

The ID of the list entity to retrieve.

required

Returns:

Type Description
List

A list entity.

Source code in src/albert/collections/lists.py
def get_by_id(self, *, id: str) -> ListItem:
    """
    Retrieves a list entity by its ID.

    Parameters
    ----------
    id : str
        The ID of the list entity to retrieve.

    Returns
    -------
    List
        A list entity.
    """
    response = self.session.get(f"{self.base_path}/{id}")
    return ListItem(**response.json())

get_matching_item

get_matching_item(
    *, name: str, list_type: str
) -> ListItem | None

Get a list item by name and list type.

Parameters:

Name Type Description Default
name str

The name of it item to retrieve.

required
list_type str

The type of list (can be the name of the custom field)

required

Returns:

Type Description
ListItem | None

A list item with the provided name and list type, or None if not found.

Source code in src/albert/collections/lists.py
def get_matching_item(self, *, name: str, list_type: str) -> ListItem | None:
    """Get a list item by name and list type.

    Parameters
    ----------
    name : str
        The name of it item to retrieve.
    list_type : str
        The type of list (can be the name of the custom field)

    Returns
    -------
    ListItem | None
        A list item with the provided name and list type, or None if not found.
    """
    for list_item in self.list(names=[name], list_type=list_type):
        if list_item.name.lower() == name.lower():
            return list_item
    return None

list

list(
    *,
    limit: int = 100,
    names: list[str] | None = None,
    category: ListItemCategory | None = None,
    list_type: str | None = None,
    start_key: str | None = None,
) -> Iterator[ListItem]

Generates a list of list entities with optional filters.

Parameters:

Name Type Description Default
limit int

The maximum number of list entities to return.

100
names list[str]

A list of names of the list entity to retrieve.

None
category ListItemCategory

The category of the list entity to retrieve.

None
list_type str

The type of list entity to retrieve.

None

Returns:

Type Description
Iterator[ListItem]

An iterator of ListItems.

Source code in src/albert/collections/lists.py
def list(
    self,
    *,
    limit: int = 100,
    names: list[str] | None = None,
    category: ListItemCategory | None = None,
    list_type: str | None = None,
    start_key: str | None = None,
) -> Iterator[ListItem]:
    """
    Generates a list of list entities with optional filters.

    Parameters
    ----------
    limit : int, optional
        The maximum number of list entities to return.
    names : list[str], optional
        A list of names of the list entity to retrieve.
    category : ListItemCategory, optional
        The category of the list entity to retrieve.
    list_type : str, optional
        The type of list entity to retrieve.
    Returns
    ------
    Iterator[ListItem]
        An iterator of ListItems.
    """
    params = {
        "limit": limit,
        "startKey": start_key,
        "name": [names] if isinstance(names, str) else names,
        "category": category.value if isinstance(category, ListItemCategory) else category,
        "listType": list_type,
    }
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [ListItem(**item) for item in items],
    )

update

update(*, list_item=ListItem) -> ListItem

Update a list item.

Parameters:

Name Type Description Default
list_item ListItem

The list item to update.

ListItem

Returns:

Type Description
ListItem

The updated list item.

Source code in src/albert/collections/lists.py
def update(self, *, list_item=ListItem) -> ListItem:
    """Update a list item.

    Parameters
    ----------
    list_item : ListItem
        The list item to update.

    Returns
    -------
    ListItem
        The updated list item.
    """
    existing = self.get_by_id(id=list_item.id)
    patches = self._generate_patch_payload(
        existing=existing, updated=list_item, generate_metadata_diff=False
    )
    if len(patches.data) == 0:
        return existing
    self.session.patch(
        url=f"{self.base_path}/{list_item.id}",
        json=patches.model_dump(mode="json", by_alias=True, exclude_none=True),
    )
    return self.get_by_id(id=list_item.id)

LocationCollection

LocationCollection(*, session: AlbertSession)

Bases: BaseCollection

LocationCollection is a collection class for managing Location entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Creates a new Location entity.

delete

Deletes a Location entity.

get_by_id

Retrieves a location by its ID.

list

Searches for locations matching the provided criteria.

location_exists

Determines if a location, with the same name, exists in the collection.

update

Update a Location entity.

Source code in src/albert/collections/locations.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the LocationCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, location: Location) -> Location

Creates a new Location entity.

Parameters:

Name Type Description Default
location Location

The Location object to create.

required

Returns:

Type Description
Location

The created Location object.

Source code in src/albert/collections/locations.py
def create(self, *, location: Location) -> Location:
    """
    Creates a new Location entity.

    Parameters
    ----------
    location : Location
        The Location object to create.

    Returns
    -------
    Location
        The created Location object.
    """
    exists = self.location_exists(location=location)
    if exists:
        logging.warning(
            f"Location with name {location.name} matches an existing location. Returning the existing Location."
        )
        return exists

    payload = location.model_dump(by_alias=True, exclude_unset=True, mode="json")
    response = self.session.post(self.base_path, json=payload)

    return Location(**response.json())

delete

delete(*, id: str) -> None

Deletes a Location entity.

Parameters:

Name Type Description Default
id Str

The id of the Location object to delete.

required

Returns:

Type Description
None
Source code in src/albert/collections/locations.py
def delete(self, *, id: str) -> None:
    """
    Deletes a Location entity.

    Parameters
    ----------
    id : Str
        The id of the Location object to delete.

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

get_by_id

get_by_id(*, id: str) -> Location

Retrieves a location by its ID.

Parameters:

Name Type Description Default
id str

The ID of the location to retrieve.

required

Returns:

Type Description
Location

The Location object.

Source code in src/albert/collections/locations.py
def get_by_id(self, *, id: str) -> Location:
    """
    Retrieves a location by its ID.

    Parameters
    ----------
    id : str
        The ID of the location to retrieve.

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

list

list(
    *,
    ids: list[str] | None = None,
    name: str | list[str] | None = None,
    country: str | None = None,
    exact_match: bool = False,
    limit: int = 50,
    start_key: str | None = None,
) -> Iterator[Location]

Searches for locations matching the provided criteria.

Parameters:

Name Type Description Default
ids list[str] | None

The list of IDs to filter the locations, by default None. Max length is 100.

None
name str | list[str] | None

The name or names of locations to search for, by default None

None
country str | None

The country code of the country to filter the locations , by default None

None
exact_match bool

Whether to return exact matches only, by default False

False

Yields:

Type Description
Iterator[Location]

An iterator of Location objects matching the search criteria.

Source code in src/albert/collections/locations.py
def list(
    self,
    *,
    ids: list[str] | None = None,
    name: str | list[str] | None = None,
    country: str | None = None,
    exact_match: bool = False,
    limit: int = 50,
    start_key: str | None = None,
) -> Iterator[Location]:
    """Searches for locations matching the provided criteria.

    Parameters
    ----------
    ids: list[str] | None, optional
        The list of IDs to filter the locations, by default None.
        Max length is 100.
    name : str | list[str] | None, optional
        The name or names of locations to search for, by default None
    country : str | None, optional
        The country code of the country to filter the locations , by default None
    exact_match : bool, optional
        Whether to return exact matches only, by default False

    Yields
    ------
    Iterator[Location]
        An iterator of Location objects matching the search criteria.
    """
    params = {"limit": limit, "startKey": start_key, "country": country}
    if ids:
        params["id"] = ids
    if name:
        params["name"] = [name] if isinstance(name, str) else name
        params["exactMatch"] = json.dumps(exact_match)
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [Location(**item) for item in items],
    )

location_exists

location_exists(*, location: Location) -> Location | None

Determines if a location, with the same name, exists in the collection.

Parameters:

Name Type Description Default
location Location

The Location object to check

required

Returns:

Type Description
Location | None

The existing registered Location object if found, otherwise None.

Source code in src/albert/collections/locations.py
def location_exists(self, *, location: Location) -> Location | None:
    """Determines if a location, with the same name, exists in the collection.

    Parameters
    ----------
    location : Location
        The Location object to check

    Returns
    -------
    Location | None
        The existing registered Location object if found, otherwise None.
    """
    hits = self.list(name=location.name)
    if hits:
        for hit in hits:
            if hit and hit.name.lower() == location.name.lower():
                return hit
    return None

update

update(*, location: Location) -> Location

Update a Location entity.

Parameters:

Name Type Description Default
location Location

The Location object to update. The ID of the Location object must be provided.

required

Returns:

Type Description
Location

The updated Location object as returned by the server.

Source code in src/albert/collections/locations.py
def update(self, *, location: Location) -> Location:
    """Update a Location entity.

    Parameters
    ----------
    location : Location
        The Location object to update. The ID of the Location object must be provided.

    Returns
    -------
    Location
        The updated Location object as returned by the server.
    """
    # Fetch the current object state from the server or database
    current_object = self.get_by_id(id=location.id)
    # Generate the PATCH payload
    patch_payload = self._generate_patch_payload(
        existing=current_object,
        updated=location,
        stringify_values=True,
    )
    url = f"{self.base_path}/{location.id}"
    self.session.patch(url, json=patch_payload.model_dump(mode="json", by_alias=True))
    return self.get_by_id(id=location.id)

LotCollection

LotCollection(*, session: AlbertSession)

Bases: BaseCollection

LotCollection is a collection class for managing Lot entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

An Albert session instance.

required

Methods:

Name Description
create
delete

Delete a lot by its ID.

get_by_id

Get a lot by its ID.

get_by_ids

Get a list of lots by their IDs.

list

Lists Lot entities with optional filters.

update

Update a lot.

Source code in src/albert/collections/lots.py
def __init__(self, *, session: AlbertSession):
    """A collection for interacting with Lots in Albert.

    Parameters
    ----------
    session : AlbertSession
        An Albert session instance.
    """
    super().__init__(session=session)
    self.base_path = f"/api/{LotCollection._api_version}/lots"

base_path instance-attribute

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

create

create(*, lots: list[Lot]) -> list[Lot]
Source code in src/albert/collections/lots.py
def create(self, *, lots: list[Lot]) -> list[Lot]:
    # TODO: Once thi endpoint is fixed, go back to passing the whole list at once
    payload = [lot.model_dump(by_alias=True, exclude_none=True, mode="json") for lot in lots]
    all_lots = []
    for lot in payload:
        response = self.session.post(self.base_path, json=[lot])
        all_lots.append(Lot(**response.json()[0]))
    # response = self.session.post(self.base_path, json=payload)
    # return [Lot(**lot) for lot in response.json().get("CreatedLots", [])]
    return all_lots

delete

delete(*, id: str) -> None

Delete a lot by its ID.

Parameters:

Name Type Description Default
id str

The ID of the lot to delete.

required
Source code in src/albert/collections/lots.py
def delete(self, *, id: str) -> None:
    """Delete a lot by its ID.

    Parameters
    ----------
    id : str
        The ID of the lot to delete.
    """
    url = f"{self.base_path}?id={id}"
    self.session.delete(url)

get_by_id

get_by_id(*, id: str) -> Lot

Get a lot by its ID.

Parameters:

Name Type Description Default
id str

The ID of the lot to get.

required

Returns:

Type Description
Lot

The lot with the provided ID.

Source code in src/albert/collections/lots.py
def get_by_id(self, *, id: str) -> Lot:
    """Get a lot by its ID.

    Parameters
    ----------
    id : str
        The ID of the lot to get.

    Returns
    -------
    Lot
        The lot with the provided ID.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return Lot(**response.json())

get_by_ids

get_by_ids(*, ids: list[str]) -> list[Lot]

Get a list of lots by their IDs.

Parameters:

Name Type Description Default
ids list[str]

A list of lot IDs to get.

required

Returns:

Type Description
list[Lot]

A list of lots with the provided IDs.

Source code in src/albert/collections/lots.py
def get_by_ids(self, *, ids: list[str]) -> list[Lot]:
    """Get a list of lots by their IDs.

    Parameters
    ----------
    ids : list[str]
        A list of lot IDs to get.

    Returns
    -------
    list[Lot]
        A list of lots with the provided IDs.
    """
    url = f"{self.base_path}/ids"
    response = self.session.get(url, params={"id": ids})
    return [Lot(**lot) for lot in response.json()["Items"]]

list

list(
    *,
    limit: int = 100,
    start_key: str | None = None,
    parent_id: str | None = None,
    inventory_id: str | None = None,
    barcode_id: str | None = None,
    parent_id_category: str | None = None,
    inventory_on_hand: str | None = None,
    location_id: str | None = None,
    exact_match: bool = False,
    begins_with: bool = False,
) -> Iterator[Lot]

Lists Lot entities with optional filters.

Parameters:

Name Type Description Default
limit int

The maximum number of Lots to return, by default 100.

100
start_key Optional[str]

The primary key of the first item to evaluate for pagination.

None
parent_id Optional[str]

Fetches list of lots for a parentId (inventory).

None
inventory_id Optional[str]

Fetches list of lots for an inventory.

None
barcode_id Optional[str]

Fetches list of lots for a barcodeId.

None
parent_id_category Optional[str]

Fetches list of lots for a parentIdCategory (e.g., RawMaterials, Consumables).

None
inventory_on_hand Optional[str]

Fetches records based on inventoryOnHand (lteZero, gtZero, eqZero).

None
location_id Optional[str]

Fetches list of lots for a locationId.

None
exact_match bool

Determines if barcodeId field should be an exact match, by default False.

False
begins_with bool

Determines if barcodeId begins with a certain value, by default False.

False

Yields:

Type Description
Iterator[Lot]

An iterator of Lot objects.

Source code in src/albert/collections/lots.py
def list(
    self,
    *,
    limit: int = 100,
    start_key: str | None = None,
    parent_id: str | None = None,
    inventory_id: str | None = None,
    barcode_id: str | None = None,
    parent_id_category: str | None = None,
    inventory_on_hand: str | None = None,
    location_id: str | None = None,
    exact_match: bool = False,
    begins_with: bool = False,
) -> Iterator[Lot]:
    """
    Lists Lot entities with optional filters.

    Parameters
    ----------
    limit : int, optional
        The maximum number of Lots to return, by default 100.
    start_key : Optional[str], optional
        The primary key of the first item to evaluate for pagination.
    parent_id : Optional[str], optional
        Fetches list of lots for a parentId (inventory).
    inventory_id : Optional[str], optional
        Fetches list of lots for an inventory.
    barcode_id : Optional[str], optional
        Fetches list of lots for a barcodeId.
    parent_id_category : Optional[str], optional
        Fetches list of lots for a parentIdCategory (e.g., RawMaterials, Consumables).
    inventory_on_hand : Optional[str], optional
        Fetches records based on inventoryOnHand (lteZero, gtZero, eqZero).
    location_id : Optional[str], optional
        Fetches list of lots for a locationId.
    exact_match : bool, optional
        Determines if barcodeId field should be an exact match, by default False.
    begins_with : bool, optional
        Determines if barcodeId begins with a certain value, by default False.

    Yields
    -------
    Iterator[Lot]
        An iterator of Lot objects.
    """
    params = {
        "limit": limit,
        "startKey": start_key,
        "parentId": parent_id,
        "inventoryId": inventory_id,
        "barcodeId": barcode_id,
        "parentIdCategory": parent_id_category,
        "inventoryOnHand": inventory_on_hand,
        "locationId": location_id,
        "exactMatch": json.dumps(exact_match),
        "beginsWith": json.dumps(begins_with),
    }
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [Lot(**item) for item in items],
    )

update

update(*, lot: Lot) -> Lot

Update a lot.

Parameters:

Name Type Description Default
lot Lot

The updated lot object.

required

Returns:

Type Description
Lot

The updated lot object as returned by the server.

Source code in src/albert/collections/lots.py
def update(self, *, lot: Lot) -> Lot:
    """Update a lot.

    Parameters
    ----------
    lot : Lot
        The updated lot object.

    Returns
    -------
    Lot
        The updated lot object as returned by the server.
    """
    existing_lot = self.get_by_id(id=lot.id)
    patch_data = self._generate_patch_payload(existing=existing_lot, updated=lot)
    url = f"{self.base_path}/{lot.id}"

    self.session.patch(url, json=patch_data.model_dump(mode="json", by_alias=True))

    return self.get_by_id(id=lot.id)

NotebookCollection

NotebookCollection(*, session: AlbertSession)

Bases: BaseCollection

NotebookCollection is a collection class for managing Notebook entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
copy

Create a copy of a Notebook into a specified parent

create

Create or return notebook for the provided notebook.

delete

Deletes a notebook by its ID.

get_block_by_id

Retrieve a Notebook Block by its ID.

get_by_id

Retrieve a Notebook by its ID.

list_by_parent_id

Retrieve a Notebook by parent ID.

update

Update a notebook.

update_block_content

Updates the block content of a Notebook. This does not update the notebook name (use .update for that).

Source code in src/albert/collections/notebooks.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the NotebookCollection with the provided session.

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

base_path instance-attribute

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

copy

copy(
    *,
    notebook_copy_info: NotebookCopyInfo,
    type: NotebookCopyType,
) -> Notebook

Create a copy of a Notebook into a specified parent

Parameters:

Name Type Description Default
notebook_copy_info NotebookCopyInfo

The copy information for the Notebook copy

required
type NotebookCopyType

Differentiate whether copy is for templates, task, project or restoreTemplate

required

Returns:

Type Description
Notebook

The result of the copied Notebook.

Source code in src/albert/collections/notebooks.py
def copy(self, *, notebook_copy_info: NotebookCopyInfo, type: NotebookCopyType) -> Notebook:
    """Create a copy of a Notebook into a specified parent

    Parameters
    ----------
    notebook_copy_info : NotebookCopyInfo
        The copy information for the Notebook copy
    type : NotebookCopyType
        Differentiate whether copy is for templates, task, project or restoreTemplate

    Returns
    -------
    Notebook
        The result of the copied Notebook.
    """
    response = self.session.post(
        url=f"{self.base_path}/copy",
        json=notebook_copy_info.model_dump(mode="json", by_alias=True, exclude_none=True),
        params={"type": type, "parentId": notebook_copy_info.parent_id},
    )
    return Notebook(**response.json())

create

create(*, notebook: Notebook) -> Notebook

Create or return notebook for the provided notebook. This endpoint automatically tries to find an existing notebook with the same parameter setpoints, and will either return the existing notebook or create a new one.

Parameters:

Name Type Description Default
notebook Notebook

A list of Notebook objects to find or create.

required

Returns:

Type Description
Notebook

A list of created or found Notebook objects.

Source code in src/albert/collections/notebooks.py
def create(self, *, notebook: Notebook) -> Notebook:
    """Create or return notebook for the provided notebook.
    This endpoint automatically tries to find an existing notebook with the same parameter setpoints, and will either return the existing notebook or create a new one.

    Parameters
    ----------
    notebook : Notebook
        A list of Notebook objects to find or create.

    Returns
    -------
    Notebook
        A list of created or found Notebook objects.
    """
    if notebook.blocks:
        # This check keeps a user from corrupting the Notebook data.
        msg = (
            "Cannot create a Notebook with pre-filled blocks. "
            "Set `blocks=[]` (or do not set it) when creating it. "
            "Use `.update_block_content()` afterward to add, update, or delete blocks."
        )
        raise AlbertException(msg)
    response = self.session.post(
        url=self.base_path,
        json=notebook.model_dump(mode="json", by_alias=True, exclude_none=True),
        params={"parentId": notebook.parent_id},
    )
    return Notebook(**response.json())

delete

delete(*, id: str) -> None

Deletes a notebook by its ID.

Parameters:

Name Type Description Default
id str

The ID of the notebook to delete.

required
Source code in src/albert/collections/notebooks.py
def delete(self, *, id: str) -> None:
    """
    Deletes a notebook by its ID.

    Parameters
    ----------
    id : str
        The ID of the notebook to delete.
    """
    self.session.delete(f"{self.base_path}/{id}")

get_block_by_id

get_block_by_id(
    *, notebook_id: str, block_id: str
) -> NotebookBlock

Retrieve a Notebook Block by its ID.

Parameters:

Name Type Description Default
notebook_id str

The ID of the Notebook to which the Block belongs.

required
block_id str

The ID of the Notebook Block to retrieve.

required

Returns:

Type Description
NotebookBlock

The NotebookBlock object.

Source code in src/albert/collections/notebooks.py
def get_block_by_id(self, *, notebook_id: str, block_id: str) -> NotebookBlock:
    """Retrieve a Notebook Block by its ID.

    Parameters
    ----------
    notebook_id : str
        The ID of the Notebook to which the Block belongs.
    block_id : str
        The ID of the Notebook Block to retrieve.

    Returns
    -------
    NotebookBlock
        The NotebookBlock object.
    """
    response = self.session.get(f"{self.base_path}/{notebook_id}/blocks/{block_id}")
    return TypeAdapter(NotebookBlock).validate_python(response.json())

get_by_id

get_by_id(*, id: str) -> Notebook

Retrieve a Notebook by its ID.

Parameters:

Name Type Description Default
id str

The ID of the Notebook to retrieve.

required

Returns:

Type Description
Notebook

The Notebook object.

Source code in src/albert/collections/notebooks.py
def get_by_id(self, *, id: str) -> Notebook:
    """Retrieve a Notebook by its ID.

    Parameters
    ----------
    id : str
        The ID of the Notebook to retrieve.

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

list_by_parent_id

list_by_parent_id(*, parent_id: str) -> list[Notebook]

Retrieve a Notebook by parent ID.

Parameters:

Name Type Description Default
parent_id str

The ID of the parent ID, e.g. task.

required

Returns:

Type Description
list[Notebook]

list of notebook references.

Source code in src/albert/collections/notebooks.py
def list_by_parent_id(self, *, parent_id: str) -> list[Notebook]:
    """Retrieve a Notebook by parent ID.

    Parameters
    ----------
    parent_id : str
        The ID of the parent ID, e.g. task.

    Returns
    -------
    list[Notebook]
        list of notebook references.

    """

    # search
    response = self.session.get(f"{self.base_path}/{parent_id}/search")
    # return
    return [self.get_by_id(id=x["id"]) for x in response.json()["Items"]]

update

update(*, notebook: Notebook) -> Notebook

Update a notebook.

Parameters:

Name Type Description Default
notebook Notebook

The updated notebook object.

required

Returns:

Type Description
Notebook

The updated notebook object as returned by the server.

Source code in src/albert/collections/notebooks.py
def update(self, *, notebook: Notebook) -> Notebook:
    """Update a notebook.

    Parameters
    ----------
    notebook : Notebook
        The updated notebook object.

    Returns
    -------
    Notebook
        The updated notebook object as returned by the server.
    """
    existing_notebook = self.get_by_id(id=notebook.id)
    patch_data = self._generate_patch_payload(existing=existing_notebook, updated=notebook)
    url = f"{self.base_path}/{notebook.id}"

    self.session.patch(url, json=patch_data.model_dump(mode="json", by_alias=True))

    return self.get_by_id(id=notebook.id)

update_block_content

update_block_content(*, notebook: Notebook) -> Notebook

Updates the block content of a Notebook. This does not update the notebook name (use .update for that). If a block in the Notebook does not already exist on Albert, it will be created. Note: The order of the Blocks in your Notebook matter and will be used in the updated Notebook!

Parameters:

Name Type Description Default
notebook Notebook

The updated notebook object.

required

Returns:

Type Description
Notebook

The updated notebook object as returned by the server.

Source code in src/albert/collections/notebooks.py
def update_block_content(self, *, notebook: Notebook) -> Notebook:
    """
    Updates the block content of a Notebook. This does not update the notebook name (use .update for that).
    If a block in the Notebook does not already exist on Albert, it will be created.
    *Note: The order of the Blocks in your Notebook matter and will be used in the updated Notebook!*


    Parameters
    ----------
    notebook : Notebook
        The updated notebook object.

    Returns
    -------
    Notebook
        The updated notebook object as returned by the server.
    """
    put_data = self._generate_put_block_payload(notebook=notebook)
    url = f"{self.base_path}/{notebook.id}/content"

    self.session.put(url, json=put_data.model_dump(mode="json", by_alias=True))

    return self.get_by_id(id=notebook.id)

NotesCollection

NotesCollection(*, session: AlbertSession)

Bases: BaseCollection

NotesCollection is a collection class for managing Note entities in the Albert platform.

Methods:

Name Description
create

Creates a new note.

delete

Deletes a note by its ID.

get_by_id

Retrieves a note by its ID.

list

Lists notes by their parent ID.

update

Updates a note.

Source code in src/albert/collections/notes.py
def __init__(self, *, session: AlbertSession):
    super().__init__(session=session)
    self.base_path = f"/api/{NotesCollection._api_version}/notes"

base_path instance-attribute

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

create

create(*, note: Note) -> Note

Creates a new note.

Parameters:

Name Type Description Default
note str

The note content.

required

Returns:

Type Description
Note

The created note.

Source code in src/albert/collections/notes.py
def create(self, *, note: Note) -> Note:
    """
    Creates a new note.

    Parameters
    ----------
    note : str
        The note content.

    Returns
    -------
    Note
        The created note.
    """
    response = self.session.post(
        self.base_path, json=note.model_dump(by_alias=True, exclude_unset=True, mode="json")
    )
    return Note(**response.json())

delete

delete(*, id: str) -> None

Deletes a note by its ID.

Parameters:

Name Type Description Default
id str

The ID of the note to delete.

required
Source code in src/albert/collections/notes.py
def delete(self, *, id: str) -> None:
    """
    Deletes a note by its ID.

    Parameters
    ----------
    id : str
        The ID of the note to delete.
    """
    self.session.delete(f"{self.base_path}/{id}")

get_by_id

get_by_id(*, id: str) -> Note

Retrieves a note by its ID.

Parameters:

Name Type Description Default
id str

The ID of the note to retrieve.

required

Returns:

Type Description
Note

The note if found, None otherwise.

Source code in src/albert/collections/notes.py
def get_by_id(self, *, id: str) -> Note:
    """
    Retrieves a note by its ID.

    Parameters
    ----------
    id : str
        The ID of the note to retrieve.

    Returns
    -------
    Note
        The note if found, None otherwise.
    """
    response = self.session.get(f"{self.base_path}/{id}")
    return Note(**response.json())

list

list(
    *, parent_id: str, order_by: OrderBy = DESCENDING
) -> list[Note]

Lists notes by their parent ID.

Parameters:

Name Type Description Default
parent_id str

The parent ID of the notes to list.

required
order_by OrderBy

The order to list notes in, by default OrderBy.DESCENDING.

DESCENDING

Returns:

Type Description
List[Note]

The list of notes.

Source code in src/albert/collections/notes.py
def list(self, *, parent_id: str, order_by: OrderBy = OrderBy.DESCENDING) -> list[Note]:
    """
    Lists notes by their parent ID.

    Parameters
    ----------
    parent_id : str
        The parent ID of the notes to list.
    order_by : OrderBy, optional
        The order to list notes in, by default OrderBy.DESCENDING.

    Returns
    -------
    List[Note]
        The list of notes.
    """

    params = {"parentId": parent_id, "orderBy": order_by.value}
    return AlbertPaginator(
        session=self.session,
        path=self.base_path,
        mode=PaginationMode.KEY,
        params=params,
        deserialize=lambda items: [Note(**item) for item in items],
    )

update

update(*, note: Note) -> Note

Updates a note.

Parameters:

Name Type Description Default
note Note

The note to update. The note must have an ID.

required

Returns:

Type Description
Note

The updated note as returned by the server.

Source code in src/albert/collections/notes.py
def update(self, *, note: Note) -> Note:
    """Updates a note.

    Parameters
    ----------
    note : Note
        The note to update. The note must have an ID.

    Returns
    -------
    Note
        The updated note as returned by the server.
    """
    patch = self._generate_patch_payload(
        existing=self.get_by_id(id=note.id), updated=note, generate_metadata_diff=False
    )
    self.session.patch(
        f"{self.base_path}/{note.id}",
        json=patch.model_dump(mode="json", by_alias=True, exclude_unset=True),
    )
    return self.get_by_id(id=note.id)

ParameterCollection

ParameterCollection(*, session: AlbertSession)

Bases: BaseCollection

ParameterCollection is a collection class for managing Parameter entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Create a new parameter.

delete

Delete a parameter by its ID.

get_by_id

Retrieve a parameter by its ID.

list

Lists parameters that match the provided criteria.

update

Update a parameter.

Source code in src/albert/collections/parameters.py
def __init__(self, *, session: AlbertSession):
    """Initializes the ParameterCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, parameter: Parameter) -> Parameter

Create a new parameter.

Parameters:

Name Type Description Default
parameter Parameter

The parameter to create.

required

Returns:

Type Description
Parameter

Returns the created parameter or the existing parameter if it already exists.

Source code in src/albert/collections/parameters.py
def create(self, *, parameter: Parameter) -> Parameter:
    """Create a new parameter.

    Parameters
    ----------
    parameter : Parameter
        The parameter to create.

    Returns
    -------
    Parameter
        Returns the created parameter or the existing parameter if it already exists.
    """
    match = next(self.list(names=parameter.name, exact_match=True), None)
    if match is not None:
        logging.warning(
            f"Parameter with name {parameter.name} already exists. Returning existing parameter."
        )
        return match
    response = self.session.post(
        self.base_path,
        json=parameter.model_dump(by_alias=True, exclude_none=True, mode="json"),
    )
    return Parameter(**response.json())

delete

delete(*, id: str) -> None

Delete a parameter by its ID.

Parameters:

Name Type Description Default
id str

The ID of the parameter to delete.

required
Source code in src/albert/collections/parameters.py
def delete(self, *, id: str) -> None:
    """Delete a parameter by its ID.

    Parameters
    ----------
    id : str
        The ID of the parameter to delete.
    """
    url = f"{self.base_path}/{id}"
    self.session.delete(url)

get_by_id

get_by_id(*, id: str) -> Parameter

Retrieve a parameter by its ID.

Parameters:

Name Type Description Default
id str

The ID of the parameter to retrieve.

required

Returns:

Type Description
Parameter

The parameter with the given ID.

Source code in src/albert/collections/parameters.py
def get_by_id(self, *, id: str) -> Parameter:
    """Retrieve a parameter by its ID.

    Parameters
    ----------
    id : str
        The ID of the parameter to retrieve.

    Returns
    -------
    Parameter
        The parameter with the given ID.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return Parameter(**response.json())

list

list(
    *,
    ids: list[str] | None = None,
    names: str | list[str] = None,
    exact_match: bool = False,
    order_by: OrderBy = DESCENDING,
    start_key: str | None = None,
    limit: int = 50,
    return_full: bool = True,
) -> Iterator[Parameter]

Lists parameters that match the provided criteria.

Parameters:

Name Type Description Default
ids list[str] | None

A list of parameter IDs to retrieve, by default None

None
names str | list[str]

A list of parameter names to retrieve, by default None

None
exact_match bool

Whether to match the name exactly, by default False

False
order_by OrderBy

The order in which to return results, by default OrderBy.DESCENDING

DESCENDING
return_full bool

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

True

Yields:

Type Description
Iterator[Parameter]

An iterator of Parameters matching the given criteria.

Source code in src/albert/collections/parameters.py
def list(
    self,
    *,
    ids: list[str] | None = None,
    names: str | list[str] = None,
    exact_match: bool = False,
    order_by: OrderBy = OrderBy.DESCENDING,
    start_key: str | None = None,
    limit: int = 50,
    return_full: bool = True,
) -> Iterator[Parameter]:
    """Lists parameters that match the provided criteria.

    Parameters
    ----------
    ids : list[str] | None, optional
        A list of parameter IDs to retrieve, by default None
    names : str | list[str], optional
        A list of parameter names to retrieve, by default None
    exact_match : bool, optional
        Whether to match the name exactly, by default False
    order_by : OrderBy, optional
        The order in which to return results, by default OrderBy.DESCENDING
    return_full : bool, optional
        Whether to make additional API call to fetch the full object, by default True

    Yields
    ------
    Iterator[Parameter]
        An iterator of Parameters matching the given criteria.
    """

    def deserialize(items: list[dict]) -> Iterator[Parameter]:
        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 Parameter '{id}': {e}")
        else:
            yield from (Parameter(**item) for item in items)

    params = {"limit": limit, "orderBy": order_by, "parameters": ids, "startKey": start_key}
    if names:
        params["name"] = [names] if isinstance(names, str) else names
        params["exactMatch"] = json.dumps(exact_match)

    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=deserialize,
    )

update

update(*, parameter: Parameter) -> Parameter

Update a parameter.

Parameters:

Name Type Description Default
parameter Parameter

The updated parameter to save. The parameter must have an ID.

required

Returns:

Type Description
Parameter

The updated parameter as returned by the server.

Source code in src/albert/collections/parameters.py
def update(self, *, parameter: Parameter) -> Parameter:
    """Update a parameter.

    Parameters
    ----------
    parameter : Parameter
        The updated parameter to save. The parameter must have an ID.

    Returns
    -------
    Parameter
        The updated parameter as returned by the server.
    """
    existing = self.get_by_id(id=parameter.id)
    payload = self._generate_patch_payload(
        existing=existing,
        updated=parameter,
    )
    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=parameter,
            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 parameter
    for e in payload_dump["data"]:
        self.session.patch(
            f"{self.base_path}/{parameter.id}",
            json={"data": [e]},
        )
    return self.get_by_id(id=parameter.id)

ParameterGroupCollection

ParameterGroupCollection(*, session: AlbertSession)

Bases: BaseCollection

ParameterGroupCollection is a collection class for managing ParameterGroup entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session to use for making requests.

required

Methods:

Name Description
create

Create a new parameter group.

delete

Delete a parameter group by its ID.

get_by_id

Get a parameter group by its ID.

get_by_ids
get_by_name

Get a parameter group by its name.

list

Search for Parameter Groups matching the given criteria.

update

Update a parameter group.

Source code in src/albert/collections/parameter_groups.py
def __init__(self, *, session: AlbertSession):
    """A collection for interacting with Albert parameter groups.

    Parameters
    ----------
    session : AlbertSession
        The Albert session to use for making requests.
    """
    super().__init__(session=session)
    self.base_path = f"/api/{ParameterGroupCollection._api_version}/parametergroups"

base_path instance-attribute

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

create

create(
    *, parameter_group: ParameterGroup
) -> ParameterGroup

Create a new parameter group.

Parameters:

Name Type Description Default
parameter_group ParameterGroup

The parameter group to create.

required

Returns:

Type Description
ParameterGroup

The created parameter group.

Source code in src/albert/collections/parameter_groups.py
def create(self, *, parameter_group: ParameterGroup) -> ParameterGroup:
    """Create a new parameter group.

    Parameters
    ----------
    parameter_group : ParameterGroup
        The parameter group to create.

    Returns
    -------
    ParameterGroup
        The created parameter group.
    """

    response = self.session.post(
        self.base_path,
        json=parameter_group.model_dump(by_alias=True, exclude_none=True, mode="json"),
    )
    return ParameterGroup(**response.json())

delete

delete(*, id: str) -> None

Delete a parameter group by its ID.

Parameters:

Name Type Description Default
id str

The ID of the parameter group to delete

required
Source code in src/albert/collections/parameter_groups.py
def delete(self, *, id: str) -> None:
    """Delete a parameter group by its ID.

    Parameters
    ----------
    id : str
        The ID of the parameter group to delete
    """
    path = f"{self.base_path}/{id}"
    self.session.delete(path)

get_by_id

get_by_id(*, id: str) -> ParameterGroup

Get a parameter group by its ID.

Parameters:

Name Type Description Default
id str

The ID of the parameter group to retrieve.

required

Returns:

Type Description
ParameterGroup

The parameter group with the given ID.

Source code in src/albert/collections/parameter_groups.py
def get_by_id(self, *, id: str) -> ParameterGroup:
    """Get a parameter group by its ID.

    Parameters
    ----------
    id : str
        The ID of the parameter group to retrieve.

    Returns
    -------
    ParameterGroup
        The parameter group with the given ID.
    """
    path = f"{self.base_path}/{id}"
    response = self.session.get(path)
    return ParameterGroup(**response.json())

get_by_ids

get_by_ids(*, ids: list[str]) -> ParameterGroup
Source code in src/albert/collections/parameter_groups.py
def get_by_ids(self, *, ids: list[str]) -> ParameterGroup:
    url = f"{self.base_path}/ids"
    batches = [ids[i : i + 100] for i in range(0, len(ids), 100)]
    return [
        ParameterGroup(**item)
        for batch in batches
        for item in self.session.get(url, params={"id": batch}).json()["Items"]
    ]

get_by_name

get_by_name(*, name: str) -> ParameterGroup | None

Get a parameter group by its name.

Parameters:

Name Type Description Default
name str

The name of the parameter group to retrieve.

required

Returns:

Type Description
ParameterGroup | None

The parameter group with the given name, or None if not found.

Source code in src/albert/collections/parameter_groups.py
def get_by_name(self, *, name: str) -> ParameterGroup | None:
    """Get a parameter group by its name.

    Parameters
    ----------
    name : str
        The name of the parameter group to retrieve.

    Returns
    -------
    ParameterGroup | None
        The parameter group with the given name, or None if not found.
    """
    matches = self.list(text=name)
    for m in matches:
        if m.name.lower() == name.lower():
            return m
    return None

list

list(
    *,
    text: str | None = None,
    types: PGType | list[PGType] | None = None,
    order_by: OrderBy = DESCENDING,
    limit: int = 25,
    offset: int | None = None,
) -> Iterator[ParameterGroup]

Search for Parameter Groups matching the given criteria.

Parameters:

Name Type Description Default
text str | None

Text to search for, by default None

None
types PGType | list[PGType] | None

Filer the returned Parameter Groups by Type, by default None

None
order_by OrderBy

The order in which to return results, by default OrderBy.DESCENDING

DESCENDING

Yields:

Type Description
Iterator[ParameterGroup]

An iterator of Parameter Groups matching the given criteria.

Source code in src/albert/collections/parameter_groups.py
def list(
    self,
    *,
    text: str | None = None,
    types: PGType | list[PGType] | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    limit: int = 25,
    offset: int | None = None,
) -> Iterator[ParameterGroup]:
    """Search for Parameter Groups matching the given criteria.

    Parameters
    ----------
    text : str | None, optional
        Text to search for, by default None
    types : PGType | list[PGType] | None, optional
        Filer the returned Parameter Groups by Type, by default None
    order_by : OrderBy, optional
        The order in which to return results, by default OrderBy.DESCENDING

    Yields
    ------
    Iterator[ParameterGroup]
        An iterator of Parameter Groups matching the given criteria.
    """

    def deserialize(items: list[dict]) -> Iterator[ParameterGroup]:
        for item in items:
            id = item["albertId"]
            try:
                yield self.get_by_id(id=id)
            except AlbertHTTPError as e:  # pragma: no cover
                logger.warning(f"Error fetching parameter group {id}: {e}")
        # Currently, the API is not returning metadata for the list_by_ids endpoint, so we need to fetch individually until that is fixed
        # return self.get_by_ids(ids=[x["albertId"] for x in items])

    params = {
        "limit": limit,
        "offset": offset,
        "order": order_by.value,
        "text": text,
        "types": [types] if isinstance(types, PGType) else types,
    }

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        params=params,
        deserialize=deserialize,
    )

update

update(
    *, parameter_group: ParameterGroup
) -> ParameterGroup

Update a parameter group.

Parameters:

Name Type Description Default
parameter_group ParameterGroup

The updated ParameterGroup. The ParameterGroup must have an ID.

required

Returns:

Type Description
ParameterGroup

The updated ParameterGroup as returned by the server.

Source code in src/albert/collections/parameter_groups.py
def update(self, *, parameter_group: ParameterGroup) -> ParameterGroup:
    """Update a parameter group.

    Parameters
    ----------
    parameter_group : ParameterGroup
        The updated ParameterGroup. The ParameterGroup must have an ID.

    Returns
    -------
    ParameterGroup
        The updated ParameterGroup as returned by the server.
    """
    existing = self.get_by_id(id=parameter_group.id)
    path = f"{self.base_path}/{existing.id}"

    base_payload = self._generate_patch_payload(
        existing=existing, updated=parameter_group, generate_metadata_diff=True
    )

    general_patches, new_parameter_values, enum_patches = generate_parameter_group_patches(
        initial_patches=base_payload,
        updated_parameter_group=parameter_group,
        existing_parameter_group=existing,
    )

    # add new parameters
    new_param_url = f"{self.base_path}/{parameter_group.id}/parameters"
    if len(new_parameter_values) > 0:
        self.session.put(
            url=new_param_url,
            json=[
                x.model_dump(mode="json", by_alias=True, exclude_none=True)
                for x in new_parameter_values
            ],
        )
    new_param_sequences = [x.sequence for x in new_parameter_values]
    # handle enum updates
    for sequence, ep in enum_patches.items():
        if sequence in new_param_sequences:
            # we don't need to handle enum updates for new parameters
            continue
        if len(ep) > 0:
            enum_url = f"{self.base_path}/{parameter_group.id}/parameters/{sequence}/enums"
            self.session.put(
                url=enum_url,
                json=ep,
            )
    if len(general_patches.data) > 0:
        # patch the general patches
        self.session.patch(
            url=path,
            json=general_patches.model_dump(mode="json", by_alias=True, exclude_none=True),
        )

    return self.get_by_id(id=parameter_group.id)

PricingCollection

PricingCollection(*, session: AlbertSession)

Bases: BaseCollection

PricingCollection is a collection class for managing Pricing entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Creates a new Pricing entity.

delete

Deletes a Pricing entity by its ID.

get_by_id

Retrieves a Pricing entity by its ID.

get_by_inventory_id

Returns a list of Pricing entities for the given inventory ID as per the provided parameters.

get_by_inventory_ids

Returns a list of Pricing resources for each parent inventory ID.

update

Updates a Pricing entity.

Source code in src/albert/collections/pricings.py
def __init__(self, *, session: AlbertSession):
    """Initializes the PricingCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, pricing: Pricing) -> Pricing

Creates a new Pricing entity.

Parameters:

Name Type Description Default
pricing Pricing

The Pricing entity to create.

required

Returns:

Type Description
Pricing

The created Pricing entity.

Source code in src/albert/collections/pricings.py
def create(self, *, pricing: Pricing) -> Pricing:
    """Creates a new Pricing entity.

    Parameters
    ----------
    pricing : Pricing
        The Pricing entity to create.

    Returns
    -------
    Pricing
        The created Pricing entity.
    """
    payload = pricing.model_dump(by_alias=True, exclude_none=True, mode="json")
    response = self.session.post(self.base_path, json=payload)
    return Pricing(**response.json())

delete

delete(*, id: str) -> None

Deletes a Pricing entity by its ID.

Parameters:

Name Type Description Default
id str

The ID of the Pricing entity to delete.

required
Source code in src/albert/collections/pricings.py
def delete(self, *, id: str) -> None:
    """Deletes a Pricing entity by its ID.

    Parameters
    ----------
    id : str
        The ID of the Pricing entity to delete.
    """
    url = f"{self.base_path}/{id}"
    self.session.delete(url)

get_by_id

get_by_id(*, id: str) -> Pricing

Retrieves a Pricing entity by its ID.

Parameters:

Name Type Description Default
id str

The ID of the Pricing entity to retrieve.

required

Returns:

Type Description
Pricing

The Pricing entity if found, None otherwise.

Source code in src/albert/collections/pricings.py
def get_by_id(self, *, id: str) -> Pricing:
    """Retrieves a Pricing entity by its ID.

    Parameters
    ----------
    id : str
        The ID of the Pricing entity to retrieve.

    Returns
    -------
    Pricing
        The Pricing entity if found, None otherwise.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return Pricing(**response.json())

get_by_inventory_id

get_by_inventory_id(
    *,
    inventory_id: str,
    group_by: PricingBy | None = None,
    filter_by: PricingBy | None = None,
    filter_id: str | None = None,
    order_by: OrderBy | None = None,
) -> list[Pricing]

Returns a list of Pricing entities for the given inventory ID as per the provided parameters.

Parameters:

Name Type Description Default
inventory_id str

The ID of the inventory to retrieve pricings for.

required
group_by PricingBy | None

Grouping by PricingBy, by default None

None
filter_by PricingBy | None

Filter by PricingBy, by default None

None
filter_id str | None

The string to use as the filter, by default None

None
order_by OrderBy | None

The order to sort the results by, by default None

None

Returns:

Type Description
list[Pricing]

A list of Pricing entities matching the provided parameters.

Source code in src/albert/collections/pricings.py
def get_by_inventory_id(
    self,
    *,
    inventory_id: str,
    group_by: PricingBy | None = None,
    filter_by: PricingBy | None = None,
    filter_id: str | None = None,
    order_by: OrderBy | None = None,
) -> list[Pricing]:
    """Returns a list of Pricing entities for the given inventory ID as per the provided parameters.

    Parameters
    ----------
    inventory_id : str
        The ID of the inventory to retrieve pricings for.
    group_by : PricingBy | None, optional
        Grouping by PricingBy, by default None
    filter_by : PricingBy | None, optional
        Filter by PricingBy, by default None
    filter_id : str | None, optional
        The string to use as the filter, by default None
    order_by : OrderBy | None, optional
        The order to sort the results by, by default None

    Returns
    -------
    list[Pricing]
        A list of Pricing entities matching the provided parameters.
    """
    params = {
        "parentId": inventory_id,
        "groupBy": group_by,
        "filterBy": filter_by,
        "id": filter_id,
        "orderBy": order_by,
    }
    params = {k: v for k, v in params.items() if v is not None}
    response = self.session.get(self.base_path, params=params)
    items = response.json().get("Items", [])
    return [Pricing(**x) for x in items]

get_by_inventory_ids

get_by_inventory_ids(
    *, inventory_ids: list[InventoryId]
) -> list[InventoryPricings]

Returns a list of Pricing resources for each parent inventory ID.

Parameters:

Name Type Description Default
inventory_ids list[str]

The list of inventory IDs to retrieve pricings for.

required

Returns:

Type Description
list[InventoryPricing]

A list of InventoryPricing objects matching the provided inventory.

Source code in src/albert/collections/pricings.py
@validate_call
def get_by_inventory_ids(self, *, inventory_ids: list[InventoryId]) -> list[InventoryPricings]:
    """Returns a list of Pricing resources for each parent inventory ID.

    Parameters
    ----------
    inventory_ids : list[str]
        The list of inventory IDs to retrieve pricings for.

    Returns
    -------
    list[InventoryPricing]
        A list of InventoryPricing objects matching the provided inventory.
    """
    params = {"id": inventory_ids}
    response = self.session.get(f"{self.base_path}/ids", params=params)
    return [InventoryPricings(**x) for x in response.json()["Items"]]

update

update(*, pricing: Pricing) -> Pricing

Updates a Pricing entity.

Parameters:

Name Type Description Default
pricing Pricing

The updated Pricing entity.

required

Returns:

Type Description
Pricing

The updated Pricing entity as it appears in Albert.

Source code in src/albert/collections/pricings.py
def update(self, *, pricing: Pricing) -> Pricing:
    """Updates a Pricing entity.

    Parameters
    ----------
    pricing : Pricing
        The updated Pricing entity.

    Returns
    -------
    Pricing
        The updated Pricing entity as it appears in Albert.
    """
    current_pricing = self.get_by_id(id=pricing.id)
    patch_payload = self._pricing_patch_payload(existing=current_pricing, updated=pricing)
    self.session.patch(
        url=f"{self.base_path}/{pricing.id}",
        json=patch_payload.model_dump(mode="json", by_alias=True),
    )
    return self.get_by_id(id=pricing.id)

ProductDesignCollection

ProductDesignCollection(*, session: AlbertSession)

Bases: BaseCollection

ProductDesignCollection is a collection class for managing Product Design entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
get_unpacked_products

Get unpacked products by inventory IDs.

Source code in src/albert/collections/product_design.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the CasCollection with the provided session.

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

base_path instance-attribute

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

get_unpacked_products

get_unpacked_products(
    *,
    inventory_ids: list[InventoryId],
    unpack_id: Literal[
        "DESIGN", "PREDICTION"
    ] = "PREDICTION",
) -> list[UnpackedProductDesign]

Get unpacked products by inventory IDs.

Parameters:

Name Type Description Default
inventory_ids list[InventoryId]

The inventory ids to get unpacked formulas for.

required
unpack_id Literal['DESIGN', 'PREDICTION']

The ID for the unpack operation.

'PREDICTION'

Returns:

Type Description
list[UnpackedProductDesign]

The unpacked products/formulas.

Source code in src/albert/collections/product_design.py
@validate_call
def get_unpacked_products(
    self,
    *,
    inventory_ids: list[InventoryId],
    unpack_id: Literal["DESIGN", "PREDICTION"] = "PREDICTION",
) -> list[UnpackedProductDesign]:
    """
    Get unpacked products by inventory IDs.

    Parameters
    ----------
    inventory_ids : list[InventoryId]
        The inventory ids to get unpacked formulas for.
    unpack_id: Literal["DESIGN", "PREDICTION"]
        The ID for the unpack operation.

    Returns
    -------
    list[UnpackedProductDesign]
        The unpacked products/formulas.
    """
    url = f"{self.base_path}/{unpack_id}/unpack"
    batches = [inventory_ids[i : i + 50] for i in range(0, len(inventory_ids), 50)]
    return [
        UnpackedProductDesign(**item)
        for batch in batches
        for item in self.session.get(url, params={"formulaId": batch}).json()
    ]

ProjectCollection

ProjectCollection(*, session: AlbertSession)

Bases: BaseCollection

ProjectCollection is a collection class for managing Project entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Create a new project.

delete

Delete a project by its ID.

get_by_id

Retrieve a project by its ID.

list

List projects with optional filters.

update

Update a project.

Source code in src/albert/collections/projects.py
def __init__(self, *, session: AlbertSession):
    """
    Initialize a ProjectCollection object.

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

base_path instance-attribute

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

create

create(*, project: Project) -> Project

Create a new project.

Parameters:

Name Type Description Default
project Project

The project to create.

required

Returns:

Type Description
Optional[Project]

The created project object if successful, None otherwise.

Source code in src/albert/collections/projects.py
def create(self, *, project: Project) -> Project:
    """
    Create a new project.

    Parameters
    ----------
    project : Project
        The project to create.

    Returns
    -------
    Optional[Project]
        The created project object if successful, None otherwise.
    """
    response = self.session.post(
        self.base_path, json=project.model_dump(by_alias=True, exclude_unset=True, mode="json")
    )
    return Project(**response.json())

delete

delete(*, id: str) -> None

Delete a project by its ID.

Parameters:

Name Type Description Default
id str

The ID of the project to delete.

required

Returns:

Type Description
None
Source code in src/albert/collections/projects.py
def delete(self, *, id: str) -> None:
    """
    Delete a project by its ID.

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

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

get_by_id

get_by_id(*, id: str) -> Project

Retrieve a project by its ID.

Parameters:

Name Type Description Default
id str

The ID of the project to retrieve.

required

Returns:

Type Description
Project

The project object if found

Source code in src/albert/collections/projects.py
def get_by_id(self, *, id: str) -> Project:
    """
    Retrieve a project by its ID.

    Parameters
    ----------
    id : str
        The ID of the project to retrieve.

    Returns
    -------
    Project
        The project object if found
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)

    return Project(**response.json())

list

list(
    *,
    text: str = None,
    status: list[str] = None,
    market_segment: list[str] = None,
    application: list[str] = None,
    technology: list[str] = None,
    created_by: list[str] = None,
    location: list[str] = None,
    from_created_at: str = None,
    to_created_at: str = None,
    facet_field: str = None,
    facet_text: str = None,
    contains_field: list[str] = None,
    contains_text: list[str] = None,
    linked_to: str = None,
    my_projects: bool = None,
    my_role: list[str] = None,
    order_by: OrderBy = DESCENDING,
    sort_by: str = None,
    limit: int = 50,
) -> Iterator[Project]

List projects with optional filters.

Parameters:

Name Type Description Default
text str

Search any test in the project.

None
status list[str]

The status filter for the projects.

None
market_segment list[str]

The market segment filter for the projects.

None
application list[str]

The application filter for the projects.

None
technology list[str]

The technology filter for the projects.

None
created_by list[str]

The name of the user who created the project.

None
location list[str]

The location filter for the projects.

None
from_created_at str

The start date filter for the projects.

None
to_created_at str

The end date filter for the projects.

None
facet_field str

The facet field for the projects.

None
facet_text str

The facet text for the projects.

None
contains_field list[str]

To power project facets search

None
contains_text list[str]

To power project facets search

None
linked_to str

To pass text for linked to dropdown search in Task creation flow.

None
my_projects bool

Return Projects owned by you.

None
my_role list[str]

Filter Projects to ones which you have a specific role in.

None
order_by OrderBy

The order in which to retrieve items (default is OrderBy.DESCENDING).

DESCENDING
sort_by str

The field to sort by.

None

Returns:

Type Description
Iterator[Project]

An iterator of Project resources.

Source code in src/albert/collections/projects.py
def list(
    self,
    *,
    text: str = None,
    status: list[str] = None,
    market_segment: list[str] = None,
    application: list[str] = None,
    technology: list[str] = None,
    created_by: list[str] = None,
    location: list[str] = None,
    from_created_at: str = None,
    to_created_at: str = None,
    facet_field: str = None,
    facet_text: str = None,
    contains_field: list[str] = None,
    contains_text: list[str] = None,
    linked_to: str = None,
    my_projects: bool = None,
    my_role: list[str] = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    sort_by: str = None,
    limit: int = 50,
) -> Iterator[Project]:
    """
    List projects with optional filters.

    Parameters
    ----------
    text : str, optional
        Search any test in the project.
    status : list[str], optional
        The status filter for the projects.
    market_segment : list[str], optional
        The market segment filter for the projects.
    application : list[str], optional
        The application filter for the projects.
    technology : list[str], optional
        The technology filter for the projects.
    created_by : list[str], optional
        The name of the user who created the project.
    location : list[str], optional
        The location filter for the projects.
    from_created_at : str, optional
        The start date filter for the projects.
    to_created_at : str, optional
        The end date filter for the projects.
    facet_field : str, optional
        The facet field for the projects.
    facet_text : str, optional
        The facet text for the projects.
    contains_field : list[str], optional
        To power project facets search
    contains_text : list[str], optional
        To power project facets search
    linked_to : str, optional
        To pass text for linked to dropdown search in Task creation flow.
    my_projects : bool, optional
        Return Projects owned by you.
    my_role : list[str], optional
        Filter Projects to ones which you have a specific role in.
    order_by : OrderBy, optional
        The order in which to retrieve items (default is OrderBy.DESCENDING).
    sort_by : str, optional
        The field to sort by.

    Returns
    ------
    Iterator[Project]
        An iterator of Project resources.
    """
    params = {
        "limit": limit,
        "order": order_by.value,
        "text": text,
        "sortBy": sort_by,
        "status": status,
        "marketSegment": market_segment,
        "application": application,
        "technology": technology,
        "createdBy": created_by,
        "location": location,
        "fromCreatedAt": from_created_at,
        "toCreatedAt": to_created_at,
        "facetField": facet_field,
        "facetText": facet_text,
        "containsField": contains_field,
        "containsText": contains_text,
        "linkedTo": linked_to,
        "myProjects": my_projects,
        "myRole": my_role,
    }
    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        params=params,
        deserialize=lambda items: [Project(**item) for item in items],
    )

update

update(*, project: Project) -> Project

Update a project.

Parameters:

Name Type Description Default
project Project

The updated project object.

required

Returns:

Type Description
Project

The updated project object as returned by the server.

Source code in src/albert/collections/projects.py
def update(self, *, project: Project) -> Project:
    """Update a project.

    Parameters
    ----------
    project : Project
        The updated project object.

    Returns
    -------
    Project
        The updated project object as returned by the server.
    """
    existing_project = self.get_by_id(id=project.id)
    patch_data = self._generate_patch_payload(existing=existing_project, updated=project)
    url = f"{self.base_path}/{project.id}"

    self.session.patch(url, json=patch_data.model_dump(mode="json", by_alias=True))

    return self.get_by_id(id=project.id)

PropertyDataCollection

PropertyDataCollection(*, session: AlbertSession)

Bases: BaseCollection

PropertyDataCollection is a collection class for managing Property Data entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
add_properties_to_inventory

Add new properties to an inventory item.

add_properties_to_task

Add new task properties for a given task.

bulk_delete_task_data

Bulk delete task data for a given task.

bulk_load_task_properties

Bulk load task properties for a given task. WARNING: This will overwrite any existing properties!

check_block_interval_for_data

Check if a specific block interval has data.

check_for_task_data

Checks if a task has data.

get_all_task_properties

Returns all the properties for a specific task.

get_properties_on_inventory

Returns all the properties of an inventory item.

get_task_block_properties

Returns all the properties within a Property Task block for a specific inventory item.

search

Search for property data with various filtering options.

update_or_create_task_properties

Update or create task properties for a given task.

update_property_on_inventory

Update a property on an inventory item.

update_property_on_task

Updates a specific property on a task.

Source code in src/albert/collections/property_data.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the CompanyCollection with the provided session.

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

base_path instance-attribute

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

add_properties_to_inventory

add_properties_to_inventory(
    *,
    inventory_id: InventoryId,
    properties: list[InventoryDataColumn],
) -> list[InventoryPropertyDataCreate]

Add new properties to an inventory item.

Parameters:

Name Type Description Default
inventory_id InventoryId

The ID of the inventory item to add properties to.

required
properties list[InventoryDataColumn]

The properties to add.

required

Returns:

Type Description
list[InventoryPropertyDataCreate]

The registered properties.

Source code in src/albert/collections/property_data.py
@validate_call
def add_properties_to_inventory(
    self, *, inventory_id: InventoryId, properties: list[InventoryDataColumn]
) -> list[InventoryPropertyDataCreate]:
    """Add new properties to an inventory item.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory item to add properties to.
    properties : list[InventoryDataColumn]
        The properties to add.

    Returns
    -------
    list[InventoryPropertyDataCreate]
        The registered properties.
    """
    returned = []
    for p in properties:
        # Can only add one at a time.
        create_object = InventoryPropertyDataCreate(
            inventory_id=inventory_id, data_columns=[p]
        )
        response = self.session.post(
            self.base_path,
            json=create_object.model_dump(exclude_none=True, by_alias=True, mode="json"),
        )
        response_json = response.json()
        logger.info(response_json.get("message", None))
        returned.append(InventoryPropertyDataCreate(**response_json))
    return returned

add_properties_to_task

add_properties_to_task(
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
    properties: list[TaskPropertyCreate],
)

Add new task properties for a given task.

This method only works for new values. If a trial number is provided in the TaskPropertyCreate, it must relate to an existing trial. New trials must be added with no trial number provided. Do not try to create multiple new trials in one call as this will lead to unexpected behavior. Build out new trials in a loop if many new trials are needed.

Parameters:

Name Type Description Default
inventory_id InventoryId

The ID of the inventory.

required
task_id TaskId

The ID of the task.

required
block_id BlockId

The ID of the block.

required
lot_id LotId

The ID of the lot, by default None.

None
properties list[TaskPropertyCreate]

A list of TaskPropertyCreate objects representing the properties to add.

required

Returns:

Type Description
list[TaskPropertyData]

The newly created task properties.

Source code in src/albert/collections/property_data.py
@validate_call
def add_properties_to_task(
    self,
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
    properties: list[TaskPropertyCreate],
):
    """
    Add new task properties for a given task.

    This method only works for new values. If a trial number is provided in the TaskPropertyCreate,
    it must relate to an existing trial. New trials must be added with no trial number provided.
    Do not try to create multiple new trials in one call as this will lead to unexpected behavior.
    Build out new trials in a loop if many new trials are needed.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory.
    task_id : TaskId
        The ID of the task.
    block_id : BlockId
        The ID of the block.
    lot_id : LotId, optional
        The ID of the lot, by default None.
    properties : list[TaskPropertyCreate]
        A list of TaskPropertyCreate objects representing the properties to add.

    Returns
    -------
    list[TaskPropertyData]
        The newly created task properties.
    """
    params = {
        "blockId": block_id,
        "inventoryId": inventory_id,
        "lotId": lot_id,
        "autoCalculate": "true",
        "history": "true",
    }
    params = {k: v for k, v in params.items() if v is not None}
    response = self.session.post(
        url=f"{self.base_path}/{task_id}",
        json=[x.model_dump(exclude_none=True, by_alias=True, mode="json") for x in properties],
        params=params,
    )

    registered_properties = [
        TaskPropertyCreate(**x) for x in response.json() if "DataTemplate" in x
    ]
    existing_data_rows = self.get_task_block_properties(
        inventory_id=inventory_id, task_id=task_id, block_id=block_id, lot_id=lot_id
    )
    patches = self._form_calculated_task_property_patches(
        existing_data_rows=existing_data_rows, properties=registered_properties
    )
    if len(patches) > 0:
        return self.update_property_on_task(task_id=task_id, patch_payload=patches)
    else:
        return self.get_all_task_properties(task_id=task_id)

bulk_delete_task_data

bulk_delete_task_data(
    *,
    task_id: TaskId,
    block_id: BlockId,
    inventory_id: InventoryId,
    lot_id: LotId | None = None,
    interval_id=None,
) -> None

Bulk delete task data for a given task.

Parameters:

Name Type Description Default
task_id TaskId

The ID of the task.

required
block_id BlockId

The ID of the block.

required
inventory_id InventoryId

The ID of the inventory.

required
lot_id LotId

The ID of the lot, by default None.

None
interval_id IntervalId

The ID of the interval, by default None. If provided, will delete data for this specific interval.

None

Returns:

Type Description
None
Source code in src/albert/collections/property_data.py
def bulk_delete_task_data(
    self,
    *,
    task_id: TaskId,
    block_id: BlockId,
    inventory_id: InventoryId,
    lot_id: LotId | None = None,
    interval_id=None,
) -> None:
    """
    Bulk delete task data for a given task.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task.
    block_id : BlockId
        The ID of the block.
    inventory_id : InventoryId
        The ID of the inventory.
    lot_id : LotId, optional
        The ID of the lot, by default None.
    interval_id : IntervalId, optional
        The ID of the interval, by default None. If provided, will delete data for this specific interval.

    Returns
    -------
    None
    """
    params = {
        "inventoryId": inventory_id,
        "blockId": block_id,
        "lotId": lot_id,
        "intervalRow": interval_id if interval_id != "default" else None,
    }
    params = {k: v for k, v in params.items() if v is not None}
    self.session.delete(f"{self.base_path}/{task_id}", params=params)

bulk_load_task_properties

bulk_load_task_properties(
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    property_data: BulkPropertyData,
    interval="default",
    lot_id: LotId = None,
) -> list[TaskPropertyData]

Bulk load task properties for a given task. WARNING: This will overwrite any existing properties! BulkPropertyData column names must exactly match the names of the data columns (Case Sensitive).

Parameters:

Name Type Description Default
inventory_id InventoryId

The ID of the inventory.

required
task_id TaskId

The ID of the task.

required
block_id BlockId

The ID of the block.

required
lot_id LotId

The ID of the lot, by default None.

None
interval str

The interval to use for the properties, by default "default". Can be obtained using Workflow.get_interval_id().

'default'
property_data BulkPropertyData

A list of columnwise data containing all your rows of data for a single interval. Can be created using BulkPropertyData.from_dataframe().

required

Returns:

Type Description
list[TaskPropertyData]

The updated or newly created task properties.

Example
from albert.resources.property_data import BulkPropertyData

data = BulkPropertyData.from_dataframe(df=my_dataframe)
res = client.property_data.bulk_load_task_properties(
    block_id="BLK1",
    inventory_id="INVEXP102748-042",
    property_data=data,
    task_id="TASFOR291760",
)

[TaskPropertyData(id="TASFOR291760", ...)]
Source code in src/albert/collections/property_data.py
def bulk_load_task_properties(
    self,
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    property_data: BulkPropertyData,
    interval="default",
    lot_id: LotId = None,
) -> list[TaskPropertyData]:
    """
    Bulk load task properties for a given task. WARNING: This will overwrite any existing properties!
    BulkPropertyData column names must exactly match the names of the data columns (Case Sensitive).

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory.
    task_id : TaskId
        The ID of the task.
    block_id : BlockId
        The ID of the block.
    lot_id : LotId, optional
        The ID of the lot, by default None.
    interval : str, optional
        The interval to use for the properties, by default "default". Can be obtained using Workflow.get_interval_id().
    property_data : BulkPropertyData
        A list of columnwise data containing all your rows of data for a single interval. Can be created using BulkPropertyData.from_dataframe().

    Returns
    -------
    list[TaskPropertyData]
        The updated or newly created task properties.

    Example
    -------

    ```python
    from albert.resources.property_data import BulkPropertyData

    data = BulkPropertyData.from_dataframe(df=my_dataframe)
    res = client.property_data.bulk_load_task_properties(
        block_id="BLK1",
        inventory_id="INVEXP102748-042",
        property_data=data,
        task_id="TASFOR291760",
    )

    [TaskPropertyData(id="TASFOR291760", ...)]
    ```
    """
    property_df = pd.DataFrame(
        {x.data_column_name: x.data_series for x in property_data.columns}
    )

    def _get_column_map(dataframe: pd.DataFrame, property_data: TaskPropertyData):
        data_col_info = property_data.data[0].trials[0].data_columns  # PropertyValue
        column_map = {}
        for col in dataframe.columns:
            column = [x for x in data_col_info if x.name == col]
            if len(column) == 1:
                column_map[col] = column[0]
            else:
                raise ValueError(
                    f"Column '{col}' not found in block data columns or multiple matches found."
                )
        return column_map

    def _df_to_task_prop_create_list(
        dataframe: pd.DataFrame,
        column_map: dict[str, PropertyValue],
        data_template_id: DataTemplateId,
    ) -> list[TaskPropertyCreate]:
        task_prop_create_list = []
        for i, row in dataframe.iterrows():
            for col_name, col_info in column_map.items():
                if col_name not in dataframe.columns:
                    raise ValueError(f"Column '{col_name}' not found in DataFrame.")

                task_prop_create = TaskPropertyCreate(
                    data_column=TaskDataColumn(
                        data_column_id=col_info.id,
                        column_sequence=col_info.sequence,
                    ),
                    value=str(row[col_name]),
                    visible_trial_number=i + 1,
                    interval_combination=interval,
                    data_template=EntityLink(id=data_template_id),
                )
                task_prop_create_list.append(task_prop_create)
        return task_prop_create_list

    task_prop_data = self.get_task_block_properties(
        inventory_id=inventory_id, task_id=task_id, block_id=block_id, lot_id=lot_id
    )
    column_map = _get_column_map(property_df, task_prop_data)
    all_task_prop_create = _df_to_task_prop_create_list(
        dataframe=property_df,
        column_map=column_map,
        data_template_id=task_prop_data.data_template.id,
    )
    with suppress(NotFoundError):
        # This is expected if the task is new and has no data yet.
        self.bulk_delete_task_data(
            task_id=task_id,
            block_id=block_id,
            inventory_id=inventory_id,
            lot_id=lot_id,
            interval_id=interval,
        )
    return self.add_properties_to_task(
        inventory_id=inventory_id,
        task_id=task_id,
        block_id=block_id,
        lot_id=lot_id,
        properties=all_task_prop_create,
    )

check_block_interval_for_data

check_block_interval_for_data(
    *,
    block_id: BlockId,
    task_id: TaskId,
    interval_id: IntervalId,
) -> CheckPropertyData

Check if a specific block interval has data.

Parameters:

Name Type Description Default
block_id BlockId

The ID of the block.

required
task_id TaskId

The ID of the task.

required
interval_id IntervalId

The ID of the interval.

required

Returns:

Type Description
CheckPropertyData

description

Source code in src/albert/collections/property_data.py
@validate_call
def check_block_interval_for_data(
    self, *, block_id: BlockId, task_id: TaskId, interval_id: IntervalId
) -> CheckPropertyData:
    """Check if a specific block interval has data.

    Parameters
    ----------
    block_id : BlockId
        The ID of the block.
    task_id : TaskId
        The ID of the task.
    interval_id : IntervalId
        The ID of the interval.

    Returns
    -------
    CheckPropertyData
        _description_
    """
    params = {
        "entity": "block",
        "action": "checkdata",
        "id": block_id,
        "parentId": task_id,
        "intervalId": interval_id,
    }

    response = self.session.get(url=self.base_path, params=params)
    return CheckPropertyData(response.json())

check_for_task_data

check_for_task_data(
    *, task_id: TaskId
) -> list[CheckPropertyData]

Checks if a task has data.

Parameters:

Name Type Description Default
task_id TaskId

The ID of the task to check for data.

required

Returns:

Type Description
list[CheckPropertyData]

A list of CheckPropertyData objects representing the data status of each block + inventory item of the task.

Source code in src/albert/collections/property_data.py
@validate_call
def check_for_task_data(self, *, task_id: TaskId) -> list[CheckPropertyData]:
    """Checks if a task has data.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task to check for data.

    Returns
    -------
    list[CheckPropertyData]
        A list of CheckPropertyData objects representing the data status of each block + inventory item of the task.
    """
    task_info = self._get_task_from_id(id=task_id)

    params = {
        "entity": "block",
        "action": "checkdata",
        "parentId": task_id,
        "id": [x.id for x in task_info.blocks],
    }

    response = self.session.get(url=self.base_path, params=params)
    return [CheckPropertyData(**x) for x in response.json()]

get_all_task_properties

get_all_task_properties(
    *, task_id: TaskId
) -> list[TaskPropertyData]

Returns all the properties for a specific task.

Parameters:

Name Type Description Default
task_id TaskId

The ID of the task to retrieve properties for.

required

Returns:

Type Description
list[TaskPropertyData]

A list of TaskPropertyData objects representing the properties within the task.

Source code in src/albert/collections/property_data.py
@validate_call
def get_all_task_properties(self, *, task_id: TaskId) -> list[TaskPropertyData]:
    """Returns all the properties for a specific task.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task to retrieve properties for.

    Returns
    -------
    list[TaskPropertyData]
        A list of TaskPropertyData objects representing the properties within the task.
    """
    all_info = []
    task_data_info = self.check_for_task_data(task_id=task_id)
    for combo_info in task_data_info:
        all_info.append(
            self.get_task_block_properties(
                inventory_id=combo_info.inventory_id,
                task_id=task_id,
                block_id=combo_info.block_id,
                lot_id=combo_info.lot_id,
            )
        )

    return all_info

get_properties_on_inventory

get_properties_on_inventory(
    *, inventory_id: InventoryId
) -> InventoryPropertyData

Returns all the properties of an inventory item.

Parameters:

Name Type Description Default
inventory_id InventoryId

The ID of the inventory item to retrieve properties for.

required

Returns:

Type Description
InventoryPropertyData

The properties of the inventory item.

Source code in src/albert/collections/property_data.py
@validate_call
def get_properties_on_inventory(self, *, inventory_id: InventoryId) -> InventoryPropertyData:
    """Returns all the properties of an inventory item.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory item to retrieve properties for.

    Returns
    -------
    InventoryPropertyData
        The properties of the inventory item.
    """
    params = {"entity": "inventory", "id": [inventory_id]}
    response = self.session.get(url=self.base_path, params=params)
    response_json = response.json()
    return InventoryPropertyData(**response_json[0])

get_task_block_properties

get_task_block_properties(
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
) -> TaskPropertyData

Returns all the properties within a Property Task block for a specific inventory item.

Parameters:

Name Type Description Default
inventory_id InventoryId

The ID of the inventory.

required
task_id TaskId

The Property task ID.

required
block_id BlockId

The Block ID of the block to retrieve properties for.

required
lot_id LotId | None

The specific Lot of the inventory Item to retrieve lots for, by default None

None

Returns:

Type Description
TaskPropertyData

The properties of the inventory item within the block.

Source code in src/albert/collections/property_data.py
@validate_call
def get_task_block_properties(
    self,
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
) -> TaskPropertyData:
    """Returns all the properties within a Property Task block for a specific inventory item.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory.
    task_id : TaskId
        The Property task ID.
    block_id : BlockId
        The Block ID of the block to retrieve properties for.
    lot_id : LotId | None, optional
        The specific Lot of the inventory Item to retrieve lots for, by default None

    Returns
    -------
    TaskPropertyData
        The properties of the inventory item within the block.
    """
    params = {
        "entity": "task",
        "blockId": block_id,
        "id": task_id,
        "inventoryId": inventory_id,
        "lotId": lot_id,
    }
    params = {k: v for k, v in params.items() if v is not None}

    response = self.session.get(url=self.base_path, params=params)
    response_json = response.json()
    return TaskPropertyData(**response_json[0])

search

search(
    *,
    limit: int = 100,
    result: str | None = None,
    text: str | None = None,
    order: OrderBy | None = None,
    sort_by: str | None = None,
    inventory_ids: list[SearchInventoryId]
    | SearchInventoryId
    | None = None,
    project_ids: list[SearchProjectId]
    | SearchProjectId
    | None = None,
    lot_ids: list[LotId] | LotId | None = None,
    data_template_ids: DataTemplateId
    | list[DataTemplateId]
    | None = None,
    data_column_ids: DataColumnId
    | list[DataColumnId]
    | None = None,
    category: list[DataEntity] | DataEntity | None = None,
    data_templates: list[str] | str | None = None,
    data_columns: list[str] | str | None = None,
    parameters: list[str] | str | None = None,
    parameter_group: list[str] | str | None = None,
    unit: list[str] | str | None = None,
    created_by: list[UserId] | UserId | None = None,
    task_created_by: list[UserId] | UserId | None = None,
    return_fields: list[str] | str | None = None,
    return_facets: list[str] | str | None = None,
) -> Iterator[PropertyDataSearchItem]

Search for property data with various filtering options.

Parameters:

Name Type Description Default
limit int

Maximum number of results to return.

100
result str

Find results using search syntax. e.g. to find all results with viscosity < 200 at a temperature of 25 we would do result=viscosity(<200)@temperature(25)

None
text str

Free text search across all searchable fields.

None
order OrderBy

Sort order (ascending/descending).

None
sort_by str

Field to sort results by.

None
inventory_ids SearchInventoryIdType or list of SearchInventoryIdType

Filter by inventory IDs.

None
project_ids ProjectIdType or list of ProjectIdType

Filter by project IDs.

None
lot_ids LotIdType or list of LotIdType

Filter by lot IDs.

None
data_template_ids DataTemplateId or list of DataTemplateId

Filter by data template IDs.

None
data_column_ids DataColumnId | list[DataColumnId] | None

Filter by data column IDs.

None
category DataEntity or list of DataEntity

Filter by data entity categories.

None
data_templates str or list of str (exact match)

Filter by data template names.

None
data_columns str or list of str (exact match)

Filter by data column names (currently non-functional).

None
parameters str or list of str (exact match)

Filter by parameter names.

None
parameter_group str or list of str (exact match)

Filter by parameter group names.

None
unit str or list of str (exact match)

Filter by unit names.

None
created_by UserIdType or list of UserIdType

Filter by creator user IDs.

None
task_created_by UserIdType or list of UserIdType

Filter by task creator user IDs.

None
return_fields str or list of str

Specific fields to include in results. If None, returns all fields.

None
return_facets str or list of str

Specific facets to include in results.

None

Returns:

Type Description
dict

Search results matching the specified criteria.

Source code in src/albert/collections/property_data.py
@validate_call
def search(
    self,
    *,
    limit: int = 100,
    result: str | None = None,
    text: str | None = None,
    # Sorting/pagination
    order: OrderBy | None = None,
    sort_by: str | None = None,
    # Core platform identifiers
    inventory_ids: list[SearchInventoryId] | SearchInventoryId | None = None,
    project_ids: list[SearchProjectId] | SearchProjectId | None = None,
    lot_ids: list[LotId] | LotId | None = None,
    data_template_ids: DataTemplateId | list[DataTemplateId] | None = None,
    data_column_ids: DataColumnId | list[DataColumnId] | None = None,
    # Data structure filters
    category: list[DataEntity] | DataEntity | None = None,
    data_templates: list[str] | str | None = None,
    data_columns: list[str] | str | None = None,
    # Data content filters
    parameters: list[str] | str | None = None,
    parameter_group: list[str] | str | None = None,
    unit: list[str] | str | None = None,
    # User filters
    created_by: list[UserId] | UserId | None = None,
    task_created_by: list[UserId] | UserId | None = None,
    # Response customization
    return_fields: list[str] | str | None = None,
    return_facets: list[str] | str | None = None,
) -> Iterator[PropertyDataSearchItem]:
    """Search for property data with various filtering options.

    Parameters
    ----------
    limit : int, default=100
        Maximum number of results to return.
    result : str, optional
        Find results using search syntax. e.g. to find all results with viscosity < 200 at a temperature of 25 we would do
        result=viscosity(<200)@temperature(25)
    text : str, optional
        Free text search across all searchable fields.
    order : OrderBy, optional
        Sort order (ascending/descending).
    sort_by : str, optional
        Field to sort results by.
    inventory_ids : SearchInventoryIdType or list of SearchInventoryIdType, optional
        Filter by inventory IDs.
    project_ids : ProjectIdType or list of ProjectIdType, optional
        Filter by project IDs.
    lot_ids : LotIdType or list of LotIdType, optional
        Filter by lot IDs.
    data_template_ids : DataTemplateId or list of DataTemplateId, optional
        Filter by data template IDs.
    data_column_ids: DataColumnId or list of DataColumnId, optional
        Filter by data column IDs.
    category : DataEntity or list of DataEntity, optional
        Filter by data entity categories.
    data_templates : str or list of str (exact match), optional
        Filter by data template names.
    data_columns : str or list of str (exact match), optional
        Filter by data column names (currently non-functional).
    parameters : str or list of str (exact match), optional
        Filter by parameter names.
    parameter_group : str or list of str (exact match), optional
        Filter by parameter group names.
    unit : str or list of str (exact match), optional
        Filter by unit names.
    created_by : UserIdType or list of UserIdType, optional
        Filter by creator user IDs.
    task_created_by : UserIdType or list of UserIdType, optional
        Filter by task creator user IDs.
    return_fields : str or list of str, optional
        Specific fields to include in results. If None, returns all fields.
    return_facets : str or list of str, optional
        Specific facets to include in results.

    Returns
    -------
    dict
        Search results matching the specified criteria.
    """

    def deserialize(items: list[dict]) -> list[PropertyDataSearchItem]:
        return [PropertyDataSearchItem.model_validate(x) for x in items]

    if isinstance(inventory_ids, str):
        inventory_ids = [inventory_ids]
    if isinstance(project_ids, str):
        project_ids = [project_ids]
    if isinstance(lot_ids, str):
        lot_ids = [lot_ids]
    if isinstance(data_template_ids, str):
        data_template_ids = [data_template_ids]
    if isinstance(data_column_ids, str):
        data_column_ids = [data_column_ids]
    if isinstance(category, DataEntity):
        category = [category]
    if isinstance(data_templates, str):
        data_templates = [data_templates]
    if isinstance(data_columns, str):
        data_columns = [data_columns]
    if isinstance(parameters, str):
        parameters = [parameters]
    if isinstance(parameter_group, str):
        parameter_group = [parameter_group]
    if isinstance(unit, str):
        unit = [unit]
    if isinstance(created_by, str):
        created_by = [created_by]
    if isinstance(task_created_by, str):
        task_created_by = [task_created_by]
    if isinstance(return_fields, str):
        return_fields = [return_fields]
    if isinstance(return_facets, str):
        return_facets = [return_facets]

    params = {
        "limit": limit,
        "result": result,
        "text": text,
        "order": order.value if order is not None else None,
        "sortBy": sort_by,
        "inventoryIds": inventory_ids if inventory_ids is not None else None,
        "projectIds": project_ids if project_ids is not None else None,
        "lotIds": lot_ids if lot_ids is not None else None,
        "dataTemplateId": data_template_ids if data_template_ids is not None else None,
        "dataColumnId": data_column_ids if data_column_ids is not None else None,
        "category": [c.value for c in category] if category is not None else None,
        "dataTemplates": data_templates,
        "dataColumns": data_columns,
        "parameters": parameters,
        "parameterGroup": parameter_group,
        "unit": unit,
        "createdBy": created_by if created_by is not None else None,
        "taskCreatedBy": task_created_by if task_created_by is not None else None,
        "returnFields": return_fields,
        "returnFacets": return_facets,
    }

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        params=params,
        session=self.session,
        deserialize=deserialize,
    )

update_or_create_task_properties

update_or_create_task_properties(
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
    properties: list[TaskPropertyCreate],
) -> list[TaskPropertyData]

Update or create task properties for a given task.

If a trial number is provided in the TaskPropertyCreate, it must relate to an existing trial. New trials must be added with no trial number provided. Do not try to create multiple new trials in one call as this will lead to unexpected behavior. Build out new trials in a loop if many new trials are needed.

Parameters:

Name Type Description Default
inventory_id InventoryId

The ID of the inventory.

required
task_id TaskId

The ID of the task.

required
block_id BlockId

The ID of the block.

required
lot_id LotId

The ID of the lot, by default None.

None
properties list[TaskPropertyCreate]

A list of TaskPropertyCreate objects representing the properties to update or create.

required

Returns:

Type Description
list[TaskPropertyData]

The updated or newly created task properties.

Source code in src/albert/collections/property_data.py
@validate_call
def update_or_create_task_properties(
    self,
    *,
    inventory_id: InventoryId,
    task_id: TaskId,
    block_id: BlockId,
    lot_id: LotId | None = None,
    properties: list[TaskPropertyCreate],
) -> list[TaskPropertyData]:
    """
    Update or create task properties for a given task.

    If a trial number is provided in the TaskPropertyCreate, it must relate to an existing trial.
    New trials must be added with no trial number provided. Do not try to create multiple new trials
    in one call as this will lead to unexpected behavior. Build out new trials in a loop if many new
    trials are needed.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory.
    task_id : TaskId
        The ID of the task.
    block_id : BlockId
        The ID of the block.
    lot_id : LotId, optional
        The ID of the lot, by default None.
    properties : list[TaskPropertyCreate]
        A list of TaskPropertyCreate objects representing the properties to update or create.

    Returns
    -------
    list[TaskPropertyData]
        The updated or newly created task properties.

    """
    existing_data_rows = self.get_task_block_properties(
        inventory_id=inventory_id, task_id=task_id, block_id=block_id, lot_id=lot_id
    )
    update_patches, new_values = self._form_existing_row_value_patches(
        existing_data_rows=existing_data_rows, properties=properties
    )

    calculated_patches = self._form_calculated_task_property_patches(
        existing_data_rows=existing_data_rows, properties=properties
    )
    all_patches = update_patches + calculated_patches
    if len(new_values) > 0:
        self.update_property_on_task(task_id=task_id, patch_payload=all_patches)
        return self.add_properties_to_task(
            inventory_id=inventory_id,
            task_id=task_id,
            block_id=block_id,
            lot_id=lot_id,
            properties=new_values,
        )
    else:
        return self.update_property_on_task(task_id=task_id, patch_payload=all_patches)

update_property_on_inventory

update_property_on_inventory(
    *,
    inventory_id: InventoryId,
    property_data: InventoryDataColumn,
) -> InventoryPropertyData

Update a property on an inventory item.

Parameters:

Name Type Description Default
inventory_id InventoryId

The ID of the inventory item to update the property on.

required
property_data InventoryDataColumn

The updated property data.

required

Returns:

Type Description
InventoryPropertyData

The updated property data as returned by the server.

Source code in src/albert/collections/property_data.py
@validate_call
def update_property_on_inventory(
    self, *, inventory_id: InventoryId, property_data: InventoryDataColumn
) -> InventoryPropertyData:
    """Update a property on an inventory item.

    Parameters
    ----------
    inventory_id : InventoryId
        The ID of the inventory item to update the property on.
    property_data : InventoryDataColumn
        The updated property data.

    Returns
    -------
    InventoryPropertyData
        The updated property data as returned by the server.
    """
    existing_properties = self.get_properties_on_inventory(inventory_id=inventory_id)
    existing_value = None
    for p in existing_properties.custom_property_data:
        if p.data_column.data_column_id == property_data.data_column_id:
            existing_value = (
                p.data_column.property_data.value
                if p.data_column.property_data.value is not None
                else p.data_column.property_data.string_value
                if p.data_column.property_data.string_value is not None
                else str(p.data_column.property_data.numeric_value)
                if p.data_column.property_data.numeric_value is not None
                else None
            )
            existing_id = p.data_column.property_data.id
            break
    if existing_value is not None:
        payload = [
            PropertyDataPatchDatum(
                operation=PatchOperation.UPDATE,
                id=existing_id,
                attribute="value",
                new_value=property_data.value,
                old_value=existing_value,
            )
        ]
    else:
        payload = [
            PropertyDataPatchDatum(
                operation=PatchOperation.ADD,
                id=existing_id,
                attribute="value",
                new_value=property_data.value,
            )
        ]

    self.session.patch(
        url=f"{self.base_path}/{inventory_id}",
        json=[x.model_dump(exclude_none=True, by_alias=True, mode="json") for x in payload],
    )
    return self.get_properties_on_inventory(inventory_id=inventory_id)

update_property_on_task

update_property_on_task(
    *,
    task_id: TaskId,
    patch_payload: list[PropertyDataPatchDatum],
) -> list[TaskPropertyData]

Updates a specific property on a task.

Parameters:

Name Type Description Default
task_id TaskId

The ID of the task.

required
patch_payload list[PropertyDataPatchDatum]

The specific patch to make to update the property.

required

Returns:

Type Description
list[TaskPropertyData]

A list of TaskPropertyData objects representing the properties within the task.

Source code in src/albert/collections/property_data.py
@validate_call
def update_property_on_task(
    self, *, task_id: TaskId, patch_payload: list[PropertyDataPatchDatum]
) -> list[TaskPropertyData]:
    """Updates a specific property on a task.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task.
    patch_payload : list[PropertyDataPatchDatum]
        The specific patch to make to update the property.

    Returns
    -------
    list[TaskPropertyData]
        A list of TaskPropertyData objects representing the properties within the task.
    """
    if len(patch_payload) > 0:
        self.session.patch(
            url=f"{self.base_path}/{task_id}",
            json=[
                x.model_dump(exclude_none=True, by_alias=True, mode="json")
                for x in patch_payload
            ],
        )
    return self.get_all_task_properties(task_id=task_id)

ReportCollection

ReportCollection(*, session: AlbertSession)

Bases: BaseCollection

ReportCollection is a collection class for managing Report entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
get_datascience_report

Get a datascience report by its report type ID.

Source code in src/albert/collections/reports.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the ReportCollection with the provided session.

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

base_path instance-attribute

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

get_datascience_report

get_datascience_report(
    *,
    report_type_id: str,
    input_data: dict[str, Any] | None = None,
) -> ReportInfo

Get a datascience report by its report type ID.

Parameters:

Name Type Description Default
report_type_id str

The report type ID for the report.

required
input_data dict[str, Any] | None

Additional input data for generating the report (e.g., project IDs and unique IDs).

None

Returns:

Type Description
ReportInfo

The info for the report.

Examples:

>>> report = client.reports.get_datascience_report(
...     report_type_id="RET51",
...     input_data={
...         "projectId": ["PRO123"],
...         "uniqueId": ["DAT123_DAC123"]
...     }
... )
Source code in src/albert/collections/reports.py
def get_datascience_report(
    self,
    *,
    report_type_id: str,
    input_data: dict[str, Any] | None = None,
) -> ReportInfo:
    """Get a datascience report by its report type ID.

    Parameters
    ----------
    report_type_id : str
        The report type ID for the report.
    input_data : dict[str, Any] | None
        Additional input data for generating the report
        (e.g., project IDs and unique IDs).

    Returns
    -------
    ReportInfo
        The info for the report.

    Examples
    --------
    >>> report = client.reports.get_datascience_report(
    ...     report_type_id="RET51",
    ...     input_data={
    ...         "projectId": ["PRO123"],
    ...         "uniqueId": ["DAT123_DAC123"]
    ...     }
    ... )
    """
    path = f"{self.base_path}/datascience/{report_type_id}"

    params = {}
    input_data = input_data or {}
    for key, value in input_data.items():
        params[f"inputData[{key}]"] = value

    response = self.session.get(path, params=params)
    return ReportInfo(**response.json())

RoleCollection

RoleCollection(*, session: AlbertSession)

Bases: BaseCollection

RoleCollection is a collection class for managing Role entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Create a new role.

get_by_id

Retrieve a Role by its ID.

list

Lists the available Roles

Source code in src/albert/collections/roles.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the RoleCollection with the provided session.

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

base_path instance-attribute

base_path = f'/api/{_api_version}/acl/roles'

create

create(*, role: Role)

Create a new role.

Parameters:

Name Type Description Default
role Role

The role to create.

required
Source code in src/albert/collections/roles.py
def create(self, *, role: Role):
    """
    Create a new role.
    Parameters
    ----------
    role : Role
        The role to create.
    """
    response = self.session.post(
        self.base_path,
        json=role.model_dump(by_alias=True, exclude_none=True, mode="json"),
    )
    return Role(**response.json())

get_by_id

get_by_id(*, id: str) -> Role

Retrieve a Role by its ID.

Parameters:

Name Type Description Default
id str

The ID of the role.

required

Returns:

Type Description
Role

The retrieved role.

Source code in src/albert/collections/roles.py
def get_by_id(self, *, id: str) -> Role:
    """
    Retrieve a Role by its ID.
    Parameters
    ----------
    id : str
        The ID of the role.
    Returns
    -------
    Role
        The retrieved role.
    """
    # role IDs have # symbols
    url = urllib.parse.quote(f"{self.base_path}/{id}")
    response = self.session.get(url=url)
    return Role(**response.json())

list

list(*, params: dict | None = None) -> list[Role]

Lists the available Roles

Parameters:

Name Type Description Default
params dict

description, by default {}

None

Returns:

Type Description
List

List of available Roles

Source code in src/albert/collections/roles.py
def list(self, *, params: dict | None = None) -> list[Role]:
    """Lists the available Roles

    Parameters
    ----------
    params : dict, optional
        _description_, by default {}

    Returns
    -------
    List
        List of available Roles
    """
    if params is None:
        params = {}
    response = self.session.get(self.base_path, params=params)
    role_data = response.json().get("Items", [])
    return [Role(**r) for r in role_data]

StorageLocationsCollection

StorageLocationsCollection(*, session: AlbertSession)

Bases: BaseCollection

StorageLocationsCollection is a collection class for managing StorageLoction entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert Session information

required

Methods:

Name Description
create

Create a new storage location.

delete

Delete a storage location by its ID.

get_by_id

Get a storage location by its ID.

list

List storage locations with optional filtering.

update

Update a storage location.

Source code in src/albert/collections/storage_locations.py
def __init__(self, *, session: AlbertSession):
    """Initialize the StorageLocationsCollection.

    Parameters
    ----------
    session : AlbertSession
        The Albert Session information
    """
    super().__init__(session=session)
    self.base_path = f"/api/{StorageLocationsCollection._api_version}/storagelocations"

base_path instance-attribute

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

create

create(
    *, storage_location: StorageLocation
) -> StorageLocation

Create a new storage location.

Parameters:

Name Type Description Default
storage_location StorageLocation

The storage location to create.

required

Returns:

Type Description
StorageLocation

The created storage location.

Source code in src/albert/collections/storage_locations.py
def create(self, *, storage_location: StorageLocation) -> StorageLocation:
    """Create a new storage location.

    Parameters
    ----------
    storage_location : StorageLocation
        The storage location to create.

    Returns
    -------
    StorageLocation
        The created storage location.
    """
    matching = self.list(
        name=storage_location.name, location=storage_location.location, exact_match=True
    )
    for m in matching:
        if m.name.lower() == storage_location.name.lower():
            logging.warning(
                f"Storage location with name {storage_location.name} already exists, returning existing."
            )
            return m

    path = self.base_path
    response = self.session.post(
        path, json=storage_location.model_dump(by_alias=True, exclude_none=True, mode="json")
    )
    return StorageLocation(**response.json())

delete

delete(*, id: str) -> None

Delete a storage location by its ID.

Parameters:

Name Type Description Default
id str

The ID of the storage location to delete.

required
Source code in src/albert/collections/storage_locations.py
def delete(self, *, id: str) -> None:
    """Delete a storage location by its ID.

    Parameters
    ----------
    id : str
        The ID of the storage location to delete.
    """
    path = f"{self.base_path}/{id}"
    self.session.delete(path)

get_by_id

get_by_id(*, id: str) -> StorageLocation

Get a storage location by its ID.

Parameters:

Name Type Description Default
id str

The ID of the storage location to retrieve.

required

Returns:

Type Description
StorageLocation

The retrieved storage location with the given ID.

Source code in src/albert/collections/storage_locations.py
def get_by_id(self, *, id: str) -> StorageLocation:
    """Get a storage location by its ID.

    Parameters
    ----------
    id : str
        The ID of the storage location to retrieve.

    Returns
    -------
    StorageLocation
        The retrieved storage location with the given ID.
    """
    path = f"{self.base_path}/{id}"
    response = self.session.get(path)
    return StorageLocation(**response.json())

list

list(
    *,
    name: str | list[str] | None = None,
    exact_match: bool = False,
    location: str | Location | None = None,
    start_key: str | None = None,
    limit: int = 50,
) -> Generator[StorageLocation, None, None]

List storage locations with optional filtering.

Parameters:

Name Type Description Default
name str | list[str] | None

The name or names of the storage locations to filter by, by default None

None
exact_match bool

Whether to perform an exact match on the name, by default False

False
location str | Location | None

The location ID or Location object to filter by, by default None

None

Yields:

Type Description
Generator[StorageLocation, None, None]

description

Source code in src/albert/collections/storage_locations.py
def list(
    self,
    *,
    name: str | list[str] | None = None,
    exact_match: bool = False,
    location: str | Location | None = None,
    start_key: str | None = None,
    limit: int = 50,
) -> Generator[StorageLocation, None, None]:
    """List storage locations with optional filtering.

    Parameters
    ----------
    name : str | list[str] | None, optional
        The name or names of the storage locations to filter by, by default None
    exact_match : bool, optional
        Whether to perform an exact match on the name, by default False
    location : str | Location | None, optional
        The location ID or Location object to filter by, by default None

    Yields
    ------
    Generator[StorageLocation, None, None]
        _description_
    """

    def deserialize(items: list[dict]) -> Iterator[StorageLocation]:
        for x in items:
            id = x["albertId"]
            try:
                yield self.get_by_id(id=id)
            except AlbertHTTPError as e:
                logger.warning(f"Error fetching storage location {id}: {e}")

    params = {
        "limit": limit,
        "locationId": location.id if isinstance(location, Location | EntityLink) else location,
        "startKey": start_key,
    }
    if name:
        params["name"] = [name] if isinstance(name, str) else name
        params["exactMatch"] = json.dumps(exact_match)

    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=deserialize,
    )

update

update(
    *, storage_location: StorageLocation
) -> StorageLocation

Update a storage location.

Parameters:

Name Type Description Default
storage_location StorageLocation

The storage location to update.

required

Returns:

Type Description
StorageLocation

The updated storage location as returned by the server.

Source code in src/albert/collections/storage_locations.py
def update(self, *, storage_location: StorageLocation) -> StorageLocation:
    """Update a storage location.

    Parameters
    ----------
    storage_location : StorageLocation
        The storage location to update.

    Returns
    -------
    StorageLocation
        The updated storage location as returned by the server.
    """
    path = f"{self.base_path}/{storage_location.id}"
    payload = self._generate_patch_payload(
        existing=self.get_by_id(id=storage_location.id),
        updated=storage_location,
    )
    self.session.patch(path, json=payload.model_dump(mode="json", by_alias=True))
    return self.get_by_id(id=storage_location.id)

SubstanceCollection

SubstanceCollection(*, session: AlbertSession)

Bases: BaseCollection

SubstanceCollection is a collection class for managing Substance entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

An instance of the Albert session used for API interactions.

required

Attributes:

Name Type Description
base_path str

The base URL for making API requests related to substances.

Methods:

Name Description
get_by_ids

Retrieves a list of substances by their CAS IDs and optional region.

get_by_id

Retrieves a single substance by its CAS ID and optional region.

Source code in src/albert/collections/substance.py
def __init__(self, *, session: AlbertSession):
    super().__init__(session=session)
    self.base_path = f"/api/{SubstanceCollection._api_version}/substances"

base_path instance-attribute

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

get_by_id

get_by_id(
    *,
    cas_id: str,
    region: str = "US",
    catch_errors: bool | None = None,
) -> SubstanceInfo

Get a substance by its CAS ID.

Parameters:

Name Type Description Default
cas_id str

The CAS ID of the substance to retrieve.

required
region str

The region to filter the substance by, by default "US".

'US'
catch_errors bool

Whether to catch errors for unknown CAS, by default False.

None

Returns:

Type Description
SubstanceInfo

The retrieved substance or raises an error if not found.

Source code in src/albert/collections/substance.py
def get_by_id(
    self,
    *,
    cas_id: str,
    region: str = "US",
    catch_errors: bool | None = None,
) -> SubstanceInfo:
    """
    Get a substance by its CAS ID.

    Parameters
    ----------
    cas_id : str
        The CAS ID of the substance to retrieve.
    region : str, optional
        The region to filter the substance by, by default "US".
    catch_errors : bool, optional
        Whether to catch errors for unknown CAS, by default False.

    Returns
    -------
    SubstanceInfo
        The retrieved substance or raises an error if not found.
    """
    return self.get_by_ids(cas_ids=[cas_id], region=region, catch_errors=catch_errors)[0]

get_by_ids

get_by_ids(
    *,
    cas_ids: list[str],
    region: str = "US",
    catch_errors: bool | None = None,
) -> list[SubstanceInfo]

Get substances by their CAS IDs.

If catch_errors is set to False, the number of substances returned may be less than the number of CAS IDs provided if any of the CAS IDs result in an error.

Parameters:

Name Type Description Default
cas_ids list[str]

A list of CAS IDs to retrieve substances for.

required
region str

The region to filter the subastance by, by default "US"

'US'
catch_errors bool

Whether to catch errors for unknown CAS, by default True.

None

Returns:

Type Description
list[SubstanceInfo]

A list of substances with the given CAS IDs.

Source code in src/albert/collections/substance.py
def get_by_ids(
    self,
    *,
    cas_ids: list[str],
    region: str = "US",
    catch_errors: bool | None = None,
) -> list[SubstanceInfo]:
    """Get substances by their CAS IDs.

    If `catch_errors` is set to False, the number of substances returned
    may be less than the number of CAS IDs provided if any of the CAS IDs result in an error.

    Parameters
    ----------
    cas_ids : list[str]
        A list of CAS IDs to retrieve substances for.
    region : str, optional
        The region to filter the subastance by, by default "US"
    catch_errors : bool, optional
        Whether to catch errors for unknown CAS, by default True.

    Returns
    -------
    list[SubstanceInfo]
        A list of substances with the given CAS IDs.
    """
    params = {
        "casIDs": ",".join(cas_ids),
        "region": region,
        "catchErrors": json.dumps(catch_errors) if catch_errors is not None else None,
    }
    params = {k: v for k, v in params.items() if v is not None}
    response = self.session.get(self.base_path, params=params)
    return SubstanceResponse.model_validate(response.json()).substances

TagCollection

TagCollection(*, session: AlbertSession)

Bases: BaseCollection

TagCollection is a collection class for managing Tag entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Attributes:

Name Type Description
base_path str

The base URL for tag API requests.

Methods:

Name Description
list

Lists tag entities with optional filters.

tag_exists

Checks if a tag exists by its name.

create

Creates a new tag entity.

get_by_id

Retrieves a tag by its ID.

get_by_ids

Retrieve a list of tags by their IDs.

get_by_tag

Retrieves a tag by its name.

delete

Deletes a tag by its ID.

rename

Renames an existing tag entity.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required
Source code in src/albert/collections/tags.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the TagCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, tag: str | Tag) -> Tag

Creates a new tag entity if the given tag does not exist.

Parameters:

Name Type Description Default
tag Union[str, Tag]

The tag name or Tag object to create.

required

Returns:

Type Description
Tag

The created Tag object or the existing Tag object of it already exists.

Source code in src/albert/collections/tags.py
def create(self, *, tag: str | Tag) -> Tag:
    """
    Creates a new tag entity if the given tag does not exist.

    Parameters
    ----------
    tag : Union[str, Tag]
        The tag name or Tag object to create.

    Returns
    -------
    Tag
        The created Tag object or the existing Tag object of it already exists.
    """
    if isinstance(tag, str):
        tag = Tag(tag=tag)
    hit = self.get_by_tag(tag=tag.tag, exact_match=True)
    if hit is not None:
        logging.warning(f"Tag {hit.tag} already exists with id {hit.id}")
        return hit
    payload = {"name": tag.tag}
    response = self.session.post(self.base_path, json=payload)
    tag = Tag(**response.json())
    return tag

delete

delete(*, id: str) -> None

Deletes a tag by its ID.

Parameters:

Name Type Description Default
id str

The ID of the tag to delete.

required

Returns:

Type Description
None
Source code in src/albert/collections/tags.py
def delete(self, *, id: str) -> None:
    """
    Deletes a tag by its ID.

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

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

get_by_id

get_by_id(*, id: str) -> Tag

Get a tag by its ID.

Parameters:

Name Type Description Default
id str

The ID of the tag to get.

required

Returns:

Type Description
Tag

The Tag object.

Source code in src/albert/collections/tags.py
def get_by_id(self, *, id: str) -> Tag:
    """
    Get a tag by its ID.

    Parameters
    ----------
    id : str
        The ID of the tag to get.

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

get_by_ids

get_by_ids(*, ids: list[str]) -> list[Tag]
Source code in src/albert/collections/tags.py
def get_by_ids(self, *, ids: list[str]) -> list[Tag]:
    url = f"{self.base_path}/ids"
    batches = [ids[i : i + 100] for i in range(0, len(ids), 100)]
    return [
        Tag(**item)
        for batch in batches
        for item in self.session.get(url, params={"id": batch}).json()
    ]

get_by_tag

get_by_tag(
    *, tag: str, exact_match: bool = True
) -> Tag | None

Retrieves a tag by its name of None if not found.

Parameters:

Name Type Description Default
tag str

The name of the tag to retrieve.

required
exact_match bool

Whether to match the name exactly, by default True.

True

Returns:

Type Description
Tag

The Tag object if found, None otherwise.

Source code in src/albert/collections/tags.py
def get_by_tag(self, *, tag: str, exact_match: bool = True) -> Tag | None:
    """
    Retrieves a tag by its name of None if not found.

    Parameters
    ----------
    tag : str
        The name of the tag to retrieve.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.

    Returns
    -------
    Tag
        The Tag object if found, None otherwise.
    """
    found = self.list(name=tag, exact_match=exact_match)
    return next(found, None)

list

list(
    *,
    limit: int = 50,
    order_by: OrderBy = DESCENDING,
    name: str | list[str] | None = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Tag]

Lists Tag entities with optional filters.

Parameters:

Name Type Description Default
limit int

The maximum number of tags to return, by default 50.

50
order_by OrderBy

The order by which to sort the results, by default OrderBy.DESCENDING.

DESCENDING
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.

True
start_key Optional[str]

The starting point for the next set of results, by default None.

None

Returns:

Type Description
Iterator[Tag]

An iterator of Tag objects.

Source code in src/albert/collections/tags.py
def list(
    self,
    *,
    limit: int = 50,
    order_by: OrderBy = OrderBy.DESCENDING,
    name: str | list[str] | None = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Tag]:
    """
    Lists Tag entities with optional filters.

    Parameters
    ----------
    limit : int, optional
        The maximum number of tags to return, by default 50.
    order_by : OrderBy, optional
        The order by which to sort the results, by default OrderBy.DESCENDING.
    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.
    start_key : Optional[str], optional
        The starting point for the next set of results, by default None.

    Returns
    -------
    Iterator[Tag]
        An iterator of Tag objects.
    """
    params = {"limit": limit, "orderBy": order_by.value, "startKey": start_key}
    if name:
        params["name"] = [name] if isinstance(name, str) else name
        params["exactMatch"] = json.dumps(exact_match)
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [Tag(**item) for item in items],
    )

rename

rename(*, old_name: str, new_name: str) -> Tag

Renames an existing tag entity.

Parameters:

Name Type Description Default
old_name str

The current name of the tag.

required
new_name str

The new name of the tag.

required

Returns:

Type Description
Tag

The renamed Tag.

Source code in src/albert/collections/tags.py
def rename(self, *, old_name: str, new_name: str) -> Tag:
    """
    Renames an existing tag entity.

    Parameters
    ----------
    old_name : str
        The current name of the tag.
    new_name : str
        The new name of the tag.

    Returns
    -------
    Tag
        The renamed Tag.
    """
    found_tag = self.get_by_tag(tag=old_name, exact_match=True)
    if not found_tag:
        msg = f'Tag "{old_name}" not found.'
        logger.error(msg)
        raise AlbertException(msg)
    tag_id = found_tag.id
    payload = [
        {
            "data": [
                {
                    "operation": "update",
                    "attribute": "name",
                    "oldValue": old_name,
                    "newValue": new_name,
                }
            ],
            "id": tag_id,
        }
    ]
    self.session.patch(self.base_path, json=payload)
    return self.get_by_id(id=tag_id)

tag_exists

tag_exists(*, tag: str, exact_match: bool = True) -> bool

Checks if a tag exists by its name.

Parameters:

Name Type Description Default
tag str

The name of the tag to check.

required
exact_match bool

Whether to match the name exactly, by default True.

True

Returns:

Type Description
bool

True if the tag exists, False otherwise.

Source code in src/albert/collections/tags.py
def tag_exists(self, *, tag: str, exact_match: bool = True) -> bool:
    """
    Checks if a tag exists by its name.

    Parameters
    ----------
    tag : str
        The name of the tag to check.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.

    Returns
    -------
    bool
        True if the tag exists, False otherwise.
    """

    return self.get_by_tag(tag=tag, exact_match=exact_match) is not None

TaskCollection

TaskCollection(*, session: AlbertSession)

Bases: BaseCollection

TaskCollection is a collection class for managing Task entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert Session information

required

Methods:

Name Description
add_block

Add a block to a Property task.

create

Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.

delete

Delete a task.

get_by_id

Retrieve a task by its ID.

get_history
list

Search for tasks matching the given criteria.

remove_block

Remove a block from a Property task.

update

Update a task.

update_block_workflow

Update the workflow of a specific block within a task.

Source code in src/albert/collections/tasks.py
def __init__(self, *, session: AlbertSession):
    """Initialize the TaskCollection.

    Parameters
    ----------
    session : AlbertSession
        The Albert Session information
    """
    super().__init__(session=session)
    self.base_path = f"/api/{TaskCollection._api_version}/tasks"

base_path instance-attribute

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

add_block

add_block(
    *,
    task_id: TaskId,
    data_template_id: DataTemplateId,
    workflow_id: WorkflowId,
) -> None

Add a block to a Property task.

Parameters:

Name Type Description Default
task_id TaskId

The ID of the task to add the block to.

required
data_template_id DataTemplateId

The ID of the data template to use for the block.

required
workflow_id WorkflowId

The ID of the workflow to assign to the block.

required

Returns:

Type Description
None

This method does not return any value.

Source code in src/albert/collections/tasks.py
@validate_call
def add_block(
    self, *, task_id: TaskId, data_template_id: DataTemplateId, workflow_id: WorkflowId
) -> None:
    """Add a block to a Property task.

    Parameters
    ----------
    task_id : TaskId
        The ID of the task to add the block to.
    data_template_id : DataTemplateId
        The ID of the data template to use for the block.
    workflow_id : WorkflowId
        The ID of the workflow to assign to the block.

    Returns
    -------
    None
        This method does not return any value.

    """
    url = f"{self.base_path}/{task_id}"
    payload = [
        {
            "id": task_id,
            "data": [
                {
                    "operation": "add",
                    "attribute": "Block",
                    "newValue": [{"datId": data_template_id, "Workflow": {"id": workflow_id}}],
                }
            ],
        }
    ]
    self.session.patch(url=url, json=payload)
    return None

create

create(*, task: BaseTask) -> BaseTask

Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.

Parameters:

Name Type Description Default
task BaseTask

The task object to create.

required

Returns:

Type Description
BaseTask

The registered task object.

Source code in src/albert/collections/tasks.py
def create(self, *, task: BaseTask) -> BaseTask:
    """Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.

    Parameters
    ----------
    task : BaseTask
        The task object to create.

    Returns
    -------
    BaseTask
        The registered task object.
    """
    payload = [task.model_dump(mode="json", by_alias=True, exclude_none=True)]
    url = f"{self.base_path}/multi?category={task.category.value}"
    if task.parent_id is not None:
        url = f"{url}&parentId={task.parent_id}"
    response = self.session.post(url=url, json=payload)
    task_data = response.json()[0]
    return TaskAdapter.validate_python(task_data)

delete

delete(*, id: TaskId) -> None

Delete a task.

Parameters:

Name Type Description Default
id TaskId

The ID of the task to delete.

required
Source code in src/albert/collections/tasks.py
@validate_call
def delete(self, *, id: TaskId) -> None:
    """Delete a task.

    Parameters
    ----------
    id : TaskId
        The ID of the task to delete.
    """
    url = f"{self.base_path}/{id}"
    self.session.delete(url)

get_by_id

get_by_id(*, id: TaskId) -> BaseTask

Retrieve a task by its ID.

Parameters:

Name Type Description Default
id TaskId

The ID of the task to retrieve.

required

Returns:

Type Description
BaseTask

The task object with the provided ID.

Source code in src/albert/collections/tasks.py
@validate_call
def get_by_id(self, *, id: TaskId) -> BaseTask:
    """Retrieve a task by its ID.

    Parameters
    ----------
    id : TaskId
        The ID of the task to retrieve.

    Returns
    -------
    BaseTask
        The task object with the provided ID.
    """
    url = f"{self.base_path}/multi/{id}"
    response = self.session.get(url)
    return TaskAdapter.validate_python(response.json())

get_history

get_history(
    *,
    id: TaskId,
    order: OrderBy = DESCENDING,
    limit: int = 1000,
    entity: HistoryEntity | None = None,
    blockId: str | None = None,
    startKey: str | None = None,
) -> TaskHistory
Source code in src/albert/collections/tasks.py
def get_history(
    self,
    *,
    id: TaskId,
    order: OrderBy = OrderBy.DESCENDING,
    limit: int = 1000,
    entity: HistoryEntity | None = None,
    blockId: str | None = None,
    startKey: str | None = None,
) -> TaskHistory:
    params = {
        "limit": limit,
        "orderBy": OrderBy(order).value if order else None,
        "entity": entity,
        "blockId": blockId,
        "startKey": startKey,
    }
    url = f"{self.base_path}/{id}/history"
    response = self.session.get(url, params=params)
    return TaskHistory(**response.json())

list

list(
    *,
    order: OrderBy = DESCENDING,
    text: str | None = None,
    sort_by: str | None = None,
    tags: list[str] | None = None,
    task_id: list[str] | None = None,
    linked_task: list[str] | None = None,
    category: TaskCategory | None = None,
    albert_id: list[str] | None = None,
    data_template: list[str] | None = None,
    assigned_to: list[str] | None = None,
    location: list[str] | None = None,
    priority: list[str] | None = None,
    status: list[str] | None = None,
    parameter_group: list[str] | None = None,
    created_by: list[str] | None = None,
    project_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[BaseTask]

Search for tasks matching the given criteria.

Parameters:

Name Type Description Default
order OrderBy

The order in which to return results, by default OrderBy.DESCENDING

DESCENDING
text str | None

The text to search for, by default None

None
sort_by str | None

The attribute to sort by, by default None

None
tags list[str] | None

The tags to search for, by default None

None
task_id list[str] | None

The related task IDs to search for, by default None

None
linked_task list[str] | None

The Linked Task IDs to search for, by default None

None
category TaskCategory | None

The category of the task to search for, by default None

None
albert_id list[str] | None

The Albert IDs to search for, by default None

None
data_template list[str] | None

The data template IDs to search for, by default None

None
assigned_to list[str] | None

The User IDs to search for, by default None

None
location list[str] | None

The Locations names to search for, by default None

None
priority list[str] | None

The Priority levels to search for, by default None

None
status list[str] | None

The Task Statuses to search for, by default None

None
parameter_group list[str] | None

The related Parameter Group IDs to search for, by default None

None
created_by list[str] | None

The User IDs of the task creators to search for, by default None

None
project_id str | None

The Project ID to search for, by default None

None

Yields:

Type Description
Iterator[BaseTask]

An iterator of matching Task objects.

Source code in src/albert/collections/tasks.py
def list(
    self,
    *,
    order: OrderBy = OrderBy.DESCENDING,
    text: str | None = None,
    sort_by: str | None = None,
    tags: list[str] | None = None,
    task_id: list[str] | None = None,
    linked_task: list[str] | None = None,
    category: TaskCategory | None = None,
    albert_id: list[str] | None = None,
    data_template: list[str] | None = None,
    assigned_to: list[str] | None = None,
    location: list[str] | None = None,
    priority: list[str] | None = None,
    status: list[str] | None = None,
    parameter_group: list[str] | None = None,
    created_by: list[str] | None = None,
    project_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[BaseTask]:
    """Search for tasks matching the given criteria.

    Parameters
    ----------
    order : OrderBy, optional
        The order in which to return results, by default OrderBy.DESCENDING
    text : str | None, optional
        The text to search for, by default None
    sort_by : str | None, optional
        The attribute to sort by, by default None
    tags : list[str] | None, optional
        The tags to search for, by default None
    task_id : list[str] | None, optional
        The related task IDs to search for, by default None
    linked_task : list[str] | None, optional
        The Linked Task IDs to search for, by default None
    category : TaskCategory | None, optional
        The category of the task to search for, by default None
    albert_id : list[str] | None, optional
        The Albert IDs to search for, by default None
    data_template : list[str] | None, optional
        The data template IDs to search for, by default None
    assigned_to : list[str] | None, optional
        The User IDs to search for, by default None
    location : list[str] | None, optional
        The Locations names to search for, by default None
    priority : list[str] | None, optional
        The Priority levels to search for, by default None
    status : list[str] | None, optional
        The Task Statuses to search for, by default None
    parameter_group : list[str] | None, optional
        The related Parameter Group IDs to search for, by default None
    created_by : list[str] | None, optional
        The User IDs of the task creators to search for, by default None
    project_id : str | None, optional
        The Project ID to search for, by default None

    Yields
    ------
    Iterator[BaseTask]
        An iterator of matching Task objects.
    """

    def deserialize(items: list[dict]) -> Iterator[BaseTask]:
        for item in items:
            id = item["albertId"]
            try:
                yield self.get_by_id(id=id)
            except (
                AlbertHTTPError,
                RetryError,
            ) as e:  # some legacy poorly formed Tasks raise 500s. The allowance on Retry error to also ignore these.
                logger.warning(f"Error fetching task '{id}': {e}")

    params = {
        "limit": limit,
        "offset": offset,
        "order": OrderBy(order).value if order else None,
        "text": text,
        "sortBy": sort_by,
        "tags": tags,
        "taskId": task_id,
        "linkedTask": linked_task,
        "category": category,
        "albertId": albert_id,
        "dataTemplate": data_template,
        "assignedTo": assigned_to,
        "location": location,
        "priority": priority,
        "status": status,
        "parameterGroup": parameter_group,
        "createdBy": created_by,
        "projectId": project_id,
    }

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        deserialize=deserialize,
        params=params,
    )

remove_block

remove_block(*, task_id: TaskId, block_id: BlockId) -> None

Remove a block from a Property task.

Parameters:

Name Type Description Default
task_id str

ID of the Task to remove the block from (e.g., TASFOR1234)

required
block_id str

ID of the Block to remove (e.g., BLK1)

required

Returns:

Type Description
None
Source code in src/albert/collections/tasks.py
@validate_call
def remove_block(self, *, task_id: TaskId, block_id: BlockId) -> None:
    """Remove a block from a Property task.

    Parameters
    ----------
    task_id : str
        ID of the Task to remove the block from (e.g., TASFOR1234)
    block_id : str
        ID of the Block to remove (e.g., BLK1)

    Returns
    -------
    None
    """
    url = f"{self.base_path}/{task_id}"
    payload = [
        {
            "id": task_id,
            "data": [
                {
                    "operation": "delete",
                    "attribute": "Block",
                    "oldValue": [block_id],
                }
            ],
        }
    ]
    self.session.patch(url=url, json=payload)
    return None

update

update(*, task: BaseTask) -> BaseTask

Update a task.

Parameters:

Name Type Description Default
task BaseTask

The updated Task object.

required

Returns:

Type Description
BaseTask

The updated Task object as it exists in the Albert platform.

Source code in src/albert/collections/tasks.py
def update(self, *, task: BaseTask) -> BaseTask:
    """Update a task.

    Parameters
    ----------
    task : BaseTask
        The updated Task object.

    Returns
    -------
    BaseTask
        The updated Task object as it exists in the Albert platform.
    """
    existing = self.get_by_id(id=task.id)
    patch_payload, list_metadata_updates = self._generate_adv_patch_payload(
        updated=task, existing=existing
    )
    patch_operations = patch_payload.get("data", [])

    if len(patch_operations) == 0 and len(list_metadata_updates) == 0:
        logger.info(f"Task {task.id} is already up to date")
        return task
    path = f"{self.base_path}/{task.id}"

    for datum in patch_operations:
        patch_payload = TaskPatchPayload(data=[datum], id=task.id)
        self.session.patch(
            url=path,
            json=[patch_payload.model_dump(mode="json", by_alias=True, exclude_none=True)],
        )

    # For metadata list field updates, we clear, then update
    # since duplicate attribute values are not allowed in single patch request.
    for attribute, values in list_metadata_updates.items():
        entity_links = existing.metadata.get(attribute.split(".")[1])
        old_values = [item.id if hasattr(item, "id") else item for item in entity_links]
        clear_datum = PatchDatum(
            operation=PatchOperation.DELETE, attribute=attribute, oldValue=old_values
        )
        clear_payload = TaskPatchPayload(data=[clear_datum], id=task.id)
        self.session.patch(
            url=path,
            json=[clear_payload.model_dump(mode="json", by_alias=True, exclude_none=True)],
        )
        if values:
            update_datum = PatchDatum(
                operation=PatchOperation.UPDATE,
                attribute=attribute,
                newValue=values,
                oldValue=[],
            )

            update_payload = TaskPatchPayload(data=[update_datum], id=task.id)
            self.session.patch(
                url=path,
                json=[
                    update_payload.model_dump(mode="json", by_alias=True, exclude_none=False)
                ],
            )
    return self.get_by_id(id=task.id)

update_block_workflow

update_block_workflow(
    *,
    task_id: TaskId,
    block_id: BlockId,
    workflow_id: WorkflowId,
) -> None

Update the workflow of a specific block within a task.

This method updates the workflow of a specified block within a task.

Parameters:

Name Type Description Default
task_id str

The ID of the task.

required
block_id str

The ID of the block within the task.

required
workflow_id str

The ID of the new workflow to be assigned to the block.

required

Returns:

Type Description
None

This method does not return any value.

Notes
  • The method asserts that the retrieved task is an instance of PropertyTask.
  • If the block's current workflow matches the new workflow ID, no update is performed.
  • The method handles the case where the block has a default workflow named "No Parameter Group".
Source code in src/albert/collections/tasks.py
@validate_call
def update_block_workflow(
    self, *, task_id: TaskId, block_id: BlockId, workflow_id: WorkflowId
) -> None:
    """
    Update the workflow of a specific block within a task.

    This method updates the workflow of a specified block within a task.
    Parameters
    ----------
    task_id : str
        The ID of the task.
    block_id : str
        The ID of the block within the task.
    workflow_id : str
        The ID of the new workflow to be assigned to the block.

    Returns
    -------
    None
        This method does not return any value.

    Notes
    -----
    - The method asserts that the retrieved task is an instance of `PropertyTask`.
    - If the block's current workflow matches the new workflow ID, no update is performed.
    - The method handles the case where the block has a default workflow named "No Parameter Group".
    """
    url = f"{self.base_path}/{task_id}"
    task = self.get_by_id(id=task_id)
    if not isinstance(task, PropertyTask):
        logger.error(f"Task {task_id} is not an instance of PropertyTask")
        raise TypeError(f"Task {task_id} is not an instance of PropertyTask")
    for b in task.blocks:
        if b.id != block_id:
            continue
        for w in b.workflow:
            if w.name == "No Parameter Group" and len(b.workflow) > 1:
                # hardcoded default workflow
                continue
            existing_workflow_id = w.id
    if existing_workflow_id == workflow_id:
        logger.info(f"Block {block_id} already has workflow {workflow_id}")
        return None
    patch = [
        {
            "data": [
                {
                    "operation": "update",
                    "attribute": "workflow",
                    "oldValue": existing_workflow_id,
                    "newValue": workflow_id,
                    "blockId": block_id,
                }
            ],
            "id": task_id,
        }
    ]
    self.session.patch(url=url, json=patch)
    return None

UnNumberCollection

UnNumberCollection(*, session: AlbertSession)

Bases: BaseCollection

UnNumberCollection is a collection class for managing UnNumber entities in the Albert platform.

Note

Creating UN Numbers is not supported via the SDK, as UN Numbers are highly controlled by Albert.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

This method is not implemented as UN Numbers cannot be created through the SDK.

get_by_id

Retrieve a UN Number by its ID.

get_by_name

Retrieve a UN Number by its name.

list

List UN Numbers matching the provided criteria.

Source code in src/albert/collections/un_numbers.py
def __init__(self, *, session: AlbertSession):
    """Initializes the UnNumberCollection with the provided session.

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

base_path instance-attribute

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

create

create() -> None

This method is not implemented as UN Numbers cannot be created through the SDK.

Source code in src/albert/collections/un_numbers.py
def create(self) -> None:
    """
    This method is not implemented as UN Numbers cannot be created through the SDK.
    """
    raise NotImplementedError()

get_by_id

get_by_id(*, id: str) -> UnNumber

Retrieve a UN Number by its ID.

Parameters:

Name Type Description Default
id str

The ID of the UN Number to retrieve.

required

Returns:

Type Description
UnNumber

The corresponding UN Number

Source code in src/albert/collections/un_numbers.py
def get_by_id(self, *, id: str) -> UnNumber:
    """Retrieve a UN Number by its ID.

    Parameters
    ----------
    id : str
        The ID of the UN Number to retrieve.

    Returns
    -------
    UnNumber
        The corresponding UN Number
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    return UnNumber(**response.json())

get_by_name

get_by_name(*, name: str) -> UnNumber | None

Retrieve a UN Number by its name.

Parameters:

Name Type Description Default
name str

The name of the UN Number to retrieve

required

Returns:

Type Description
UnNumber | None

The corresponding UN Number or None if not found

Source code in src/albert/collections/un_numbers.py
def get_by_name(self, *, name: str) -> UnNumber | None:
    """Retrieve a UN Number by its name.

    Parameters
    ----------
    name : str
        The name of the UN Number to retrieve

    Returns
    -------
    UnNumber | None
        The corresponding UN Number or None if not found
    """
    found = self.list(exact_match=True, name=name)
    return next(found, None)

list

list(
    *,
    name: str | None = None,
    exact_match: bool = False,
    limit: int = 50,
    start_key: str | None = None,
) -> Iterator[UnNumber]

List UN Numbers matching the provided criteria.

Parameters:

Name Type Description Default
name str | None

The name of the UN Number to search for, by default None

None
exact_match bool

Weather to return exact matches only, by default False

False

Yields:

Type Description
Iterator[UnNumber]

The UN Numbers matching the search criteria

Source code in src/albert/collections/un_numbers.py
def list(
    self,
    *,
    name: str | None = None,
    exact_match: bool = False,
    limit: int = 50,
    start_key: str | None = None,
) -> Iterator[UnNumber]:
    """List UN Numbers matching the provided criteria.

    Parameters
    ----------
    name : str | None, optional
        The name of the UN Number to search for, by default None
    exact_match : bool, optional
        Weather to return exact matches only, by default False

    Yields
    ------
    Iterator[UnNumber]
        The UN Numbers matching the search criteria
    """
    params = {"limit": limit, "startKey": start_key}
    if name:
        params["name"] = name
        params["exactMatch"] = json.dumps(exact_match)
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [UnNumber(**item) for item in items],
    )

UnitCollection

UnitCollection(*, session: AlbertSession)

Bases: BaseCollection

UnitCollection is a collection class for managing Unit entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Creates a new unit entity.

delete

Deletes a unit by its ID.

get_by_id

Retrieves a unit by its ID.

get_by_ids

Retrieves a set of units by their IDs

get_by_name

Retrieves a unit by its name.

list

Lists unit entities with optional filters.

unit_exists

Checks if a unit exists by its name.

update

Updates a unit entity by its ID.

Source code in src/albert/collections/units.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the UnitCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, unit: Unit) -> Unit

Creates a new unit entity.

Parameters:

Name Type Description Default
unit Unit

The unit object to create.

required

Returns:

Type Description
Unit

The created Unit object.

Source code in src/albert/collections/units.py
def create(self, *, unit: Unit) -> Unit:
    """
    Creates a new unit entity.

    Parameters
    ----------
    unit : Unit
        The unit object to create.

    Returns
    -------
    Unit
        The created Unit object.
    """
    hit = self.get_by_name(name=unit.name, exact_match=True)
    if hit is not None:
        logging.warning(
            f"Unit with the name {hit.name} already exists. Returning the existing unit."
        )
        return hit
    response = self.session.post(
        self.base_path, json=unit.model_dump(by_alias=True, exclude_unset=True, mode="json")
    )
    this_unit = Unit(**response.json())
    return this_unit

delete

delete(*, id: str) -> None

Deletes a unit by its ID.

Parameters:

Name Type Description Default
id str

The ID of the unit to delete.

required

Returns:

Type Description
None
Source code in src/albert/collections/units.py
def delete(self, *, id: str) -> None:
    """
    Deletes a unit by its ID.

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

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

get_by_id

get_by_id(*, id: str) -> Unit

Retrieves a unit by its ID.

Parameters:

Name Type Description Default
id str

The ID of the unit to retrieve.

required

Returns:

Type Description
Unit

The Unit object if found.

Source code in src/albert/collections/units.py
def get_by_id(self, *, id: str) -> Unit:
    """
    Retrieves a unit by its ID.

    Parameters
    ----------
    id : str
        The ID of the unit to retrieve.

    Returns
    -------
    Unit
        The Unit object if found.
    """
    url = f"{self.base_path}/{id}"
    response = self.session.get(url)
    this_unit = Unit(**response.json())
    return this_unit

get_by_ids

get_by_ids(*, ids: list[str]) -> list[Unit]

Retrieves a set of units by their IDs

Parameters:

Name Type Description Default
ids list[str]

The IDs of the units to retrieve.

required

Returns:

Type Description
list[Unit]

The Unit objects

Source code in src/albert/collections/units.py
def get_by_ids(self, *, ids: list[str]) -> list[Unit]:
    """
    Retrieves a set of units by their IDs

    Parameters
    ----------
    ids : list[str]
        The IDs of the units to retrieve.

    Returns
    -------
    list[Unit]
        The Unit objects
    """
    url = f"{self.base_path}/ids"
    batches = [ids[i : i + 500] for i in range(0, len(ids), 500)]
    return [
        Unit(**item)
        for batch in batches
        for item in self.session.get(url, params={"id": batch}).json()["Items"]
    ]

get_by_name

get_by_name(
    *, name: str, exact_match: bool = False
) -> Unit | None

Retrieves a unit by its name.

Parameters:

Name Type Description Default
name str

The name of the unit to retrieve.

required
exact_match bool

Whether to match the name exactly, by default False.

False

Returns:

Type Description
Optional[Unit]

The Unit object if found, None otherwise.

Source code in src/albert/collections/units.py
def get_by_name(self, *, name: str, exact_match: bool = False) -> Unit | None:
    """
    Retrieves a unit by its name.

    Parameters
    ----------
    name : str
        The name of the unit to retrieve.
    exact_match : bool, optional
        Whether to match the name exactly, by default False.

    Returns
    -------
    Optional[Unit]
        The Unit object if found, None otherwise.
    """
    found = self.list(name=name, exact_match=exact_match)
    return next(found, None)

list

list(
    *,
    limit: int = 100,
    name: str | list[str] | None = None,
    category: UnitCategory | None = None,
    order_by: OrderBy = DESCENDING,
    exact_match: bool = False,
    start_key: str | None = None,
    verified: bool | None = None,
) -> Iterator[Unit]

Lists unit entities with optional filters.

Parameters:

Name Type Description Default
limit int

The maximum number of units to return, by default 50.

100
name Optional[str]

The name of the unit to filter by, by default None.

None
category Optional[UnitCategory]

The category of the unit to filter by, by default None.

None
order_by OrderBy

The order by which to sort the results, by default OrderBy.DESCENDING.

DESCENDING
exact_match bool

Whether to match the name exactly, by default False.

False
start_key Optional[str]

The starting point for the next set of results, by default None.

None

Returns:

Type Description
Iterator[Unit]

An iterator of Unit objects.

Source code in src/albert/collections/units.py
def list(
    self,
    *,
    limit: int = 100,
    name: str | list[str] | None = None,
    category: UnitCategory | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    exact_match: bool = False,
    start_key: str | None = None,
    verified: bool | None = None,
) -> Iterator[Unit]:
    """
    Lists unit entities with optional filters.

    Parameters
    ----------
    limit : int, optional
        The maximum number of units to return, by default 50.
    name : Optional[str], optional
        The name of the unit to filter by, by default None.
    category : Optional[UnitCategory], optional
        The category of the unit to filter by, by default None.
    order_by : OrderBy, optional
        The order by which to sort the results, by default OrderBy.DESCENDING.
    exact_match : bool, optional
        Whether to match the name exactly, by default False.
    start_key : Optional[str], optional
        The starting point for the next set of results, by default None.

    Returns
    -------
    Iterator[Unit]
        An iterator of Unit objects.
    """
    params = {
        "limit": limit,
        "startKey": start_key,
        "orderBy": order_by.value,
        "name": [name] if isinstance(name, str) else name,
        "exactMatch": json.dumps(exact_match),
        "verified": json.dumps(verified) if verified is not None else None,
        "category": category.value if isinstance(category, UnitCategory) else category,
    }
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [Unit(**item) for item in items],
    )

unit_exists

unit_exists(*, name: str, exact_match: bool = True) -> bool

Checks if a unit exists by its name.

Parameters:

Name Type Description Default
name str

The name of the unit to check.

required
exact_match bool

Whether to match the name exactly, by default True.

True

Returns:

Type Description
bool

True if the unit exists, False otherwise.

Source code in src/albert/collections/units.py
def unit_exists(self, *, name: str, exact_match: bool = True) -> bool:
    """
    Checks if a unit exists by its name.

    Parameters
    ----------
    name : str
        The name of the unit to check.
    exact_match : bool, optional
        Whether to match the name exactly, by default True.

    Returns
    -------
    bool
        True if the unit exists, False otherwise.
    """
    return self.get_by_name(name=name, exact_match=exact_match) is not None

update

update(*, unit: Unit) -> Unit

Updates a unit entity by its ID.

Parameters:

Name Type Description Default
unit Unit

The updated Unit object.

required

Returns:

Type Description
Unit

The updated Unit

Source code in src/albert/collections/units.py
def update(self, *, unit: Unit) -> Unit:
    """
    Updates a unit entity by its ID.

    Parameters
    ----------
    unit : Unit
        The updated Unit object.

    Returns
    -------
    Unit
        The updated Unit
    """
    unit_id = unit.id
    original_unit = self.get_by_id(id=unit_id)
    payload = self._generate_patch_payload(existing=original_unit, updated=unit)
    url = f"{self.base_path}/{unit_id}"
    self.session.patch(url, json=payload.model_dump(mode="json", by_alias=True))
    unit = self.get_by_id(id=unit_id)
    return unit

UserCollection

UserCollection(*, session: AlbertSession)

Bases: BaseCollection

UserCollection is a collection class for managing User entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Create a new User

get_by_id

Retrieves a User by its ID.

get_current_user

Retrieves the current authenticated user.

list

Lists Users based on criteria

update

Update a User entity.

Source code in src/albert/collections/users.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the UserCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, user: User) -> User

Create a new User

Parameters:

Name Type Description Default
user User

The user to create

required

Returns:

Type Description
User

The created User

Source code in src/albert/collections/users.py
def create(self, *, user: User) -> User:  # pragma: no cover
    """Create a new User

    Parameters
    ----------
    user : User
        The user to create

    Returns
    -------
    User
        The created User
    """

    response = self.session.post(
        self.base_path,
        json=user.model_dump(by_alias=True, exclude_none=True, mode="json"),
    )
    return User(**response.json())

get_by_id

get_by_id(*, id: str) -> User

Retrieves a User by its ID.

Parameters:

Name Type Description Default
id str

The ID of the user to retrieve.

required

Returns:

Type Description
User

The User object.

Source code in src/albert/collections/users.py
def get_by_id(self, *, id: str) -> User:
    """
    Retrieves a User by its ID.

    Parameters
    ----------
    id : str
        The ID of the user to retrieve.

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

get_current_user

get_current_user() -> User

Retrieves the current authenticated user.

Returns:

Type Description
User

The current User object.

Source code in src/albert/collections/users.py
def get_current_user(self) -> User:
    """
    Retrieves the current authenticated user.

    Returns
    -------
    User
        The current User object.
    """
    claims = jwt.decode(self.session._access_token, options={"verify_signature": False})
    return self.get_by_id(id=claims["id"])

list

list(
    *,
    limit: int = 50,
    offset: int | None = None,
    text: str | None = None,
    status: Status | None = None,
    search_fields: str | None = None,
) -> Iterator[User]

Lists Users based on criteria

Parameters:

Name Type Description Default
text Optional[str]

text to search against, by default None

None

Returns:

Type Description
Generator

Generator of matching Users or None

Source code in src/albert/collections/users.py
def list(
    self,
    *,
    limit: int = 50,
    offset: int | None = None,
    text: str | None = None,
    status: Status | None = None,
    search_fields: str | None = None,
) -> Iterator[User]:
    """Lists Users based on criteria

    Parameters
    ----------
    text : Optional[str], optional
        text to search against, by default None

    Returns
    -------
    Generator
        Generator of matching Users or None
    """

    def deserialize(items: list[dict]) -> Iterator[User]:
        for item in items:
            id = item["albertId"]
            try:
                yield self.get_by_id(id=id)
            except AlbertHTTPError as e:
                logger.warning(f"Error fetching user '{id}': {e}")

    params = {
        "limit": limit,
        "offset": offset,
        "status": status,
        "text": text,
        "searchFields": search_fields,
    }

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        params=params,
        deserialize=deserialize,
    )

update

update(*, user: User) -> User

Update a User entity.

Parameters:

Name Type Description Default
user User

The updated User entity.

required

Returns:

Type Description
User

The updated User entity as returned by the server.

Source code in src/albert/collections/users.py
def update(self, *, user: User) -> User:
    """Update a User entity.

    Parameters
    ----------
    user : User
        The updated User entity.

    Returns
    -------
    User
        The updated User entity as returned by the server.
    """
    # Fetch the current object state from the server or database
    current_object = self.get_by_id(id=user.id)

    # Generate the PATCH payload
    payload = self._generate_patch_payload(existing=current_object, updated=user)

    url = f"{self.base_path}/{user.id}"
    self.session.patch(url, json=payload.model_dump(mode="json", by_alias=True))

    updated_user = self.get_by_id(id=user.id)
    return updated_user

WorkflowCollection

WorkflowCollection(*, session: AlbertSession)

Bases: BaseCollection

WorkflowCollection is a collection class for managing Workflow entities in the Albert platform.

Parameters:

Name Type Description Default
session AlbertSession

The Albert session instance.

required

Methods:

Name Description
create

Create or return matching workflows for the provided list of workflows.

get_by_id

Retrieve a Workflow by its ID.

get_by_ids

Returns a list of Workflow objects by their IDs.

list

List all workflows. Unlikly to be used in production.

Source code in src/albert/collections/workflows.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the WorkflowCollection with the provided session.

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

base_path instance-attribute

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

create

create(*, workflows: list[Workflow]) -> list[Workflow]

Create or return matching workflows for the provided list of workflows. This endpoint automatically tries to find an existing workflow with the same parameter setpoints, and will either return the existing workflow or create a new one.

Parameters:

Name Type Description Default
workflows list[Workflow]

A list of Workflow objects to find or create.

required

Returns:

Type Description
list[Workflow]

A list of created or found Workflow objects.

Source code in src/albert/collections/workflows.py
def create(self, *, workflows: list[Workflow]) -> list[Workflow]:
    """Create or return matching workflows for the provided list of workflows.
    This endpoint automatically tries to find an existing workflow with the same parameter setpoints, and will either return the existing workflow or create a new one.

    Parameters
    ----------
    workflows : list[Workflow]
        A list of Workflow objects to find or create.

    Returns
    -------
    list[Workflow]
        A list of created or found Workflow objects.
    """
    if isinstance(workflows, Workflow):
        # in case the user forgets this should be a list
        workflows = [workflows]
    response = self.session.post(
        url=f"{self.base_path}/bulk",
        json=[x.model_dump(mode="json", by_alias=True, exclude_none=True) for x in workflows],
    )
    return [Workflow(**x) for x in response.json()]

get_by_id

get_by_id(*, id: str) -> Workflow

Retrieve a Workflow by its ID.

Parameters:

Name Type Description Default
id str

The ID of the Workflow to retrieve.

required

Returns:

Type Description
Workflow

The Workflow object.

Source code in src/albert/collections/workflows.py
def get_by_id(self, *, id: str) -> Workflow:
    """Retrieve a Workflow by its ID.

    Parameters
    ----------
    id : str
        The ID of the Workflow to retrieve.

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

get_by_ids

get_by_ids(*, ids: list[str]) -> list[Workflow]

Returns a list of Workflow objects by their IDs.

Parameters:

Name Type Description Default
ids list[str]

The list of Workflow IDs to retrieve.

required

Returns:

Type Description
list[Workflow]

The list of Workflow objects matching the provided IDs.

Source code in src/albert/collections/workflows.py
def get_by_ids(self, *, ids: list[str]) -> list[Workflow]:
    """Returns a list of Workflow objects by their IDs.

    Parameters
    ----------
    ids : list[str]
        The list of Workflow IDs to retrieve.

    Returns
    -------
    list[Workflow]
        The list of Workflow objects matching the provided IDs.
    """
    url = f"{self.base_path}/ids"
    batches = [ids[i : i + 100] for i in range(0, len(ids), 100)]
    return [
        Workflow(**item)
        for batch in batches
        for item in self.session.get(url, params={"id": batch}).json()["Items"]
    ]

list

list(limit: int = 50) -> Iterator[Workflow]

List all workflows. Unlikly to be used in production.

Parameters:

Name Type Description Default
limit int

The number of workflows to return, by default 50.

50

Yields:

Type Description
Iterator[Workflow]

An iterator of Workflow objects.

Source code in src/albert/collections/workflows.py
def list(self, limit: int = 50) -> Iterator[Workflow]:
    """List all workflows. Unlikly to be used in production.

    Parameters
    ----------
    limit : int, optional
        The number of workflows to return, by default 50.

    Yields
    ------
    Iterator[Workflow]
        An iterator of Workflow objects.
    """

    def deserialize(items: list[dict]) -> list[Workflow]:
        return self.get_by_ids(ids=[x["albertId"] for x in items])

    params = {"limit": limit}
    return AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        params=params,
        session=self.session,
        deserialize=deserialize,
    )

WorksheetCollection

WorksheetCollection(*, session: AlbertSession)

Bases: BaseCollection

WorksheetCollection is a collection class for managing Worksheet entities in the Albert platform.

Methods:

Name Description
add_sheet

Create a new blank sheet in the Worksheet with the specified name.

get_by_project_id

Retrieve a worksheet by its project ID. Projects and Worksheets are 1:1 in the Albert platform.

setup_new_sheet_from_template

Create a new sheet in the Worksheet related to the specified Project from a template.

setup_worksheet

Setup a new worksheet for a project.

Source code in src/albert/collections/worksheets.py
def __init__(self, *, session: AlbertSession):
    super().__init__(session=session)
    self.base_path = f"/api/{WorksheetCollection._api_version}/worksheet"

base_path instance-attribute

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

add_sheet

add_sheet(
    *, project_id: ProjectId, sheet_name: str
) -> Worksheet

Create a new blank sheet in the Worksheet with the specified name.

Parameters:

Name Type Description Default
project_id str

The project ID for the Worksheet to add the sheet to.

required
sheet_name str

The name of the new sheet.

required

Returns:

Type Description
Worksheet

The Worksheet object for the project.

Source code in src/albert/collections/worksheets.py
def add_sheet(self, *, project_id: ProjectId, sheet_name: str) -> Worksheet:
    """Create a new blank sheet in the Worksheet with the specified name.

    Parameters
    ----------
    project_id : str
        The project ID for the Worksheet to add the sheet to.
    sheet_name : str
        The name of the new sheet.

    Returns
    -------
    Worksheet
        The Worksheet object for the project.
    """
    payload = {"name": sheet_name}
    url = f"{self.base_path}/project/{project_id}/sheets"
    self.session.put(url=url, json=payload)
    return self.get_by_project_id(project_id=project_id)

get_by_project_id

get_by_project_id(*, project_id: ProjectId) -> Worksheet

Retrieve a worksheet by its project ID. Projects and Worksheets are 1:1 in the Albert platform.

Parameters:

Name Type Description Default
project_id str

The project ID to retrieve the worksheet for.

required

Returns:

Type Description
Worksheet

The Worksheet object for that project.

Source code in src/albert/collections/worksheets.py
def get_by_project_id(self, *, project_id: ProjectId) -> Worksheet:
    """Retrieve a worksheet by its project ID. Projects and Worksheets are 1:1 in the Albert platform.

    Parameters
    ----------
    project_id : str
        The project ID to retrieve the worksheet for.

    Returns
    -------
    Worksheet
        The Worksheet object for that project.
    """

    params = {"type": "project", "id": project_id}
    response = self.session.get(self.base_path, params=params)

    response_json = response.json()

    # Sheets are themselves collections, and therefore need access to the session
    response_json = self._add_session_to_sheets(response_json)

    return Worksheet(**response_json)

setup_new_sheet_from_template

setup_new_sheet_from_template(
    *,
    project_id: ProjectId,
    sheet_template_id: str,
    sheet_name: str,
) -> Worksheet

Create a new sheet in the Worksheet related to the specified Project from a template.

Parameters:

Name Type Description Default
project_id str

description

required
sheet_template_id str

description

required
sheet_name str

description

required

Returns:

Type Description
Worksheet

The Worksheet object for the project.

Source code in src/albert/collections/worksheets.py
def setup_new_sheet_from_template(
    self, *, project_id: ProjectId, sheet_template_id: str, sheet_name: str
) -> Worksheet:
    """Create a new sheet in the Worksheet related to the specified Project from a template.

    Parameters
    ----------
    project_id : str
        _description_
    sheet_template_id : str
        _description_
    sheet_name : str
        _description_

    Returns
    -------
    Worksheet
        The Worksheet object for the project.
    """
    payload = {"name": sheet_name}
    params = {"templateId": sheet_template_id}
    path = f"{self.base_path}/project/{project_id}/sheets"
    self.session.post(path, json=payload, params=params)
    return self.get_by_project_id(project_id=project_id)

setup_worksheet

setup_worksheet(
    *, project_id: ProjectId, add_sheet=False
) -> Worksheet

Setup a new worksheet for a project.

Parameters:

Name Type Description Default
project_id str

The project ID to setup the worksheet for.

required
add_sheet bool

Whether to add a blank sheet to the worksheet, by default False

False

Returns:

Type Description
Worksheet

The Worksheet object for the project.

Source code in src/albert/collections/worksheets.py
def setup_worksheet(self, *, project_id: ProjectId, add_sheet=False) -> Worksheet:
    """Setup a new worksheet for a project.

    Parameters
    ----------
    project_id : str
        The project ID to setup the worksheet for.
    add_sheet : bool, optional
        Whether to add a blank sheet to the worksheet, by default False

    Returns
    -------
    Worksheet
        The Worksheet object for the project.
    """

    params = {"sheets": str(add_sheet).lower()}
    path = f"{self.base_path}/{project_id}/setup"
    self.session.post(path, json=params)
    return self.get_by_project_id(project_id=project_id)