Skip to content

Albert Client

albert.client

Albert

Albert(
    *,
    base_url: str | None = None,
    token: str | None = None,
    auth_manager: AlbertClientCredentials
    | AlbertSSOClient
    | None = None,
    retries: int | None = None,
    session: AlbertSession | None = None,
)

Main client for interacting with the Albert API.

This class manages authentication and access to API resource collections. It supports token-based, SSO, and client credentials authentication via a unified interface.

Parameters:

Name Type Description Default
base_url str

The base URL of the Albert API. Defaults to the "ALBERT_BASE_URL" environment variable or "https://app.albertinvent.com" if not set.

None
token str

A static token for authentication. If provided, it overrides any auth_manager. Defaults to the "ALBERT_TOKEN" environment variable.

None
auth_manager AlbertClientCredentials | AlbertSSOClient

An authentication manager for OAuth2-based authentication flows. Ignored if token is provided.

None
retries int

Maximum number of retries for failed HTTP requests.

None
session AlbertSession

A fully configured session instance. If provided, base_url, token, and auth_manager are all ignored.

None

Attributes:

Name Type Description
session AlbertSession

The internal session used for authenticated requests.

projects ProjectCollection

Access to project-related API methods.

tags TagCollection

Access to tag-related API methods.

inventory InventoryCollection

Access to inventory-related API methods.

companies CompanyCollection

Access to company-related API methods.

Helpers
  • from_token — Create a client using a static token.
  • from_sso — Create a client using interactive browser-based SSO login.
  • from_client_credentials — Create a client using OAuth2 client credentials.

Methods:

Name Description
from_client_credentials

Create an Albert client using client credentials authentication.

from_sso

Create an Albert client using interactive OAuth2 SSO login.

from_token

Create an Albert client using a static token for authentication.

Source code in src/albert/client.py
def __init__(
    self,
    *,
    base_url: str | None = None,
    token: str | None = None,
    auth_manager: AlbertClientCredentials | AlbertSSOClient | 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"),
        auth_manager=auth_manager,
        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

custom_templates property

custom_templates: CustomTemplatesCollection

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"),
    auth_manager=auth_manager,
    retries=retries,
)

storage_locations property

storage_locations: StorageLocationsCollection

substances property

substances: SubstanceCollection

tags property

tasks property

un_numbers property

un_numbers: UnNumberCollection

units property

users property

workflows property

workflows: WorkflowCollection

worksheets property

worksheets: WorksheetCollection

from_client_credentials classmethod

from_client_credentials(
    *,
    base_url: str,
    client_id: str,
    client_secret: str,
    retries: int | None = None,
) -> Albert

Create an Albert client using client credentials authentication.

Source code in src/albert/client.py
@classmethod
def from_client_credentials(
    cls,
    *,
    base_url: str,
    client_id: str,
    client_secret: str,
    retries: int | None = None,
) -> Albert:
    """Create an Albert client using client credentials authentication."""
    creds = AlbertClientCredentials(
        id=client_id,
        secret=SecretStr(client_secret),
        base_url=base_url,
    )
    return cls(base_url=base_url, auth_manager=creds, retries=retries)

from_sso classmethod

from_sso(
    *,
    base_url: str,
    email: str,
    port: int = 5000,
    tenant_id: str | None = None,
    retries: int | None = None,
) -> Albert

Create an Albert client using interactive OAuth2 SSO login.

Source code in src/albert/client.py
@classmethod
def from_sso(
    cls,
    *,
    base_url: str,
    email: str,
    port: int = 5000,
    tenant_id: str | None = None,
    retries: int | None = None,
) -> Albert:
    """Create an Albert client using interactive OAuth2 SSO login."""
    oauth = AlbertSSOClient(base_url=base_url, email=email)
    oauth.authenticate(minimum_port=port, tenant_id=tenant_id)
    return cls(base_url=base_url, auth_manager=oauth, retries=retries)

from_token classmethod

from_token(*, base_url: str, token: str) -> Albert

Create an Albert client using a static token for authentication.

Source code in src/albert/client.py
@classmethod
def from_token(cls, *, base_url: str, token: str) -> Albert:
    """Create an Albert client using a static token for authentication."""
    return cls(base_url=base_url, token=token)

AlbertClientCredentials pydantic-model

Bases: BaseAlbertModel, AuthManager

Client credentials manager for programmatic OAuth2 access to the Albert API.

This class implements the OAuth2 Client Credentials flow, allowing automated systems (non-interactive) to authenticate securely using a client ID and secret.

Attributes:

Name Type Description
id str

The client ID used for authentication.

secret SecretStr

The client secret used for authentication.

base_url str

The base URL of the Albert API.

Usage

creds = AlbertClientCredentials( ... id="your-client-id", ... secret=SecretStr("your-client-secret"), ... base_url="https://app.albertinvent.com", ... ) client = Albert(auth_manager=creds) client.roles.get_all()

Show JSON schema:
{
  "description": "Client credentials manager for programmatic OAuth2 access to the Albert API.\n\nThis class implements the OAuth2 Client Credentials flow, allowing automated\nsystems (non-interactive) to authenticate securely using a client ID and secret.\n\nAttributes\n----------\nid : str\n    The client ID used for authentication.\nsecret : SecretStr\n    The client secret used for authentication.\nbase_url : str\n    The base URL of the Albert API.\n\nUsage\n-----\n>>> creds = AlbertClientCredentials(\n...     id=\"your-client-id\",\n...     secret=SecretStr(\"your-client-secret\"),\n...     base_url=\"https://app.albertinvent.com\",\n... )\n>>> client = Albert(auth_manager=creds)\n>>> client.roles.get_all()",
  "properties": {
    "id": {
      "title": "Id",
      "type": "string"
    },
    "secret": {
      "format": "password",
      "title": "Secret",
      "type": "string",
      "writeOnly": true
    },
    "base_url": {
      "title": "Base Url",
      "type": "string"
    }
  },
  "required": [
    "id",
    "secret"
  ],
  "title": "AlbertClientCredentials",
  "type": "object"
}

