Companies

albert.collections.companies

logger module-attribute

logger = create_logger()

AlbertException

AlbertException(message: str)

Bases: Exception

Source code in src/albert/exceptions.py

def __init__(self, message: str):
    super().__init__(message)
    self.message = message

message instance-attribute

message = message

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: _zfor (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

Company

Bases: BaseResource

Company is a Pydantic model representing a company entity.

Attributes:

Name	Type	Description
name	str	The name of the company.
id	str | None	The Albert ID of the company. Set when the company is retrieved from Albert.
distance	float | None	The scores of a company in a search result, optional. Read-only.

distance class-attribute instance-attribute

distance: float | None = Field(
    default=None, exclude=True, frozen=True
)

id class-attribute instance-attribute

id: str | None = Field(default=None, alias='albertId')

name instance-attribute

name: str

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

PaginationMode

Bases: str, Enum

KEY class-attribute instance-attribute

KEY = 'key'

OFFSET class-attribute instance-attribute

OFFSET = 'offset'