Skip to content

Cas

albert.collections.cas.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
get_all

Get all CAS entities with optional filters.

exists

Checks if a CAS exists by its number.

create

Creates a new CAS entity.

get_or_create

Retrieves a CAS by its number or creates it if it does not exist.

get_by_id

Retrieves a CAS by its ID.

get_by_number

Retrieves a CAS by its number.

delete

Deletes a CAS by its ID.

update

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

Attributes:

Name Type Description
base_path
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

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

get_all

get_all(
    *,
    number: str | None = None,
    cas: list[str] | None = None,
    id: CasId | None = None,
    order_by: OrderBy = DESCENDING,
    start_key: str | None = None,
    max_items: int | None = None,
) -> Iterator[Cas]

Get all CAS entities with optional filters.

Parameters:

Name Type Description Default
number str

Filter CAS entities by CAS number.

None
cas list[str] | None

Filter CAS entities by a list of CAS numbers.

None
id str

Filter CAS entities by Albert CAS ID.

None
order_by OrderBy

Sort direction (ascending or descending). Default is DESCENDING.

DESCENDING
start_key str

The pagination key to start fetching from.

None
max_items int

Maximum number of items to return in total. If None, fetches all available items.

None

Returns:

Type Description
Iterator[Cas]

An iterator over Cas entities.

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

    Parameters
    ----------
    number : str, optional
        Filter CAS entities by CAS number.
    cas : list[str] | None, optional
        Filter CAS entities by a list of CAS numbers.
    id : str, optional
        Filter CAS entities by Albert CAS ID.
    order_by : OrderBy, optional
        Sort direction (ascending or descending). Default is DESCENDING.
    start_key : str, optional
        The pagination key to start fetching from.
    max_items : int, optional
        Maximum number of items to return in total. If None, fetches all available items.

    Returns
    -------
    Iterator[Cas]
        An iterator over Cas entities.
    """

    if cas:
        params = {
            "orderBy": order_by.value,
            "cas": cas,
        }

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

        items = response.json().get("Items", [])

        yield from [Cas(**item) for item in items]
        return

    params = {"orderBy": order_by.value, "startKey": start_key}

    cas_items = AlbertPaginator(
        mode=PaginationMode.KEY,
        path=self.base_path,
        session=self.session,
        params=params,
        max_items=None,
        deserialize=lambda items: [Cas(**item) for item in items],
    )

    # Apply custom filtering until https://linear.app/albert-invent/issue/TAS-564/inconsistent-cas-pagination-behaviour is fixed.
    def filtered_items() -> Iterator[Cas]:
        count = 0
        for item in cas_items:
            if number is not None and number not in item.number:
                continue
            if id is not None and item.id != id:
                continue
            yield item
            count += 1
            if max_items is not None and count >= max_items:
                break

    yield from filtered_items()

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

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)

    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

get_or_create

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

Retrieves a CAS by its number or creates it if it does not exist.

Parameters:

Name Type Description Default
cas Union[str, Cas]

The CAS number or Cas object to retrieve or create.

required

Returns:

Type Description
Cas

The Cas object if found or created.

Source code in src/albert/collections/cas.py
def get_or_create(self, *, cas: str | Cas) -> Cas:
    """
    Retrieves a CAS by its number or creates it if it does not exist.

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

    Returns
    -------
    Cas
        The Cas object if found or created.
    """
    if isinstance(cas, str):
        cas = Cas(number=cas)
    found = self.get_by_number(number=cas.number, exact_match=True)
    if found:
        return found
    else:
        return self.create(cas=cas)

get_by_id

get_by_id(*, id: CasId) -> 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
@validate_call
def get_by_id(self, *, id: CasId) -> 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(cas=[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)

delete

delete(*, id: CasId) -> 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
@validate_call
def delete(self, *, id: CasId) -> 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)

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