Fields:

base_url pydantic-field

base_url: str

id pydantic-field

id: str

oauth_token_url property

oauth_token_url: str

Return the full URL to the OAuth token endpoint.

secret pydantic-field

secret: SecretStr

from_env classmethod

from_env(
    *,
    base_url_env: str = "ALBERT_BASE_URL",
    client_id_env: str = "ALBERT_CLIENT_ID",
    client_secret_env: str = "ALBERT_CLIENT_SECRET",
) -> AlbertClientCredentials | None

Create AlbertClientCredentials from environment variables.

Returns None if any of the required environment variables are missing.

Parameters:

Name Type Description Default
base_url_env str

Name of the environment variable containing the base URL (defaults to "ALBERT_BASE_URL").

'ALBERT_BASE_URL'
client_id_env str

Name of the environment variable containing the client ID (defaults to "ALBERT_CLIENT_ID").

'ALBERT_CLIENT_ID'
client_secret_env str

Name of the environment variable containing the client secret (defaults to "ALBERT_CLIENT_SECRET").

'ALBERT_CLIENT_SECRET'

Returns:

Type Description
AlbertClientCredentials | None

The credentials instance if all environment variables are present; otherwise, None.

Source code in src/albert/core/auth/credentials.py
@classmethod
def from_env(
    cls,
    *,
    base_url_env: str = "ALBERT_BASE_URL",
    client_id_env: str = "ALBERT_CLIENT_ID",
    client_secret_env: str = "ALBERT_CLIENT_SECRET",
) -> AlbertClientCredentials | None:
    """
    Create `AlbertClientCredentials` from environment variables.

    Returns None if any of the required environment variables are missing.

    Parameters
    ----------
    base_url_env : str
        Name of the environment variable containing the base URL
        (defaults to "ALBERT_BASE_URL").
    client_id_env : str
        Name of the environment variable containing the client ID
        (defaults to "ALBERT_CLIENT_ID").
    client_secret_env : str
        Name of the environment variable containing the client secret
        (defaults to "ALBERT_CLIENT_SECRET").

    Returns
    -------
    AlbertClientCredentials | None
        The credentials instance if all environment variables are present;
        otherwise, None.
    """
    client_id = os.getenv(client_id_env)
    client_secret = os.getenv(client_secret_env)
    base_url = os.getenv(base_url_env)

    if client_id and client_secret and base_url:
        return cls(
            id=client_id,
            secret=SecretStr(client_secret),
            base_url=base_url,
        )

get_access_token

get_access_token() -> str

Return a valid access token, refreshing it if needed.

Source code in src/albert/core/auth/credentials.py
def get_access_token(self) -> str:
    """Return a valid access token, refreshing it if needed."""
    if self._requires_refresh():
        self._request_access_token()
    return self._token_info.access_token

AlbertSSOClient pydantic-model

Bases: BaseAlbertModel, AuthManager

OAuth2 client for performing Authorization Code Flow with the Albert API.

This client opens a browser-based SSO login flow and handles token acquisition and refresh using a local redirect server.

If base_url is not provided, it defaults to the value of the environment variable "ALBERT_BASE_URL" or "https://app.albertinvent.com".

Attributes:

Name Type Description
base_url str

The base URL of the Albert API.

email str

The email address used for initiating the login flow.

Usage

oauth = AlbertOAuthClient( ... email="user@example.com", ... ) oauth.authenticate() client = Albert(auth_manager=oauth) client.roles.get_all()

Show JSON schema:
{
  "description": "OAuth2 client for performing Authorization Code Flow with the Albert API.\n\nThis client opens a browser-based SSO login flow and handles token acquisition\nand refresh using a local redirect server.\n\nIf `base_url` is not provided, it defaults to the value of the environment\nvariable \"ALBERT_BASE_URL\" or \"https://app.albertinvent.com\".\n\nAttributes\n----------\nbase_url : str\n    The base URL of the Albert API.\nemail : str\n    The email address used for initiating the login flow.\n\nUsage\n-----\n>>> oauth = AlbertOAuthClient(\n...     email=\"user@example.com\",\n... )\n>>> oauth.authenticate()\n>>> client = Albert(auth_manager=oauth)\n>>> client.roles.get_all()",
  "properties": {
    "base_url": {
      "title": "Base Url",
      "type": "string"
    },
    "email": {
      "title": "Email",
      "type": "string"
    }
  },
  "required": [
    "email"
  ],
  "title": "AlbertSSOClient",
  "type": "object"
}

Fields:

base_url pydantic-field

base_url: str

email pydantic-field

email: str

refresh_token_url property

refresh_token_url: str

authenticate

authenticate(
    minimum_port: int = 5000,
    maximum_port: int | None = None,
    tenant_id: str | None = None,
) -> OAuthTokenInfo

Launch an interactive browser-based SSO login and return an OAuth token.

This method starts a temporary local HTTP server, opens the SSO login URL in the default browser, and waits for the authentication redirect to capture the refresh token.

