Cas

albert.collections.cas

AlbertPaginator

AlbertPaginator(
    *,
    path: str,
    mode: PaginationMode,
    session: AlbertSession,
    deserialize: Callable[
        [Iterable[dict]], Iterable[ItemType]
    ],
    params: dict[str, str] | None = None,
)

Bases: Iterator[ItemType]

Helper class for pagination through Albert endpoints.

Two pagination modes are possible: - Offset-based via by the offset query parameter - Key-based via by the startKey query parameter and 'lastKey' response field

A custom deserialize function is provided when additional logic is required to load the raw items returned by the search listing, e.g., making additional Albert API calls.

Source code in src/albert/core/pagination.py

def __init__(
    self,
    *,
    path: str,
    mode: PaginationMode,
    session: AlbertSession,
    deserialize: Callable[[Iterable[dict]], Iterable[ItemType]],
    params: dict[str, str] | None = None,
):
    self.path = path
    self.mode = mode
    self.session = session
    self.deserialize = deserialize

    params = params or {}
    self.params = {k: v for k, v in params.items() if v is not None}

    if "startKey" in self.params:
        self.params["startKey"] = quote_plus(self.params["startKey"])

    self._iterator = self._create_iterator()

deserialize instance-attribute

deserialize = deserialize

mode instance-attribute

mode = mode

params instance-attribute

params = {k: _jfor (k, v) in items() if v is not None}

path instance-attribute

path = path

session instance-attribute

session = session

AlbertSession

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

Bases: Session

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

Parameters:

Name	Type	Description	Default
base_url	str	The base URL to prefix to all relative request paths (e.g., "https://app.albertinvent.com").	required
token	str | None	A static JWT token for authentication. Ignored if auth_manager is provided.	None
auth_manager	AlbertClientCredentials | AlbertSSOClient	An authentication manager used to dynamically fetch and refresh tokens. If provided, it overrides token.	None
retries	int	The number of automatic retries on failed requests (default is 3).	None

Methods:

Name	Description
request

Source code in src/albert/core/session.py

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

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

    self._auth_manager = auth_manager
    self._provided_token = token

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

base_url instance-attribute

base_url = base_url

request

request(
    method: str, path: str, *args, **kwargs
) -> Response

Source code in src/albert/core/session.py

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

BaseCollection

BaseCollection(*, session: AlbertSession)

BaseCollection is the base class for all collection classes.

Parameters:

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

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

def __init__(self, *, session: AlbertSession):
    self.session = session

session instance-attribute

session = session

Cas pydantic-model

Bases: BaseAlbertModel

Represents a CAS entity.

Show JSON schema:

{
  "$defs": {
    "CasCategory": {
      "enum": [
        "User",
        "Verisk",
        "TSCA - Public",
        "TSCA - Private",
        "not TSCA",
        "CAS linked to External Database",
        "Unknown (Trade Secret)",
        "CL_Inventory Upload"
      ],
      "title": "CasCategory",
      "type": "string"
    },
    "Hazard": {
      "description": "Represents a chemical hazard.",
      "properties": {
        "subCategory": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Hazard subcategory",
          "title": "Subcategory"
        },
        "hCode": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Hazard code",
          "title": "Hcode"
        },
        "category": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "number"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Hazard category",
          "title": "Category"
        },
        "class": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Hazard classification",
          "title": "Class"
        },
        "hCodeText": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "description": "Hazard code text",
          "title": "Hcodetext"
        }
      },
      "title": "Hazard",
      "type": "object"
    }
  },
  "description": "Represents a CAS entity.",
  "properties": {
    "number": {
      "description": "The CAS number.",
      "title": "Number",
      "type": "string"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Name of the CAS.",
      "title": "Name"
    },
    "description": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The description or name of the CAS.",
      "title": "Description"
    },
    "notes": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Notes related to the CAS.",
      "title": "Notes"
    },
    "category": {
      "anyOf": [
        {
          "$ref": "#/$defs/CasCategory"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The category of the CAS."
    },
    "casSmiles": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "CAS SMILES notation.",
      "title": "Cassmiles"
    },
    "inchiKey": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "InChIKey of the CAS.",
      "title": "Inchikey"
    },
    "iUpacName": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "IUPAC name of the CAS.",
      "title": "Iupacname"
    },
    "albertId": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The AlbertID of the CAS.",
      "title": "Albertid"
    },
    "hazards": {
      "anyOf": [
        {
          "items": {
            "$ref": "#/$defs/Hazard"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Hazards associated with the CAS.",
      "title": "Hazards"
    },
    "wgk": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "German Water Hazard Class (WGK) number.",
      "title": "Wgk"
    },
    "ecListNo": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "European Community (EC) number.",
      "title": "Eclistno"
    },
    "type": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Type of the CAS.",
      "title": "Type"
    },
    "classificationType": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Classification type of the CAS.",
      "title": "Classificationtype"
    },
    "order": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "CAS order.",
      "title": "Order"
    }
  },
  "required": [
    "number"
  ],
  "title": "Cas",
  "type": "object"
}

Fields:

• number (str)
• name (str | None)
• description (str | None)
• notes (str | None)
• category (CasCategory | None)
• smiles (str | None)
• inchi_key (str | None)
• iupac_name (str | None)
• id (str | None)
• hazards (list[Hazard] | None)
• wgk (str | None)
• ec_number (str | None)
• type (str | None)
• classification_type (str | None)
• order (str | None)

category pydantic-field

category: CasCategory | None = None

The category of the CAS.

classification_type pydantic-field

classification_type: str | None = None

Classification type of the CAS.

description pydantic-field

description: str | None = None

The description or name of the CAS.

ec_number pydantic-field

ec_number: str | None = None

European Community (EC) number.

hazards pydantic-field

hazards: list[Hazard] | None = None

Hazards associated with the CAS.

id pydantic-field

id: str | None = None

The AlbertID of the CAS.

inchi_key pydantic-field

inchi_key: str | None = None

InChIKey of the CAS.

iupac_name pydantic-field

iupac_name: str | None = None

IUPAC name of the CAS.

name pydantic-field

name: str | None = None

Name of the CAS.

notes pydantic-field

notes: str | None = None

Notes related to the CAS.

number pydantic-field

number: str

The CAS number.

order pydantic-field

order: str | None = None

CAS order.

smiles pydantic-field

smiles: str | None = None

CAS SMILES notation.

type pydantic-field

type: str | None = None

Type of the CAS.

wgk pydantic-field

wgk: str | None = None

German Water Hazard Class (WGK) number.

from_string classmethod

from_string(*, number: str) -> Cas

Creates a Cas object from a string.

Parameters:

Name	Type	Description	Default
number	str	The CAS number.	required

Returns:

Type	Description
Cas	The Cas object created from the string.

Source code in src/albert/resources/cas.py

@classmethod
def from_string(cls, *, number: str) -> "Cas":
    """
    Creates a Cas object from a string.

    Parameters
    ----------
    number : str
        The CAS number.

    Returns
    -------
    Cas
        The Cas object created from the string.
    """
    return cls(number=number)

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

OrderBy

Bases: str, Enum

ASCENDING class-attribute instance-attribute

ASCENDING = 'asc'

DESCENDING class-attribute instance-attribute

DESCENDING = 'desc'

PaginationMode

Bases: str, Enum

KEY class-attribute instance-attribute

KEY = 'key'

OFFSET class-attribute instance-attribute

OFFSET = 'offset'