Companies

albert.collections.companies.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
get_all	Get all company entities with optional filters.
exists	Checks if a company exists by its name.
get_by_id	Get a company by its ID.
get_by_name	Retrieves a company by its name.
create	Creates a new company entity.
get_or_create	Retrieves a company by its name or creates it if it does not exist.
delete	Deletes a company entity.
rename	Renames an existing company entity.
update	Update a Company entity. The id of the company must be provided.

Attributes:

Name	Type	Description
base_path

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

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

get_all

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

Get all company entities with optional filters.

Parameters:

Name	Type	Description	Default
name	str | list[str]	The name(s) of the company to filter by.	None
exact_match	bool	Whether to match the name(s) exactly. Default is True.	True
start_key	str	Key to start paginated results from.	None
max_items	int	Maximum number of items to return in total. If None, fetches all available items.	None

Returns:

Type	Description
Iterator[Company]	An iterator of Company entities.

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

def get_all(
    self,
    *,
    name: str | list[str] = None,
    exact_match: bool = True,
    start_key: str | None = None,
    max_items: int | None = None,
) -> Iterator[Company]:
    """
    Get all company entities with optional filters.

    Parameters
    ----------
    name : str | list[str], optional
        The name(s) of the company to filter by.
    exact_match : bool, optional
        Whether to match the name(s) exactly. Default is True.
    start_key : str, optional
        Key to start paginated results from.
    max_items : int, optional
        Maximum number of items to return in total. If None, fetches all available items.

    Returns
    -------
    Iterator[Company]
        An iterator of Company entities.
    """
    params = {
        "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,
        max_items=max_items,
        deserialize=lambda items: [Company(**item) for item in items],
    )

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_by_id

get_by_id(*, id: CompanyId) -> 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

@validate_call
def get_by_id(self, *, id: CompanyId) -> 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, max_items=1)
    return next(found, None)

create

create(*, company: str | Company) -> Company

Creates a new company entity.

Parameters:

Name	Type	Description	Default
company	Union[str, Company]	The company name or Company object to create.	required

Returns:

Type	Description
Company	The created Company object.

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

def create(self, *, company: str | Company) -> Company:
    """
    Creates a new company entity.

    Parameters
    ----------
    company : Union[str, Company]
        The company name or Company object to create.

    Returns
    -------
    Company
        The created Company object.
    """
    if isinstance(company, str):
        company = Company(name=company)

    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

get_or_create

get_or_create(*, company: str | Company) -> Company

Retrieves a company by its name or creates it if it does not exist.

Parameters:

Name	Type	Description	Default
company	Union[str, Company]	The company name or Company object to retrieve or create.	required

Returns:

Type	Description
Company	The Company object if found or created.

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

def get_or_create(self, *, company: str | Company) -> Company:
    """
    Retrieves a company by its name or creates it if it does not exist.

    Parameters
    ----------
    company : Union[str, Company]
        The company name or Company object to retrieve or create.

    Returns
    -------
    Company
        The Company object if found or created.
    """
    if isinstance(company, str):
        company = Company(name=company)
    found = self.get_by_name(name=company.name, exact_match=True)
    if found:
        return found
    else:
        return self.create(company=company)

delete

delete(*, id: CompanyId) -> 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

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

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