Parameters:

Name Type Description Default
minimum_port int

The starting port to attempt for the local HTTP redirect server (default is 5000).

5000
maximum_port int | None

The maximum port to try if the minimum_port is unavailable. If None, only the minimum port will be tried.

None
tenant_id str | None

Optional tenant ID to scope the SSO login request.

None

Returns:

Type Description
OAuthTokenInfo

The initial token info containing the refresh token.

Source code in src/albert/core/auth/sso.py
def authenticate(
    self,
    minimum_port: int = 5000,
    maximum_port: int | None = None,
    tenant_id: str | None = None,
) -> OAuthTokenInfo:
    """
    Launch an interactive browser-based SSO login and return an OAuth token.

    This method starts a temporary local HTTP server, opens the SSO login URL
    in the default browser, and waits for the authentication redirect to capture
    the refresh token.

    Parameters
    ----------
    minimum_port : int, optional
        The starting port to attempt for the local HTTP redirect server (default is 5000).
    maximum_port : int | None, optional
        The maximum port to try if the `minimum_port` is unavailable. If None, only the
        minimum port will be tried.
    tenant_id : str | None, optional
        Optional tenant ID to scope the SSO login request.

    Returns
    -------
    OAuthTokenInfo
        The initial token info containing the refresh token.
    """
    with local_http_server(minimum_port=minimum_port, maximum_port=maximum_port) as (
        server,
        port,
    ):
        login_url = self._build_login_url(port=port, tenant_id=tenant_id)
        webbrowser.open(login_url)

        # Block here until one request arrives at localhost:port/?token=…
        server.handle_request()
        refresh_token = server.token

    self._token_info = OAuthTokenInfo(refresh_token=refresh_token)
    return self._token_info

get_access_token

get_access_token() -> str

Return a valid access token, refreshing it if needed.

Source code in src/albert/core/auth/sso.py
def get_access_token(self) -> str:
    """Return a valid access token, refreshing it if needed."""
    if self._requires_refresh():
        self._request_access_token()
    return self._token_info.access_token

AlbertSession

AlbertSession(
    *,
    base_url: str,
    token: str | None = None,
    auth_manager: AlbertClientCredentials
    | AlbertSSOClient
    | None = None,
    retries: int | None = None,
)

Bases: Session

A session that has a base URL, which is prefixed to all request URLs.

Parameters:

Name Type Description Default
base_url str

The base URL to prefix to all relative request paths (e.g., "https://app.albertinvent.com").

required
token str | None

A static JWT token for authentication. Ignored if auth_manager is provided.

None
auth_manager AlbertClientCredentials | AlbertSSOClient

An authentication manager used to dynamically fetch and refresh tokens. If provided, it overrides token.

None
retries int

The number of automatic retries on failed requests (default is 3).

None

Methods:

Name Description
request
Source code in src/albert/core/session.py
def __init__(
    self,
    *,
    base_url: str,
    token: str | None = None,
    auth_manager: AlbertClientCredentials | AlbertSSOClient | None = None,
    retries: int | None = None,
):
    super().__init__()
    self.base_url = base_url
    self.headers.update(
        {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": f"albert-SDK V.{albert.__version__}",
        }
    )

    if token is None and auth_manager is None:
        raise ValueError("Either `token` or `auth_manager` must be specified.")

    self._auth_manager = auth_manager
    self._provided_token = token

    # Set up retry logic
    retries = retries if retries is not None else 3
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=0.3,
        status_forcelist=(500, 502, 503, 504, 403),
        raise_on_status=False,
    )
    adapter = HTTPAdapter(max_retries=retry)
    self.mount("http://", adapter)
    self.mount("https://", adapter)

base_url instance-attribute

base_url = base_url

request

request(
    method: str, path: str, *args, **kwargs
) -> Response
Source code in src/albert/core/session.py
def request(self, method: str, path: str, *args, **kwargs) -> requests.Response:
    self.headers["Authorization"] = f"Bearer {self._access_token}"
    full_url = urljoin(self.base_url, path) if not path.startswith("http") else path
    with handle_http_errors():
        response = super().request(method, full_url, *args, **kwargs)
        response.raise_for_status()
        return response

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_by_id

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_by_id

get_by_id(
    *,
    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
@validate_call
def get_by_id(
    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
create

Creates a new CAS entity.

delete

Deletes a CAS by its ID.

exists

Checks if a CAS exists by its number.

get_all

Get all CAS entities with optional filters.

get_by_id

Retrieves a CAS by its ID.

get_by_number

Retrieves a CAS by its number.

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'

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)

exists

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

get_all

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

Get all 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 get_all(
    self,
    *,
    limit: int = 50,
    start_key: str | None = None,
    number: str | None = None,
    id: str | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
) -> Iterator[Cas]:
    """
    Get all 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],
    )

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.get_all(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)

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
create

Creates a new company entity.

delete

Deletes a company entity.

exists

Checks if a company exists by its name.

get_all

Get all company entities with optional filters.

get_by_id

Get a company by its ID.

get_by_name

Retrieves a company by its name.

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'

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)

exists

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

get_all

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

Get all 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 get_all(
    self,
    *,
    limit: int = 50,
    name: str | list[str] = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Company]:
    """
    Get all 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],
    )

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.get_all(name=name, exact_match=exact_match)
    return next(found, None)

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.projects 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_all

Get all CustomField entities with optional filters.

get_by_id

Get a CustomField item by its ID.

get_by_name

Get a CustomField item by its name.

update

Update a CustomField item.

Source code in src/albert/collections/custom_fields.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the CustomFieldCollection 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_all

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

Get all CustomField entities with optional filters.

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 entities matching the search criteria.

Source code in src/albert/collections/custom_fields.py
def get_all(
    self,
    *,
    name: str | None = None,
    service: ServiceType | None = None,
    lookup_column: bool | None = None,
    lookup_row: bool | None = None,
) -> Iterator[CustomField]:
    """Get all CustomField entities with optional filters.

    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 entities matching the search criteria.
    """
    params = {
        "name": name,
        "service": service,
        "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],
    )

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.get_all(name=name, service=service):
        if custom_field.name.lower() == name.lower():
            return custom_field
    return None

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_all

Retrieve fully hydrated CustomTemplate entities with optional filters.

get_by_id

Get a Custom Template by ID

search

Search for CustomTemplate 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_all

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

Retrieve fully hydrated CustomTemplate entities with optional filters.

This method returns complete entity data using get_by_id. Use :meth:search for faster retrieval when you only need lightweight, partial (unhydrated) entities.

Source code in src/albert/collections/custom_templates.py
def get_all(
    self,
    *,
    text: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> Iterator[CustomTemplate]:
    """Retrieve fully hydrated CustomTemplate entities with optional filters.

    This method returns complete entity data using `get_by_id`.
    Use :meth:`search` for faster retrieval when you only need lightweight, partial (unhydrated) entities.
    """

    for item in self.search(text=text, limit=limit, offset=offset):
        try:
            yield self.get_by_id(id=item.id)
        except AlbertHTTPError as e:
            logger.warning(f"Error hydrating custom template {id}: {e}")

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

search

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

Search for CustomTemplate matching the provided criteria.

⚠️ This method returns partial (unhydrated) entities to optimize performance. To retrieve fully detailed entities, use :meth:get_all instead.

Parameters:

Name Type Description Default
text str | None

The text to search for, by default None

None

Yields:

Type Description
Iterator[CustomTemplateSearchItem]

An iterator of CustomTemplateSearchItem items matching the search criteria.

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

    ⚠️ This method returns partial (unhydrated) entities to optimize performance.
    To retrieve fully detailed entities, use :meth:`get_all` instead.

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


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

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

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

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_all

Get all data column entities with optional filters.

get_by_id

Get a data column by its ID.

get_by_name

Get a data column by its name.

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_all

get_all(
    *,
    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,
) -> Iterator[DataColumn]

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

Returns:

Type Description
Iterator[DataColumn]

An iterator of DataColumns matching the provided criteria.

Source code in src/albert/collections/data_columns.py
def get_all(
    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,
) -> Iterator[DataColumn]:
    """
    Get all 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.

    Returns
    -------
    Iterator[DataColumn]
        An iterator of DataColumns matching the provided criteria.
    """

    def deserialize(items: list[dict]) -> Iterator[DataColumn]:
        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,
    )

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.get_all(name=name):
        if dc.name.lower() == name.lower():
            return dc
    return None

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_all

Retrieve fully hydrated DataTemplate entities with optional filters.

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.

search

Search for DataTemplate matching the provided criteria.

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
    if (
        isinstance(data_template.data_column_values, list)
        and len(data_template.data_column_values) == 0
    ):
        data_template.data_column_values = None
    if data_template.data_column_values is not None:
        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_all

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

Retrieve fully hydrated DataTemplate entities with optional filters.

This method returns complete entity data using get_by_ids. Use :meth:search for faster retrieval when you only need lightweight, partial (unhydrated) entities.

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

    This method returns complete entity data using `get_by_ids`.
    Use :meth:`search` for faster retrieval when you only need lightweight, partial (unhydrated) entities.
    """

    def batched(iterable, size):
        """Yield lists of up to `size` IDs from an iterable of objects with an `id` attribute."""

        it = (item.id for item in iterable)
        while batch := list(islice(it, size)):
            yield batch

    id_batches = batched(
        self.search(name=name, user_id=user_id, order_by=order_by, limit=limit, offset=offset),
        limit,  # batch size
    )

    for batch in id_batches:
        try:
            hydrated_templates = self.get_by_ids(ids=batch)
            yield from hydrated_templates
        except AlbertHTTPError as e:
            logger.warning(f"Error hydrating batch {batch}: {e}")

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.get_all(name=name))
    for h in hits:
        if h.name.lower() == name.lower():
            return h
    return None

search

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

Search for DataTemplate matching the provided criteria.

⚠️ This method returns partial (unhydrated) entities to optimize performance. To retrieve fully detailed entities, use :meth:get_all instead.

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[DataTemplateSearchItem]

An iterator of DataTemplateSearchItem objects matching the provided criteria.

Source code in src/albert/collections/data_templates.py
def search(
    self,
    *,
    name: str | None = None,
    user_id: str | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    limit: int = 100,
    offset: int = 0,
) -> Iterator[DataTemplateSearchItem]:
    """Search for DataTemplate matching the provided criteria.

    ⚠️ This method returns partial (unhydrated) entities to optimize performance.
    To retrieve fully detailed entities, use :meth:`get_all` instead.

    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[DataTemplateSearchItem]
        An iterator of DataTemplateSearchItem objects matching the provided criteria.
    """

    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=lambda items: [
            DataTemplateSearchItem.model_validate(x)._bind_collection(self) for x in items
        ],
        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 FileCollection 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

TODO: 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
        TODO: _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

S3 signed URL.

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
        S3 signed URL.
    """
    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

S3 signed URL.

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
        S3 signed URL.
    """
    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.

exists

Check if an inventory item exists.

get_all

Retrieve fully hydrated InventoryItem entities with optional filters.

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

merge

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

search

Search for Inventory matching the provided criteria.

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)

exists

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

get_all

get_all(
    *, params: InventoryFilterParams | None = None
) -> Iterator[InventoryItem]

Retrieve fully hydrated InventoryItem entities with optional filters.

This method returns complete entity data using get_by_ids. Use :meth:search for faster retrieval when you only need lightweight, partial (unhydrated) entities.

Source code in src/albert/collections/inventory.py
@validate_call
def get_all(self, *, params: InventoryFilterParams | None = None) -> Iterator[InventoryItem]:
    """
    Retrieve fully hydrated InventoryItem entities with optional filters.

    This method returns complete entity data using `get_by_ids`.
    Use :meth:`search` for faster retrieval when you only need lightweight, partial (unhydrated) entities.
    """
    params = params or InventoryFilterParams()

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

    search_text = (
        params.text if (params.text is None or len(params.text) < 50) else params.text[0:50]
    )
    query_params = self._prepare_parameters(
        limit=params.limit,
        text=search_text,
        cas=params.cas,
        category=params.category,
        company=params.company,
        order=params.order,
        sort_by=params.sort_by,
        location=params.location,
        storage_location=params.storage_location,
        project_id=params.project_id,
        sheet_id=params.sheet_id,
        created_by=params.created_by,
        lot_owner=params.lot_owner,
        tags=params.tags,
    )

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

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 a 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 a 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.get_all(
        params=InventoryFilterParams(
            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()
    ]

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(
    *, params: InventoryFilterParams | None = None
) -> Iterator[InventorySearchItem]

Search for Inventory matching the provided criteria.

⚠️ This method returns partial (unhydrated) entities to optimize performance. To retrieve fully detailed entities, use :meth:get_all instead.

Source code in src/albert/collections/inventory.py
@validate_call
def search(
    self, *, params: InventoryFilterParams | None = None
) -> Iterator[InventorySearchItem]:
    """Search for Inventory matching the provided criteria.

    ⚠️ This method returns partial (unhydrated) entities to optimize performance.
    To retrieve fully detailed entities, use :meth:`get_all` instead.
    """
    params = params or InventoryFilterParams()
    query_params = self._prepare_parameters(
        limit=params.limit,
        text=params.text,
        cas=params.cas,
        category=params.category,
        company=params.company,
        location=params.location,
        storage_location=params.storage_location,
        project_id=params.project_id,
        sheet_id=params.sheet_id,
        created_by=params.created_by,
        lot_owner=params.lot_owner,
        tags=params.tags,
    )

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

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/llmsearch"
        if params.match_all_conditions
        else f"{self.base_path}/search",
        params=query_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_all

Get all link entities with optional filters.

get_by_id

Retrieves a link entity by its ID.

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_all

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

Get all 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 get_all(
    self,
    *,
    limit: int = 100,
    type: str | None = None,
    category: LinkCategory | None = None,
    id: str | None = None,
) -> Iterator[Link]:
    """
    Get all 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],
    )

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

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_all

Get all list entities with optional filters.

get_by_id

Retrieves a list entity by its ID.

get_matching_item

Get a list item by name and list type.

update

Update a list item.

Source code in src/albert/collections/lists.py
def __init__(self, *, session: AlbertSession):
    """
    Initializes the ListsCollection 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_all

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

Get all 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 get_all(
    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]:
    """
    Get all 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.OFFSET,
        path=self.base_path,
        session=self.session,
        params=params,
        deserialize=lambda items: [ListItem(**item) for item in items],
    )

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.get_all(names=[name], list_type=list_type):
        if list_item.name.lower() == name.lower():
            return list_item
    return None

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.

exists

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

get_all

Get all Location entities matching the provided criteria.

get_by_id

Retrieves a location by its ID.

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 entity to create.

required

Returns:

Type Description
Location

The created Location entity.

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

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

    Returns
    -------
    Location
        The created Location entity.
    """
    exists = self.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 entity 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 entity to delete.

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

exists

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 entity to check

required

Returns:

Type Description
Location | None

The existing registered Location entity if found, otherwise None.

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

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

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

get_all

get_all(
    *,
    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]

Get all Location entities 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 entities matching the search criteria.

Source code in src/albert/collections/locations.py
def get_all(
    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]:
    """Get all Location entities 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 entities 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],
    )

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

update

update(*, location: Location) -> Location

Update a Location entity.

Parameters:

Name Type Description Default
location Location

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

required

Returns:

Type Description
Location

The updated Location entity 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 entity to update. The ID of the Location entity must be provided.

    Returns
    -------
    Location
        The updated Location entity 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_all

Get all Lot entities with optional filters.

get_by_id

Get a lot by its ID.

get_by_ids

Get a list of lots by their IDs.

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_all

get_all(
    *,
    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]

Get all 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 entities.

Source code in src/albert/collections/lots.py
def get_all(
    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]:
    """
    Get all 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 entities.
    """
    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],
    )

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

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

Get all notes by their parent ID.

get_by_id

Retrieves a note by its 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_all

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

Get all 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 get_all(self, *, parent_id: str, order_by: OrderBy = OrderBy.DESCENDING) -> list[Note]:
    """
    Get all 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],
    )

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

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_all

Retrieve all Parameter items with optional filters.

get_by_id

Retrieve a parameter by its ID.

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.get_all(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_all

get_all(
    *,
    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,
) -> Iterator[Parameter]

Retrieve all Parameter items with optional filters.

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

Yields:

Type Description
Iterator[Parameter]

An iterator of Parameters matching the given criteria.

Source code in src/albert/collections/parameters.py
def get_all(
    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,
) -> Iterator[Parameter]:
    """
    Retrieve all Parameter items with optional filters.

    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

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

    def deserialize(items: list[dict]) -> Iterator[Parameter]:
        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,
    )

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

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_all

Search and hydrate all Parameter Groups matching the given criteria.

get_by_id

Get a parameter group by its ID.

get_by_ids
get_by_name

Get a parameter group by its name.

search

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_all

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

Search and hydrate all 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

Filter 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
limit int

Page size for each search request, by default 25.

25
offset int | None

Offset to start from, by default None.

None

Yields:

Type Description
Iterator[ParameterGroup]

An iterator of fully hydrated Parameter Groups.

Source code in src/albert/collections/parameter_groups.py
def get_all(
    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 and hydrate all Parameter Groups matching the given criteria.

    Parameters
    ----------
    text : str | None, optional
        Text to search for, by default None.
    types : PGType | list[PGType] | None, optional
        Filter the returned Parameter Groups by Type, by default None.
    order_by : OrderBy, optional
        The order in which to return results, by default OrderBy.DESCENDING.
    limit : int, optional
        Page size for each search request, by default 25.
    offset : int | None, optional
        Offset to start from, by default None.

    Yields
    ------
    Iterator[ParameterGroup]
        An iterator of fully hydrated Parameter Groups.
    """
    for item in self.search(
        text=text,
        types=types,
        order_by=order_by,
        limit=limit,
        offset=offset,
    ):
        try:
            # Currently, the API is not returning Metadata, Tags, Documents, and ACL for the get_by_ids endpoint, so we need to fetch individually until that is fixed
            yield self.get_by_id(id=item.id)
        except AlbertHTTPError as e:  # pragma: no cover
            logger.warning(f"Error fetching parameter group {item.id}: {e}")

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.get_all(text=name)
    # TODO: optimize with explicit hydrate() after self.search()
    for m in matches:
        if m.name.lower() == name.lower():
            return m
    return None

search

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

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 search(
    self,
    *,
    text: str | None = None,
    types: PGType | list[PGType] | None = None,
    order_by: OrderBy = OrderBy.DESCENDING,
    limit: int = 25,
    offset: int | None = None,
) -> Iterator[ParameterGroupSearchItem]:
    """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.
    """

    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=lambda items: [
            ParameterGroupSearchItem(**item)._bind_collection(self) for item in items
        ],
    )

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_all

Retrieve fully hydrated Project entities with optional filters.

get_by_id

Retrieve a project by its ID.

search

Search for Project matching the provided criteria.

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_all

get_all(
    *, params: ProjectFilterParams | None = None
) -> Iterator[Project]

Retrieve fully hydrated Project entities with optional filters.

This method returns complete entity data using get_by_id. Use :meth:search for faster retrieval when you only need lightweight, partial (unhydrated) entities.

Parameters:

Name Type Description Default
params ProjectFilterParams

Structured query parameters to filter, sort, and paginate projects.

None

Returns:

Type Description
Iterator[Project]

An iterator of fully hydrated Project entities.

Source code in src/albert/collections/projects.py
def get_all(self, *, params: ProjectFilterParams | None = None) -> Iterator[Project]:
    """Retrieve fully hydrated Project entities with optional filters.

    This method returns complete entity data using `get_by_id`.
    Use :meth:`search` for faster retrieval when you only need lightweight, partial (unhydrated) entities.

    Parameters
    ----------
    params : ProjectFilterParams, optional
        Structured query parameters to filter, sort, and paginate projects.

    Returns
    -------
    Iterator[Project]
        An iterator of fully hydrated Project entities.
    """
    params = params or ProjectFilterParams()

    for project in self.search(params=params):
        project_id = getattr(project, "albertId", None) or getattr(project, "id", None)
        if not project_id:
            continue

        id = project_id if project_id.startswith("PRO") else f"PRO{project_id}"

        try:
            yield self.get_by_id(id=id)
        except AlbertHTTPError as e:
            logger.warning(f"Error fetching project details {id}: {e}")

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

search

search(
    *, params: ProjectFilterParams | None = None
) -> Iterator[ProjectSearchItem]

Search for Project matching the provided criteria.

⚠️ This method returns partial (unhydrated) entities to optimize performance. To retrieve fully detailed entities, use :meth:get_all instead.

Parameters:

Name Type Description Default
params ProjectFilterParams

Structured query parameters to filter, sort, and paginate projects.

None

Returns:

Type Description
Iterator[Project]

An iterator of Project resources.

Source code in src/albert/collections/projects.py
def search(self, *, params: ProjectFilterParams | None = None) -> Iterator[ProjectSearchItem]:
    """Search for Project matching the provided criteria.

    ⚠️ This method returns partial (unhydrated) entities to optimize performance.
    To retrieve fully detailed entities, use :meth:`get_all` instead.

    Parameters
    ----------
    params : ProjectFilterParams, optional
        Structured query parameters to filter, sort, and paginate projects.

    Returns
    -------
    Iterator[Project]
        An iterator of Project resources.
    """
    params = params or ProjectFilterParams()

    query_params = {
        "limit": params.limit,
        "order": params.order_by.value,
        "text": params.text,
        "sortBy": params.sort_by,
        "status": params.status,
        "marketSegment": params.market_segment,
        "application": params.application,
        "technology": params.technology,
        "createdBy": params.created_by,
        "location": params.location,
        "fromCreatedAt": params.from_created_at,
        "toCreatedAt": params.to_created_at,
        "facetField": params.facet_field,
        "facetText": params.facet_text,
        "containsField": params.contains_field,
        "containsText": params.contains_text,
        "linkedTo": params.linked_to,
        "myProjects": params.my_projects,
        "myRole": params.my_role,
    }

    return AlbertPaginator(
        mode=PaginationMode.OFFSET,
        path=f"{self.base_path}/search",
        session=self.session,
        params=query_params,
        deserialize=lambda items: [
            ProjectSearchItem(**item)._bind_collection(self) 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_all

Get all the available Roles

get_by_id

Retrieve a Role by its ID.

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_all

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

Get all 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 get_all(self, *, params: dict | None = None) -> list[Role]:
    """Get all 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]

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

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_all

Get all storage locations with optional filtering.

get_by_id

Get a storage location by its ID.

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.get_all(
        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_all

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

Get all 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
Iterator[StorageLocation]

An iterator of StorageLocation items matching the search criteria.

Source code in src/albert/collections/storage_locations.py
def get_all(
    self,
    *,
    name: str | list[str] | None = None,
    exact_match: bool = False,
    location: str | Location | None = None,
    start_key: str | None = None,
    limit: int = 50,
) -> Iterator[StorageLocation]:
    """Get all 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
    ------
    Iterator[StorageLocation]
        An iterator of StorageLocation items matching the search criteria.
    """

    # Remove explicit hydration when SUP-410 is fixed
    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,
    )

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

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
get_all

Lists tag entities with optional filters.

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_name

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 if 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 if it already exists.
    """
    if isinstance(tag, str):
        tag = Tag(tag=tag)
    hit = self.get_by_name(name=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)

exists

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 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_name(name=tag, exact_match=exact_match) is not None

get_all

get_all(
    *,
    limit: int = 50,
    order_by: OrderBy = DESCENDING,
    name: str | list[str] | None = None,
    exact_match: bool = True,
    start_key: str | None = None,
) -> Iterator[Tag]

Get all 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 get_all(
    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]:
    """
    Get all 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],
    )

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_name

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

Retrieves a tag by its name or None if not found.

Parameters:

Name Type Description Default
name 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_name(self, *, name: str, exact_match: bool = True) -> Tag | None:
    """
    Retrieves a tag by its name or None if not found.

    Parameters
    ----------
    name : 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.get_all(name=name, exact_match=exact_match)
    return next(found, None)

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_name(name=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)

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_all

Retrieve fully hydrated Task entities with optional filters.

get_by_id

Retrieve a task by its ID.

get_history
remove_block

Remove a block from a Property task.

search

Search for Task matching the provided criteria.

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)

create

create(
    *, task: PropertyTask | GeneralTask | BatchTask
) -> 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 PropertyTask | GeneralTask | BatchTask

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: PropertyTask | GeneralTask | BatchTask) -> BaseTask:
    """Create a new task. Tasks can be of different types, such as PropertyTask, and are created using the provided task object.

    Parameters
    ----------
    task : PropertyTask | GeneralTask | BatchTask
        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_all

get_all(
    *, params: TaskFilterParams | None = None
) -> Iterator[BaseTask]

Retrieve fully hydrated Task entities with optional filters.

This method returns complete entity data using get_by_id or get_by_ids. Use :meth:search for faster retrieval when you only need lightweight, partial (unhydrated) entities.

Parameters:

Name Type Description Default
params TaskFilterParams

Filter and pagination options passed to the search query.

None

Yields:

Type Description
Iterator[BaseTask]

A stream of fully hydrated Task objects (PropertyTask, BatchTask, or GeneralTask).

Source code in src/albert/collections/tasks.py
def get_all(self, *, params: TaskFilterParams | None = None) -> Iterator[BaseTask]:
    """Retrieve fully hydrated Task entities with optional filters.

    This method returns complete entity data using `get_by_id` or `get_by_ids`.
    Use :meth:`search` for faster retrieval when you only need lightweight, partial (unhydrated) entities.

    Parameters
    ----------
    params : TaskFilterParams, optional
        Filter and pagination options passed to the search query.

    Yields
    ------
    Iterator[BaseTask]
        A stream of fully hydrated Task objects (PropertyTask, BatchTask, or GeneralTask).
    """
    params = params or TaskFilterParams()

    for task in self.search(params=params):
        task_id = getattr(task, "id", None)
        if not task_id:
            continue

        try:
            yield self.get_by_id(id=task_id)
        except (AlbertHTTPError, RetryError) as e:
            logger.warning(f"Error fetching task '{id}': {e}")

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

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)

search

search(
    *, params: TaskFilterParams | None = None
) -> Iterator[TaskSearchItem]

Search for Task matching the provided criteria.

⚠️ This method returns partial (unhydrated) entities to optimize performance. To retrieve fully detailed entities, use :meth:get_all instead.

Parameters:

Name Type Description Default
params TaskFilterParams

Structured query parameters including filters, sort order, and pagination.

None

Yields:

Type Description
Iterator[BaseTask]

An iterator of matching, fully hydrated Task objects.

Source code in src/albert/collections/tasks.py
def search(self, *, params: TaskFilterParams | None = None) -> Iterator[TaskSearchItem]:
    """Search for Task matching the provided criteria.

    ⚠️ This method returns partial (unhydrated) entities to optimize performance.
    To retrieve fully detailed entities, use :meth:`get_all` instead.

    Parameters
    ----------
    params : TaskFilterParams, optional
        Structured query parameters including filters, sort order, and pagination.

    Yields
    ------
    Iterator[BaseTask]
        An iterator of matching, fully hydrated Task objects.
    """
    params = params or TaskFilterParams()

    query_params = {
        "limit": params.limit,
        "offset": params.offset,
        "order": params.order.value,
        "text": params.text,
        "sortBy": params.sort_by,
        "tags": params.tags,
        "taskId": params.task_id,
        "linkedTask": params.linked_task,
        "category": params.category,
        "albertId": params.albert_id,
        "dataTemplate": params.data_template,
        "assignedTo": params.assigned_to,
        "location": params.location,
        "priority": params.priority,
        "status": params.status,
        "parameterGroup": params.parameter_group,
        "createdBy": params.created_by,
        "projectId": params.project_id,
    }

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

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)

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_all

Get all UN Numbers matching the provided criteria.

get_by_id

Retrieve a UN Number by its ID.

get_by_name

Retrieve a UN Number by its name.

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_all

get_all(
    *,
    name: str | None = None,
    exact_match: bool = False,
    limit: int = 50,
    start_key: str | None = None,
) -> Iterator[UnNumber]

Get all 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 get_all(
    self,
    *,
    name: str | None = None,
    exact_match: bool = False,
    limit: int = 50,
    start_key: str | None = None,
) -> Iterator[UnNumber]:
    """Get all 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],
    )

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.get_all(exact_match=True, name=name)
    return next(found, None)

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.

exists

Checks if a unit exists by its name.

get_all

Get all unit entities with optional filters.

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.

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)

exists

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

get_all

get_all(
    *,
    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]

Get all 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 get_all(
    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]:
    """
    Get all 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],
    )

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.get_all(name=name, exact_match=exact_match)
    return next(found, 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_all

Retrieve fully hydrated User entities with optional filters.

get_by_id

Retrieves a User by its ID.

get_current_user

Retrieves the current authenticated user.

search

Searches for Users matching the provided 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_all

get_all(
    *,
    limit: int = 100,
    status: Status | None = None,
    type: UserFilterType | None = None,
    id: list[str] | None = None,
    start_key: str | None = None,
) -> Iterator[User]

Retrieve fully hydrated User entities with optional filters.

This method uses get_by_id to hydrate the results for convenience. Use :meth:search for better performance.

Parameters:

Name Type Description Default
limit int

Max results per page.

100
status Status

Filter by user status.

None
type UserFilterType

Attribute name to filter by (e.g., 'role').

None
id list[str]

Values of the attribute to filter on.

None
start_key Optional[str]

The starting point for the next set of results, by default None.

None

Returns:

Type Description
Iterator[User]

Fully hydrated User entities.

Source code in src/albert/collections/users.py
def get_all(
    self,
    *,
    limit: int = 100,
    status: Status | None = None,
    type: UserFilterType | None = None,
    id: list[str] | None = None,
    start_key: str | None = None,
) -> Iterator[User]:
    """
    Retrieve fully hydrated User entities with optional filters.

    This method uses `get_by_id` to hydrate the results for convenience.
    Use :meth:`search` for better performance.

    Parameters
    ----------
    limit : int
        Max results per page.
    status : Status, optional
        Filter by user status.
    type : UserFilterType
        Attribute name to filter by (e.g., 'role').
    id : list[str], optional
        Values of the attribute to filter on.
    start_key : Optional[str], optional
        The starting point for the next set of results, by default None.

    Returns
    -------
    Iterator[User]
        Fully hydrated User entities.
    """
    params = {
        "limit": limit,
        "status": status,
        "type": type.value if type else None,
        "id": id,
        "startKey": start_key,
    }

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

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

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

search

search(
    *, params: UserFilterParams | None = None
) -> Iterator[UserSearchItem]

Searches for Users matching the provided criteria.

⚠️ This method returns partial (unhydrated) search results for performance. To retrieve fully detailed objects, use :meth:get_all instead.

Parameters:

Name Type Description Default
params UserFilterParams

Structured search filters for user listing.

None

Returns:

Type Description
Iterator[User]

An iterator of partial User entities.

Source code in src/albert/collections/users.py
def search(self, *, params: UserFilterParams | None = None) -> Iterator[UserSearchItem]:
    """
    Searches for Users matching the provided criteria.

    ⚠️ This method returns partial (unhydrated) search results for performance.
    To retrieve fully detailed objects, use :meth:`get_all` instead.

    Parameters
    ----------
    params : UserFilterParams, optional
        Structured search filters for user listing.

    Returns
    -------
    Iterator[User]
        An iterator of partial User entities.
    """

    params = params or UserFilterParams()
    query_params = params.model_dump(exclude_none=True, by_alias=True)

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

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_all

Get all workflows. Unlikely to be used in production.

get_by_id

Retrieve a Workflow by its ID.

get_by_ids

Returns a list of Workflow objects by their IDs.

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_all

get_all(limit: int = 50) -> Iterator[Workflow]

Get all workflows. Unlikely 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 get_all(self, limit: int = 50) -> Iterator[Workflow]:
    """Get all workflows. Unlikely 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,
    )

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

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)