Api reference

components.customer_admin.public.actions ¶

edit_admin_entities ¶

edit_admin_entities(
    admin_user_id,
    added_account_ids,
    added_company_ids,
    removed_account_ids,
    removed_company_ids,
)

Source code in components/customer_admin/public/actions.py

def edit_admin_entities(  # noqa: D103
    admin_user_id: str,
    added_account_ids: set[str],
    added_company_ids: set[str],
    removed_account_ids: set[str],
    removed_company_ids: set[str],
) -> None:
    if (
        not added_account_ids
        and not added_company_ids
        and not removed_account_ids
        and not removed_company_ids
    ):
        # nothing to update, exiting early
        return

    app_name = get_current_app_name()
    match app_name:
        case AppName.ALAN_BE:
            from components.be.public.global_customer_dashboard.admin import (
                edit_admin_entities as be_edit_admin_entities,
            )

            be_edit_admin_entities(
                admin_user_id=UUID(admin_user_id),
                added_account_ids={UUID(a) for a in added_account_ids},
                added_company_ids={UUID(c) for c in added_company_ids},
                removed_account_ids={UUID(a) for a in removed_account_ids},
                removed_company_ids={UUID(c) for c in removed_company_ids},
            )
        case AppName.ALAN_ES:
            from components.es.public.global_customer_dashboard.admin import (
                edit_admin_entities as es_edit_admin_entities,
            )

            es_edit_admin_entities(
                admin_user_id=UUID(admin_user_id),
                added_company_ids={UUID(c) for c in added_company_ids},
                removed_company_ids={UUID(c) for c in removed_company_ids},
            )
        case _:
            raise NotImplementedError(
                f"can't find edit_admin_entities for app {current_app}"
            )

edit_admin_responsibilities ¶

edit_admin_responsibilities(
    admin_user_id,
    account_id,
    responsibilities_to_companies,
    responsibilities_to_accounts,
    responsibilities_to_scopes,
)

Source code in components/customer_admin/public/actions.py

def edit_admin_responsibilities(  # noqa: D103
    admin_user_id: str,
    account_id: UUID | None,
    responsibilities_to_companies: dict[AdminResponsibility, list[str]],
    responsibilities_to_accounts: dict[AdminResponsibility, list[str]],
    responsibilities_to_scopes: dict[AdminResponsibility, list[str]] | None,
) -> None:
    app_name = get_current_app_name()
    match app_name:
        case AppName.ALAN_FR:
            from components.fr.internal.business_logic.global_customer_dashboard.admin import (  # noqa: ALN043, ALN039
                edit_admin_responsibilities_and_publish_event,
            )

            edit_admin_responsibilities_and_publish_event(
                admin_user_id=int(admin_user_id),
                account_id=account_id,
                responsibilities_to_accounts={
                    resp: {UUID(a) for a in account_list}
                    for resp, account_list in responsibilities_to_accounts.items()
                },
                responsibilities_to_companies={
                    resp: {int(c) for c in company_list}
                    for resp, company_list in responsibilities_to_companies.items()
                },
                responsibilities_to_scopes={
                    resp: {UUID(s) for s in scope_list}
                    for resp, scope_list in responsibilities_to_scopes.items()
                }
                if responsibilities_to_scopes
                else None,
            )
        case AppName.ALAN_BE:
            raise NotImplementedError("can't edit admin responsibilities in BE")
        case AppName.ALAN_ES:
            raise NotImplementedError("can't edit admin responsibilities in ES")
        case _:
            raise NotImplementedError(
                f"can't find edit_admin_responsibilities for app {current_app}"
            )

on_completed_customer_admin_onboarding ¶

on_completed_customer_admin_onboarding(onboarding)

Called when a customer admin has completed onboarding, used to create the admin rights :param onboarding: The onboarding entity containing the information of the invitations :return: nothing

Source code in components/customer_admin/public/actions.py

def on_completed_customer_admin_onboarding(onboarding: CustomerAdminOnboarding) -> None:
    """
    Called when a customer admin has completed onboarding, used to create the admin rights
    :param onboarding:  The onboarding entity containing the information of the invitations
    :return: nothing
    """
    from components.customer_admin.internal.business_logic.actions.customer_admin import (
        create_admin_rights,
    )

    create_admin_rights(onboarding)

promote_user_to_admin ¶

promote_user_to_admin(
    *,
    user_id,
    invited_user_id,
    company_ids,
    account_ids,
    responsibilities_dict
)

Source code in components/customer_admin/public/actions.py

def promote_user_to_admin(  # noqa: D103
    *,  # forcing keyword argument as there's too many arguments here
    user_id: str,
    invited_user_id: str,
    company_ids: list[str],
    account_ids: list[UUID],
    responsibilities_dict: dict[str, bool],
) -> None:
    app_name = get_current_app_name()

    if len(company_ids) > 0 and len(account_ids) > 0:
        raise BaseErrorCode.invalid_arguments()

    match app_name:
        case AppName.ALAN_FR:
            from components.fr.internal.business_logic.global_customer_dashboard.admin import (  # noqa: ALN043, ALN039
                promote_user_to_admin_and_publish_event as fr_promote_user_to_admin_and_publish_event,
            )

            return fr_promote_user_to_admin_and_publish_event(
                user_id=int(user_id),
                invited_user_id=int(invited_user_id),
                company_ids=[int(company_id) for company_id in company_ids],
                account_ids=account_ids,
                responsibilities_dict=responsibilities_dict,
            )
        case AppName.ALAN_BE:
            from components.be.public.global_customer_dashboard.admin import (
                promote_user_to_admin as be_promote_user_to_admin,
            )

            return be_promote_user_to_admin(
                company_ids=[UUID(company_id) for company_id in company_ids],
                account_id=first_or_none(account_ids),
                invited_user_id=UUID(invited_user_id),
            )
        case AppName.ALAN_ES:
            from components.es.public.global_customer_dashboard.admin import (
                promote_user_to_admin as es_promote_user_to_admin,
            )

            return es_promote_user_to_admin(
                company_ids=[UUID(company_id) for company_id in company_ids],
                invited_user_id=UUID(invited_user_id),
            )
        case _:
            raise NotImplementedError(f"can't find invite_admin for app {current_app}")

promote_user_to_admin_with_responsibilities ¶

promote_user_to_admin_with_responsibilities(
    user_id, invited_user_id, responsibility_to_entities
)

Source code in components/customer_admin/public/actions.py

def promote_user_to_admin_with_responsibilities(  # noqa: D103
    user_id: str,
    invited_user_id: str,
    responsibility_to_entities: AdminEntitiesByResponsibility,
) -> None:
    app_name = get_current_app_name()
    match app_name:
        case AppName.ALAN_FR:
            from components.fr.internal.business_logic.global_customer_dashboard.admin import (  # noqa: ALN043, ALN039
                promote_user_to_admin_with_responsibilities_and_publish_event as fr_promote_user_to_admin_with_responsibilities_and_publish_event,
            )

            return fr_promote_user_to_admin_with_responsibilities_and_publish_event(
                user_id=int(user_id),
                invited_user_id=int(invited_user_id),
                responsibility_to_entities=responsibility_to_entities,
            )
        case _:
            raise NotImplementedError(
                f"can't find promote_user_to_admin_with_responsibilities for app {current_app}"
            )

remove_admin ¶

remove_admin(user_id, context_account_id=None)

Source code in components/customer_admin/public/actions.py

def remove_admin(user_id: str, context_account_id: UUID | None = None) -> None:  # noqa: D103
    current_user_id = g.current_user.id
    if str(current_user_id) == user_id:
        raise BaseErrorCode.invalid_arguments(
            message="You can't remove yourself as admin"
        )
    app_name = get_current_app_name()
    match app_name:
        case AppName.ALAN_FR:
            from components.fr.internal.business_logic.global_customer_dashboard.admin import (  # noqa: ALN043, ALN039
                remove_admin_and_publish_event,
            )

            return remove_admin_and_publish_event(
                admin_user_id=current_user_id,
                target_admin_user_id=int(user_id),
                context_account_id=context_account_id,
            )
        case AppName.ALAN_BE:
            from components.be.public.global_customer_dashboard.admin import (
                remove_admin as be_remove_admin,
            )

            return be_remove_admin(
                admin_user_id=current_user_id, target_admin_user_id=UUID(user_id)
            )
        case AppName.ALAN_ES:
            from components.es.public.global_customer_dashboard.admin import (
                remove_admin as es_remove_admin,
            )

            return es_remove_admin(
                admin_user_id=current_user_id, target_admin_user_id=UUID(user_id)
            )
        case _:
            raise NotImplementedError(f"can't find remove_admin for app {current_app}")

components.customer_admin.public.admined_entities ¶

AdminedEntitiesQueryApi ¶

AdminedEntitiesQueryApi(
    *,
    company_admin_model,
    account_admin_model=None,
    scope_admin_model=None,
    get_account_id_to_company_ids_fn,
    get_company_id_to_operational_scope_ids_fn=None,
    get_user_active_admined_companies_ids_fn=None,
    group_scope_ids_by_company_id_fn=None
)

Design TLDR: SDK approach into Modular Monolith

This class takes a local CompanyAdmin model as argument to the constructor to centralize all the logic and make it easier to later migrate to a modular monolith component. Caveats: - this approach is only possible because FR, BE and ES data models are similar - this code is brittle to any data model change. We can strengthen it by defining a BaseAbstractModel that will be extended by the 3 countries CompanyAdmin models.

Source code in components/customer_admin/public/admined_entities.py

def __init__(
    self,
    *,  # Prevent using positional arguments
    # The following 2 arguments can't be typed using a Protocol.
    # If we commit to isolating the company admin model, we can use an abstract base class
    # or go all the way into a proper component
    company_admin_model: Any,
    account_admin_model: Any = None,
    scope_admin_model: Any = None,
    get_account_id_to_company_ids_fn: Callable[
        [Iterable[str]],
        Mapping[str, Iterable[str]],
    ],
    get_company_id_to_operational_scope_ids_fn: Callable[
        [Iterable[str]],
        Mapping[str, Iterable[str]],
    ]
    | None = None,
    get_user_active_admined_companies_ids_fn: Callable[[str], CompanyIds]
    | None = None,
    group_scope_ids_by_company_id_fn: Callable[
        [Iterable[str]], Mapping[str, Iterable[str]]
    ]
    | None = None,
) -> None:
    # Making explicit what's required from external code
    # be it country specific code or other modular monolith modules
    self.company_admin_model = company_admin_model
    self.account_admin_model = account_admin_model
    self.scope_admin_model = scope_admin_model
    self.get_account_id_to_company_ids_fn = get_account_id_to_company_ids_fn
    self.get_company_id_to_operational_scope_ids_fn = (
        get_company_id_to_operational_scope_ids_fn
    )
    self.get_user_active_admined_companies_ids_fn = (
        get_user_active_admined_companies_ids_fn
    )
    self.group_scope_ids_by_company_id_fn = group_scope_ids_by_company_id_fn

account_admin_model `instance-attribute` ¶

account_admin_model = account_admin_model

company_admin_model `instance-attribute` ¶

company_admin_model = company_admin_model

get_account_id_to_company_ids_fn `instance-attribute` ¶

get_account_id_to_company_ids_fn = (
    get_account_id_to_company_ids_fn
)

get_admin_account_trees ¶

get_admin_account_trees(user_id, account_id)

:param user_id: the admin user id :param account_id: filtering account id :return: a list of one or several tree of all the entities in the account admined by the user. There might be several trees if the user is admin of several companies in the account without being account admin.

Source code in components/customer_admin/public/admined_entities.py

def get_admin_account_trees(
    self, user_id: str, account_id: UUID
) -> Iterable[RawAdminedEntitiesTree]:
    """
    :param user_id: the admin user id
    :param account_id: filtering account id
    :return: a list of one or several tree of all the entities in the account admined by the user.
             There might be several trees if the user is admin of several companies in the account
             without being account admin.
    """
    (
        admined_account_ids,
        directly_admined_company_ids,
        directly_admined_operational_scope_ids,
    ) = self.get_admined_entity_ids(user_id=user_id)

    account_org_tree = (
        get_app_dependency()
        .get_account_org_tree_query_api()
        .get_account_org_tree(account_id)
    )
    if str(account_id) in admined_account_ids:
        return [_account_org_tree_to_admined_entities_tree(account_org_tree)]

    entities_in_account_trees = []

    for company_id in directly_admined_company_ids:
        if company_node := account_org_tree.find_first(
            match=lambda node, co_id=company_id: node.data.id == co_id  # type:ignore[misc]
            and node.data.type == EntityType.company
        ):
            entities_in_account_trees.append(
                _account_org_tree_to_admined_entities_tree(
                    account_org_tree,
                    account_org_tree[
                        AccountOrgTreeNode(
                            id=company_id,
                            type=EntityType.company,
                            display_name=company_node.data.display_name,
                        )
                    ],
                )
            )
    for scope_id in directly_admined_operational_scope_ids:
        if scope_node := account_org_tree.find_first(
            match=lambda node, sco_id=scope_id: node.data.id == sco_id  # type:ignore[misc]
            and node.data.type == EntityType.operational_scope
        ):
            entities_in_account_trees.append(
                _account_org_tree_to_admined_entities_tree(
                    account_org_tree,
                    account_org_tree[
                        AccountOrgTreeNode(
                            id=scope_id,
                            type=EntityType.operational_scope,
                            display_name=scope_node.data.display_name,
                        )
                    ],
                )
            )

    return entities_in_account_trees

get_admined_entities_context_accounts ¶

get_admined_entities_context_accounts(user_id)

:param user_id: :return: a set of the accounts in which the user is admin of something. Could be the whole account, one or several companies, one or several operational scopes.

Source code in components/customer_admin/public/admined_entities.py

def get_admined_entities_context_accounts(
    self, user_id: str
) -> set[ContextAccount]:
    """

    :param user_id:
    :return: a set of the accounts in which the user is admin of something.
    Could be the whole account, one or several companies, one or several operational scopes.
    """
    (account_ids, company_ids, scope_ids) = self.get_admined_entity_ids(
        user_id=user_id
    )
    accounts = (
        get_app_dependency()
        .get_account_org_tree_query_api()
        .get_account_for_entities(
            [(account_id, EntityType.account) for account_id in account_ids]
            + [(company_id, EntityType.company) for company_id in company_ids]
            + [(scope_id, EntityType.operational_scope) for scope_id in scope_ids]
        )
    )
    return {
        ContextAccount(id=UUID(account.id), display_name=account.display_name)
        for account in accounts
    }

get_admined_entities_for_user ¶

get_admined_entities_for_user(user_id)

Retrieves a list of lightweight admined entities for the given user ID. :param user_id: :return: a list of AdminedEntity

Source code in components/customer_admin/public/admined_entities.py

def get_admined_entities_for_user(self, user_id: str) -> list[AdminedEntity]:
    """
    Retrieves a list of lightweight admined entities for the given user ID.
    :param user_id:
    :return: a list of AdminedEntity
    """
    # 1. Get the entities the user directly admins
    (
        admined_account_ids,
        directly_admined_company_ids,
        directly_admined_operational_scope_ids,
    ) = self.get_admined_entity_ids(user_id=user_id)

    # Due to data model differences,
    # - for BE, all_admined_company_ids == directly_admined_company_ids
    # - for FR, directly_admined_company_ids is a subset of all_admined_company_ids
    # - for ES, we don't have an account admin concept
    account_id_to_company_ids = {
        account_id: company_ids
        for account_id, company_ids in self.get_account_id_to_company_ids_fn(
            admined_account_ids
        ).items()
    }

    # If set, we filter out inactive admined companies
    if self.get_user_active_admined_companies_ids_fn:
        filter_on_company_id = self.get_user_active_admined_companies_ids_fn(
            user_id
        )
        directly_admined_company_ids = {
            company_id
            for company_id in directly_admined_company_ids
            if company_id in filter_on_company_id
        }
        for _account_id, _company_ids in account_id_to_company_ids.items():
            if _account_id:
                account_id_to_company_ids[_account_id] = [
                    company_id
                    for company_id in _company_ids
                    if company_id in filter_on_company_id
                ]

    # 2. Build the full tree of what the user admins: directly & indirectly
    (all_admined_company_ids, all_admined_operational_scope_ids) = (
        self.get_list_of_all_sub_admined_entities_from_directly_admined_ones(
            admined_account_ids,
            directly_admined_company_ids,
            directly_admined_operational_scope_ids,
            account_id_to_company_ids,
        )
    )

    # 3. Group directly admined scopes of same companies in a company_for_operational_scope entity
    directly_admined_scopes_by_company_id = (
        self.group_scope_ids_by_company_id_fn(
            directly_admined_operational_scope_ids
        )
        if self.group_scope_ids_by_company_id_fn
        else {}
    )

    # 4. Build the list of entities to return
    entities = []
    for operational_scope_ids in all_admined_operational_scope_ids:
        entities.append(
            AdminedEntity(
                type=AdminedEntityType.operational_scope,
                id=operational_scope_ids,
            )
        )

    for company_id in all_admined_company_ids:
        entities.append(
            AdminedEntity(
                type=AdminedEntityType.single_company,
                id=company_id,
            )
        )

    for account_id in admined_account_ids:
        entities.append(
            AdminedEntity(
                type=AdminedEntityType.account,
                id=account_id,
            )
        )

    for company_id in directly_admined_scopes_by_company_id.keys():
        entities.append(
            AdminedEntity(
                type=AdminedEntityType.company_for_operational_scope,
                id=company_id,
            )
        )

    return entities

get_admined_entity_ids ¶

get_admined_entity_ids(user_id)

:param user_id: :return: all the ids of entities the user is admin of. /!\ There's no transitivity logic: if the user is account admin, the company_ids set won't contain that account company ids.

Source code in components/customer_admin/public/admined_entities.py

def get_admined_entity_ids(
    self,
    user_id: str,
) -> tuple[AccountIds, CompanyIds, OperationalScopeIds]:
    r"""

    :param user_id:
    :return: all the ids of entities the user is admin of.
    /!\ There's no transitivity logic: if the user is account admin,
    the company_ids set won't contain that account company ids.
    """
    scope_admin_entries = []  # type: ignore[var-annotated]
    if self.scope_admin_model is not None:
        scope_admin_entries = current_session.scalars(  # type: ignore[assignment]
            select(self.scope_admin_model).filter(
                self.scope_admin_model.user_id == user_id,
                self.scope_admin_model.is_not_ended,
                self.scope_admin_model.operational_scope_id.isnot(None),
            )
        ).all()

    scope_admin_ids = {
        scope_admin_entry.id for scope_admin_entry in scope_admin_entries
    }

    company_admin_entries = current_session.scalars(
        select(self.company_admin_model).filter(
            self.company_admin_model.user_id == user_id,
            self.company_admin_model.is_not_ended,
            self.company_admin_model.id.not_in(
                scope_admin_ids
            ),  # Act as an XOR between scope & company for french CompanyAdmin model. Remove when PAY-313 is done
        )
    ).all()

    account_admins = (
        current_session.scalars(
            select(self.account_admin_model).filter(
                self.account_admin_model.user_id == user_id,
                self.account_admin_model.is_not_ended,
            )
        ).all()
        if self.account_admin_model
        else []
    )

    account_ids = {
        str(company_admin_entry.account_id)
        for company_admin_entry in company_admin_entries
        # Not all companies have the same data model so we need the getattr
        if getattr(company_admin_entry, "account_id", None)
    } | {str(account_admin.account_id) for account_admin in account_admins}

    company_ids = {
        str(company_admin_entry.company_id)
        for company_admin_entry in company_admin_entries
        if company_admin_entry.company_id
    }

    scope_ids = {
        str(scope_admin_entry.operational_scope_id)
        for scope_admin_entry in scope_admin_entries
    }

    return account_ids, company_ids, scope_ids

get_company_id_to_operational_scope_ids_fn `instance-attribute` ¶

get_company_id_to_operational_scope_ids_fn = (
    get_company_id_to_operational_scope_ids_fn
)

get_list_of_all_sub_admined_entities_from_directly_admined_ones ¶

get_list_of_all_sub_admined_entities_from_directly_admined_ones(
    directly_admined_account_ids,
    directly_admined_company_ids,
    directly_admined_operational_scope_ids,
    account_id_to_company_ids=None,
)

Builds the full list of companies and operational scopes ids the user admins, directly and indirectly. :param account_id_to_company_ids: dictionary of company ids per account id. Optional. Useful if we want to filter out some companies :param directly_admined_account_ids: list of account ids the user directly admins :param directly_admined_company_ids: list of company ids the user directly admins :param directly_admined_operational_scope_ids: list of operational scope ids the user directly admins :return:

Source code in components/customer_admin/public/admined_entities.py

def get_list_of_all_sub_admined_entities_from_directly_admined_ones(
    self,
    directly_admined_account_ids: set[str],
    directly_admined_company_ids: set[str],
    directly_admined_operational_scope_ids: set[str],
    account_id_to_company_ids: Mapping[str, Iterable[str]] | None = None,
) -> tuple[CompanyIds, OperationalScopeIds]:
    """
    Builds the full list of companies and operational scopes ids the user admins, directly and indirectly.
    :param account_id_to_company_ids: dictionary of company ids per account id. Optional. Useful if we want to filter out some companies
    :param directly_admined_account_ids: list of account ids the user directly admins
    :param directly_admined_company_ids: list of company ids the user directly admins
    :param directly_admined_operational_scope_ids: list of operational scope ids the user directly admins
    :return:
    """
    if account_id_to_company_ids is None:
        account_id_to_company_ids = self.get_account_id_to_company_ids_fn(
            directly_admined_account_ids
        )

    admined_via_account_company_ids = {
        company_id
        for company_id in flatten(account_id_to_company_ids.values())
        if company_id
    }
    all_admined_company_ids = (
        directly_admined_company_ids | admined_via_account_company_ids
    )
    company_id_to_operational_scope_ids = (
        self.get_company_id_to_operational_scope_ids_fn(all_admined_company_ids)
        if self.get_company_id_to_operational_scope_ids_fn is not None
        else {}
    )
    admined_via_company_operational_scope_ids = {
        operational_scope_id
        for operational_scope_id in flatten(
            company_id_to_operational_scope_ids.values()
        )
        if operational_scope_id
    }
    all_admined_operational_scope_ids = (
        directly_admined_operational_scope_ids
        | admined_via_company_operational_scope_ids
    )
    return all_admined_company_ids, all_admined_operational_scope_ids

get_user_active_admined_companies_ids_fn `instance-attribute` ¶

get_user_active_admined_companies_ids_fn = (
    get_user_active_admined_companies_ids_fn
)

group_scope_ids_by_company_id_fn `instance-attribute` ¶

group_scope_ids_by_company_id_fn = (
    group_scope_ids_by_company_id_fn
)

scope_admin_model `instance-attribute` ¶

scope_admin_model = scope_admin_model

AdminedEntityTreeNode `dataclass` ¶

AdminedEntityTreeNode(
    *,
    type,
    id,
    display_name,
    parent_account_id=None,
    parent_company_id=None
)

Bases: AccountOrgTreeNode, DataClassJsonMixin

Represents an entity that a user is admin of directly or transitively (account admins are also admin of all the companies of their account)

display_name `instance-attribute` ¶

display_name

from_account_org_tree_node `classmethod` ¶

from_account_org_tree_node(account_org_tree_node)

Helper function converting an AccountOrgTreeNode into an AdminedEntityTreeNode

Source code in components/customer_admin/public/entities.py

@classmethod
def from_account_org_tree_node(
    cls, account_org_tree_node: "Node[AccountOrgTreeNode]"
) -> Self:
    """
    Helper function converting an AccountOrgTreeNode into an AdminedEntityTreeNode
    """
    match account_org_tree_node.data.type:
        case EntityType.account:
            return cls(
                type=account_org_tree_node.data.type,
                id=account_org_tree_node.data.id,
                display_name=account_org_tree_node.data.display_name,
            )
        case EntityType.company:
            return cls(
                type=account_org_tree_node.data.type,
                id=account_org_tree_node.data.id,
                display_name=account_org_tree_node.data.display_name,
                parent_account_id=mandatory(account_org_tree_node.parent).data.id,
            )
        case EntityType.operational_scope:
            return cls(
                type=account_org_tree_node.data.type,
                id=account_org_tree_node.data.id,
                display_name=account_org_tree_node.data.display_name,
                parent_account_id=mandatory(
                    mandatory(account_org_tree_node.parent).parent
                ).data.id,
                parent_company_id=mandatory(account_org_tree_node.parent).data.id,
            )
        case _:
            assert_never(account_org_tree_node.data.type)

id `instance-attribute` ¶

id

parent_account_id `class-attribute` `instance-attribute` ¶

parent_account_id = None

parent_company_id `class-attribute` `instance-attribute` ¶

parent_company_id = None

type `instance-attribute` ¶

type

components.customer_admin.public.customer_admin_read_service ¶

CustomerAdminReadService ¶

CustomerAdminReadService(
    customer_admin_repository=None,
    account_org_tree_query_api=None,
)

Utility global class that provides methods to query a given user admin rights.

Source code in components/customer_admin/public/customer_admin_read_service.py

def __init__(
    self,
    customer_admin_repository: "CustomerAdminRepository| None" = None,
    account_org_tree_query_api: "AccountOrgTreeQueryApiABC | None" = None,
) -> None:
    customer_admin_app_dependency = get_app_dependency()

    self.customer_admin_repository = (
        customer_admin_repository
        or customer_admin_app_dependency.get_customer_admin_repository()
    )
    self.account_org_tree_query_api = (
        account_org_tree_query_api
        or customer_admin_app_dependency.get_account_org_tree_query_api()
    )

account_org_tree_query_api `instance-attribute` ¶

account_org_tree_query_api = (
    account_org_tree_query_api
    or get_account_org_tree_query_api()
)

customer_admin_repository `instance-attribute` ¶

customer_admin_repository = (
    customer_admin_repository
    or get_customer_admin_repository()
)

get_admins_page_for_account ¶

get_admins_page_for_account(
    account_id,
    company_ids,
    operational_scope_ids,
    cursor,
    limit,
)

:param account_id: account id for which to get the admins. :param company_ids: if set, only the admins that are admin of the given companies will be returned :param operational_scope_ids: if set, only the admins that are admin of the given operational scope will be returned :param cursor: opaque pagination cursor. It happens to be a number for now, but it could change :param limit: number of admins to return per page (counting distinct user profiles)

:return: Paginated list of all admins that are admin in the given account

Source code in components/customer_admin/public/customer_admin_read_service.py

def get_admins_page_for_account(
    self,
    account_id: UUID,
    company_ids: Collection[str] | None,
    operational_scope_ids: Collection[str] | None,
    cursor: str | None,
    limit: int,
) -> CustomerAdminsPage:
    """
    :param account_id: account id for which to get the admins.
    :param company_ids: if set, only the admins that are admin
     of the given companies will be returned
     :param operational_scope_ids: if set, only the admins that are admin
     of the given operational scope will be returned
     :param cursor: opaque pagination cursor.
     It happens to be a number for now, but it could change
     :param limit: number of admins to return per page (counting distinct user profiles)

    :return: Paginated list of all admins that are admin in the given account
    """
    assert limit > 0, "limit must be positive"
    assert company_ids is None or len(company_ids) > 0, (
        "company_ids must be non-empty or None,"
        "empty company_ids would lead to an empty list of admins"
    )

    account_org_tree = self.account_org_tree_query_api.get_account_org_tree(
        account_id
    )

    company_ids = set(company_ids) if company_ids else set()
    assert all(
        LeanEntityInfo(id=str(company_id), type=EntityType.company)
        in account_org_tree
        for company_id in company_ids
    ), "company_ids must be in the account corresponding to the account_id"

    operational_scope_ids = (
        set(operational_scope_ids) if operational_scope_ids else set()
    )
    assert all(
        LeanEntityInfo(
            id=str(operational_scope_id), type=EntityType.operational_scope
        )
        in account_org_tree
        for operational_scope_id in operational_scope_ids
    ), (
        "operational_scope_ids must be in the account corresponding to the account_id"
    )

    for company_id in company_ids:
        scopes = account_org_tree[
            LeanEntityInfo(id=str(company_id), type=EntityType.company)
        ].children
        for scope in scopes:
            operational_scope_ids.add(scope.data.id)

    admins_page = self.customer_admin_repository.get_admins_for_account(
        account_id=account_id,
        company_ids=company_ids,
        operational_scope_ids=operational_scope_ids,
        cursor=cursor,
        limit=limit,
    )

    def directly_admined_entity_to_admined_entities_tree(
        directly_admined_entities: set[DirectlyAdminedEntity],
    ) -> AdminedEntitiesForest:
        admined_entities_by_type: dict[EntityType, list[DirectlyAdminedEntity]] = (
            group_by(directly_admined_entities, key_fn=lambda en: en.type)
        )
        return _map_directly_admined_entities_to_admined_entities_forest(
            admined_entities_by_type, [account_org_tree]
        )

    customer_admins = [
        CustomerAdmin(
            profile_id=admin.profile_id,
            first_name=admin.first_name,
            last_name=admin.last_name,
            admined_entities=directly_admined_entity_to_admined_entities_tree(
                admin.directly_admined_entities
            ),
        )
        for admin in admins_page.admins
    ]

    return CustomerAdminsPage(
        admins=customer_admins,
        meta=CursorInfoMeta(next_cursor=admins_page.next_cursor),
    )

get_user_profile_admined_entities_forest ¶

get_user_profile_admined_entities_forest(
    profile_id, filter_account_id=None
)

:param profile_id: user profile id :param filter_account_id: if set, only the companies and operational scopes under that account will be included in the returned tree data structure.

:return: a list of trees of all the entities in the account admined by the user. There might be several trees when: - the user is admin of several companies in the account without being account admin - and/or the user is admin in several accounts

Source code in components/customer_admin/public/customer_admin_read_service.py

def get_user_profile_admined_entities_forest(
    self, profile_id: UUID, filter_account_id: UUID | None = None
) -> AdminedEntitiesForest:
    """
    :param profile_id: user profile id
    :param filter_account_id: if set, only the companies and operational scopes
            under that account will be included in the returned tree data structure.

    :return: a list of trees of all the entities in the account admined by the user.
             There might be several trees when:
            - the user is admin of several companies in the account without being account admin
            - and/or the user is admin in several accounts
    """
    raw_admined_entities = (
        self.customer_admin_repository.get_admined_entities_for_profile(profile_id)
    )
    admined_entities_by_type: dict[EntityType, list[DirectlyAdminedEntity]] = (
        group_by(raw_admined_entities, key_fn=lambda en: en.type)
    )

    account_org_trees = [
        tree
        for tree in self.account_org_tree_query_api.get_account_org_tree_for_entities(
            raw_admined_entities
        ).values()
        if filter_account_id is None
        or LeanEntityInfo(id=str(filter_account_id), type=EntityType.account)
        in tree
    ]

    return _map_directly_admined_entities_to_admined_entities_forest(
        admined_entities_by_type, account_org_trees, filter_account_id
    )

is_user_admin_of_account ¶

is_user_admin_of_account(profile_id, account_id)

:param profile_id: :param account_id: :return: true if the user is admin of the account, false otherwise.

Source code in components/customer_admin/public/customer_admin_read_service.py

def is_user_admin_of_account(self, profile_id: UUID, account_id: UUID) -> bool:
    """

    :param profile_id:
    :param account_id:
    :return: true if the user is admin of the account, false otherwise.
    """
    return (
        account_id
        in self.get_user_profile_admined_entities_forest(profile_id).account_ids
    )

is_user_admin_of_company ¶

is_user_admin_of_company(profile_id, company_id)

:param profile_id: :param company_id: :return: true if the user is admin of the company, false otherwise.

Source code in components/customer_admin/public/customer_admin_read_service.py

def is_user_admin_of_company(self, profile_id: UUID, company_id: str) -> bool:
    """

    :param profile_id:
    :param company_id:
    :return: true if the user is admin of the company, false otherwise.
    """
    return (
        company_id
        in self.get_user_profile_admined_entities_forest(profile_id).company_ids
    )

is_user_admin_of_entities ¶

is_user_admin_of_entities(profile_id, entities)

:param profile_id: :param entities: :return: :return: true if the user is admin of all the entities, false otherwise.

Source code in components/customer_admin/public/customer_admin_read_service.py

def is_user_admin_of_entities(
    self, profile_id: UUID, entities: Iterable[LeanEntityInfo]
) -> bool:
    """

    :param profile_id:
    :param entities:
    :return: :return: true if the user is admin of all the entities, false otherwise.
    """
    forest = self.get_user_profile_admined_entities_forest(profile_id)
    return all(entity in forest for entity in entities)

components.customer_admin.public.dependencies ¶

CUSTOMER_ADMIN_COMPONENT_NAME `module-attribute` ¶

CUSTOMER_ADMIN_COMPONENT_NAME = 'customer_admin'

CustomerAdminDependency ¶

CustomerAdminDependency defines the interface that apps using the customer_admin component need to implement

get_account_org_tree_query_api ¶

get_account_org_tree_query_api()

:return: The method will return a tree data structure of the account entities (account, companies, operational scopes )

Source code in components/customer_admin/public/dependencies.py

def get_account_org_tree_query_api(self) -> "AccountOrgTreeQueryApi":
    """
    :return: The method will return a tree data structure of the account entities
    (account, companies, operational scopes
    )
    """
    raise NotImplementedError()

get_admined_entities_api ¶

get_admined_entities_api()

:return: The method will return a helper class that abstracts the DB queries required to fetch a user list of of admined entities

Source code in components/customer_admin/public/dependencies.py

def get_admined_entities_api(self) -> "AdminedEntitiesQueryApi":
    """
    :return: The method will return a helper class that abstracts the DB queries required
    to fetch a user list of of admined entities
    """
    raise NotImplementedError()

get_customer_admin_repository ¶

get_customer_admin_repository()

:return: The method will return a class that encapsulate DB queries

Source code in components/customer_admin/public/dependencies.py

def get_customer_admin_repository(self) -> "CustomerAdminRepository":
    """
    :return: The method will return a class that encapsulate DB queries
    """
    raise NotImplementedError()

get_app_dependency ¶

get_app_dependency()

Retrieves at runtime the account_org_tree dependency set by set_app_dependency

Source code in components/customer_admin/public/dependencies.py

def get_app_dependency() -> CustomerAdminDependency:
    """Retrieves at runtime the account_org_tree dependency set by set_app_dependency"""
    from flask import current_app

    app = cast("CustomFlask", current_app)
    return cast(
        "CustomerAdminDependency",
        app.get_component_dependency(CUSTOMER_ADMIN_COMPONENT_NAME),
    )

set_app_dependency ¶

set_app_dependency(dependency)

Sets the account_org_tree dependency to the app so it can be accessed within this component at runtime

Source code in components/customer_admin/public/dependencies.py

def set_app_dependency(dependency: CustomerAdminDependency) -> None:
    """
    Sets the account_org_tree dependency to the app so it can be accessed within this component at runtime
    """
    from flask import current_app

    cast("CustomFlask", current_app).add_component_dependency(
        CUSTOMER_ADMIN_COMPONENT_NAME, dependency
    )

components.customer_admin.public.entities ¶

AccountIds `module-attribute` ¶

AccountIds = set[str]

AdminEntitiesByResponsibility `dataclass` ¶

AdminEntitiesByResponsibility(
    contract_manager,
    people_manager,
    payroll_manager,
    invoice_manager,
    wellbeing_referent,
)

Bases: DataClassJsonMixin

Admin entities by responsibility.

all_entities `property` ¶

all_entities

Get all entities. :return: a list of AdminEntities for which the admin will have some responsibility.

contract_manager `instance-attribute` ¶

contract_manager

invoice_manager `instance-attribute` ¶

invoice_manager

payroll_manager `instance-attribute` ¶

payroll_manager

people_manager `instance-attribute` ¶

people_manager

wellbeing_referent `instance-attribute` ¶

wellbeing_referent

AdminedEntitiesForest ¶

AdminedEntitiesForest()

Bases: TypedTree[ForestData]

Hierarchical data structure containing all the entities a given user is admin of both - directly (correspond to tree root node(s)) - or indirectly (corresponding to tree leaves and intermediate nodes). with context_account nodes to regroup all admined entities under a given account, has admined entities, without necessarilyy being admin of the whole account.

The forest name is used to differentiate from the existing AdminedEntitiesTree and hint at the fact that this tree can contain data for several accounts and different kinds of nodes.

Source code in components/customer_admin/public/entities.py

def __init__(self) -> None:
    # Explicitly set calc_data_id to identify a tree node only by entity id and type
    super().__init__(calc_data_id=_data_id_fn)

DEFAULT_CHILD_TYPE `class-attribute` `instance-attribute` ¶

DEFAULT_CHILD_TYPE = entity

account_ids `property` ¶

account_ids

:return: set of all account ids

company_ids `property` ¶

company_ids

:return: set of all company ids

find_context_account ¶

find_context_account(account_id)

:param account_id :return: the context account node corresponding to the entity if found, None otherwise.

Source code in components/customer_admin/public/entities.py

def find_context_account(
    self, account_id: str
) -> TypedNode[ContextAccountTreeNode] | None:
    """
    :param account_id
    :return: the context account node corresponding to the entity if found,
            None otherwise.
    """
    return cast(
        "TypedNode[ContextAccountTreeNode] | None",
        self.find(
            match=lambda node: node.kind
            == AdminedEntityForestNodeKind.context_account
            and node.data.id == account_id
        ),
    )

find_entity ¶

find_entity(entity)

:param entity: defined by an id and a type :return: the node corresponding to the entity if found, None otherwise.

Source code in components/customer_admin/public/entities.py

def find_entity(
    self, entity: LeanEntityInfo
) -> TypedNode[AdminedEntityTreeNode] | None:
    """
    :param entity: defined by an id and a type
    :return: the node corresponding to the entity if found, None otherwise.
    """
    try:
        return cast("TypedNode[AdminedEntityTreeNode] | None", self[entity])
    except KeyError:
        return None

operational_scope_ids `property` ¶

operational_scope_ids

:return: set of all operational scope ids

AdminedEntityForestNodeKind ¶

Bases: AlanBaseEnum

We store several kinds of nodes to store the maximum amount of information: - context account (useful for company admins who are not account admins) - and regular entity nodes to store data about the admined entities, both directly and transitively via a parent entity (ex: account admin admins all companies of that account)

context_account `class-attribute` `instance-attribute` ¶

context_account = 'context_account'

entity `class-attribute` `instance-attribute` ¶

entity = 'entity'

AdminedEntityTreeNode `dataclass` ¶

AdminedEntityTreeNode(
    *,
    type,
    id,
    display_name,
    parent_account_id=None,
    parent_company_id=None
)

Bases: AccountOrgTreeNode, DataClassJsonMixin

Represents an entity that a user is admin of directly or transitively (account admins are also admin of all the companies of their account)

display_name `instance-attribute` ¶

display_name

from_account_org_tree_node `classmethod` ¶

from_account_org_tree_node(account_org_tree_node)

Helper function converting an AccountOrgTreeNode into an AdminedEntityTreeNode

Source code in components/customer_admin/public/entities.py

@classmethod
def from_account_org_tree_node(
    cls, account_org_tree_node: "Node[AccountOrgTreeNode]"
) -> Self:
    """
    Helper function converting an AccountOrgTreeNode into an AdminedEntityTreeNode
    """
    match account_org_tree_node.data.type:
        case EntityType.account:
            return cls(
                type=account_org_tree_node.data.type,
                id=account_org_tree_node.data.id,
                display_name=account_org_tree_node.data.display_name,
            )
        case EntityType.company:
            return cls(
                type=account_org_tree_node.data.type,
                id=account_org_tree_node.data.id,
                display_name=account_org_tree_node.data.display_name,
                parent_account_id=mandatory(account_org_tree_node.parent).data.id,
            )
        case EntityType.operational_scope:
            return cls(
                type=account_org_tree_node.data.type,
                id=account_org_tree_node.data.id,
                display_name=account_org_tree_node.data.display_name,
                parent_account_id=mandatory(
                    mandatory(account_org_tree_node.parent).parent
                ).data.id,
                parent_company_id=mandatory(account_org_tree_node.parent).data.id,
            )
        case _:
            assert_never(account_org_tree_node.data.type)

id `instance-attribute` ¶

id

parent_account_id `class-attribute` `instance-attribute` ¶

parent_account_id = None

parent_company_id `class-attribute` `instance-attribute` ¶

parent_company_id = None

type `instance-attribute` ¶

type

CompanyIds `module-attribute` ¶

CompanyIds = set[str]

ContextAccountTreeNode `dataclass` ¶

ContextAccountTreeNode(*, id, display_name)

Bases: DataClassJsonMixin

Contains data about the context account = the account in which the user profile has admined entities, without necessarilyy being admin of the whole account

display_name `instance-attribute` ¶

display_name

id `instance-attribute` ¶

id

CursorInfoMeta `dataclass` ¶

CursorInfoMeta(*, next_cursor)

Bases: DataClassJsonMixin

Store metadata info for pagination

next_cursor `instance-attribute` ¶

next_cursor

CustomerAdmin `dataclass` ¶

CustomerAdmin(
    *, profile_id, first_name, last_name, admined_entities
)

Bases: DataClassJsonMixin

Represent a customer admin

admined_entities `instance-attribute` ¶

admined_entities

first_name `instance-attribute` ¶

first_name

last_name `instance-attribute` ¶

last_name

profile_id `instance-attribute` ¶

profile_id

CustomerAdminsPage `dataclass` ¶

CustomerAdminsPage(*, admins, meta)

Bases: DataClassJsonMixin

Paginated list of customer admins

admins `instance-attribute` ¶

admins

meta `instance-attribute` ¶

meta

CustomerDashboardAdminWithEntities `dataclass` ¶

CustomerDashboardAdminWithEntities(
    id,
    email,
    first_name,
    last_name,
    type,
    company_ids,
    account_id=None,
    scope_ids=None,
)

Bases: CustomerDashboardCompanyAdmin

account_id `class-attribute` `instance-attribute` ¶

account_id = None

company_ids `instance-attribute` ¶

company_ids

scope_ids `class-attribute` `instance-attribute` ¶

scope_ids = None

CustomerDashboardCompany `dataclass` ¶

CustomerDashboardCompany(
    id, name, payroll_day_in_month, account_id
)

Bases: DataClassJsonMixin

account_id `instance-attribute` ¶

account_id

id `instance-attribute` ¶

id

name `instance-attribute` ¶

name

payroll_day_in_month `instance-attribute` ¶

payroll_day_in_month

CustomerDashboardCompanyAdmin `dataclass` ¶

CustomerDashboardCompanyAdmin(
    id, email, first_name, last_name, type
)

Bases: DataClassJsonMixin

email `instance-attribute` ¶

email

first_name `instance-attribute` ¶

first_name

id `instance-attribute` ¶

id

last_name `instance-attribute` ¶

last_name

type `instance-attribute` ¶

type

EntitiesByResponsibility `dataclass` ¶

EntitiesByResponsibility(
    contract_manager,
    people_manager,
    payroll_manager,
    invoice_manager,
    wellbeing_referent,
)

Bases: DataClassJsonMixin

contract_manager `instance-attribute` ¶

contract_manager

invoice_manager `instance-attribute` ¶

invoice_manager

payroll_manager `instance-attribute` ¶

payroll_manager

people_manager `instance-attribute` ¶

people_manager

wellbeing_referent `instance-attribute` ¶

wellbeing_referent

EntityInfo `dataclass` ¶

EntityInfo(type, id, name, company_ids)

Bases: DataClassJsonMixin

company_ids `instance-attribute` ¶

company_ids

id `instance-attribute` ¶

id

name `instance-attribute` ¶

name

type `instance-attribute` ¶

type

EntityInfoProtocol ¶

Bases: Protocol

All entities in the AdminedEntitiesForest should match this protocol.

id `instance-attribute` ¶

id

type `instance-attribute` ¶

type

ForestData `module-attribute` ¶

ForestData = ContextAccountTreeNode | AdminedEntityTreeNode

LastAdminStandingRolesInEntities `dataclass` ¶

LastAdminStandingRolesInEntities(
    contract_manager,
    people_manager,
    payroll_manager,
    invoice_manager,
    wellbeing_referent,
)

Bases: DataClassJsonMixin

Last admin standing for all entities - taking into account critical and non-critical roles

contract_manager `instance-attribute` ¶

contract_manager

invoice_manager `instance-attribute` ¶

invoice_manager

payroll_manager `instance-attribute` ¶

payroll_manager

people_manager `instance-attribute` ¶

people_manager

wellbeing_referent `instance-attribute` ¶

wellbeing_referent

OperationalScopeIds `module-attribute` ¶

OperationalScopeIds = set[str]

PageInfoMeta `dataclass` ¶

PageInfoMeta(*, cursor, has_next)

Bases: DataClassJsonMixin

cursor `instance-attribute` ¶

cursor

has_next `instance-attribute` ¶

has_next

PaginatedAdminsForAdminDashboard `dataclass` ¶

PaginatedAdminsForAdminDashboard(*, admins, meta)

Bases: DataClassJsonMixin

admins `instance-attribute` ¶

admins

meta `instance-attribute` ¶

meta

components.customer_admin.public.enums ¶

AdminResponsibility ¶

Bases: AlanBaseEnum

contract_manager `class-attribute` `instance-attribute` ¶

contract_manager = 'contract_manager'

invoice_manager `class-attribute` `instance-attribute` ¶

invoice_manager = 'invoice_manager'

payroll_manager `class-attribute` `instance-attribute` ¶

payroll_manager = 'payroll_manager'

people_manager `class-attribute` `instance-attribute` ¶

people_manager = 'people_manager'

wellbeing_referent `class-attribute` `instance-attribute` ¶

wellbeing_referent = 'wellbeing_referent'

CustomerDashboardCompanyAdminListSortField ¶

Bases: AlanBaseEnum

created_at `class-attribute` `instance-attribute` ¶

created_at = 'created_at'

email `class-attribute` `instance-attribute` ¶

email = 'email'

last_name `class-attribute` `instance-attribute` ¶

last_name = 'last_name'

components.customer_admin.public.interfaces ¶

AdminsForAccountPage `dataclass` ¶

AdminsForAccountPage(*, admins, next_cursor)

Represents a paginated list of customer admins for a specific account.

admins `instance-attribute` ¶

admins

next_cursor `instance-attribute` ¶

next_cursor

BasicCustomerAdminData `dataclass` ¶

BasicCustomerAdminData(
    *,
    profile_id,
    first_name,
    last_name,
    directly_admined_entities
)

Customer admin data.

Long term, all the data here should comes from customer admin component's internal data model

directly_admined_entities `instance-attribute` ¶

directly_admined_entities

first_name `instance-attribute` ¶

first_name

last_name `instance-attribute` ¶

last_name

profile_id `instance-attribute` ¶

profile_id

CustomerAdminRepository ¶

Bases: ABC

Public interface defining the expected methods for each country CustomerAdminRepository implementations.

get_admined_entities_for_profile `abstractmethod` ¶

get_admined_entities_for_profile(profile_id)

Only returns the direct admin rights. For instance, for an account admin, companies under that account might not be returned.

Source code in components/customer_admin/public/interfaces.py

@abstractmethod
def get_admined_entities_for_profile(
    self, profile_id: UUID
) -> set[DirectlyAdminedEntity]:
    """
    Only returns the direct admin rights. For instance, for an account admin,
    companies under that account might not be returned.
    """
    raise NotImplementedError

get_admins_for_account `abstractmethod` ¶

get_admins_for_account(
    account_id,
    company_ids,
    operational_scope_ids,
    cursor,
    limit,
)

Only returns the direct admin rights. For instance, for an account admin, companies under that account might not be returned.

:returns: an admins list page data as a tuple containing a dict mapping profile_id to a set of directly admined entities and the next cursor

Source code in components/customer_admin/public/interfaces.py

@abstractmethod
def get_admins_for_account(
    self,
    account_id: UUID,
    company_ids: Iterable[str],
    operational_scope_ids: Iterable[str],
    cursor: str
    | None,  # string instead of int to keep the door open to use keyset/token pagination
    limit: int,
) -> AdminsForAccountPage:
    """
    Only returns the direct admin rights. For instance, for an account admin,
    companies under that account might not be returned.

    :returns: an admins list page data as a tuple containing
     a dict mapping profile_id to a set of directly admined entities
     and the next cursor

    """
    raise NotImplementedError

DirectlyAdminedEntity `dataclass` ¶

DirectlyAdminedEntity(*, type, id, parent_account_id)

Bases: LeanEntityInfo

Info about an entity that is directly admined = top level entity admined by the user profile. A counter-example of indirectly admined entities is a company under an account for an account admin.

For convenience and typing simplicity, we also set the parent account id for accounts, in which case the id will match the parent_account_id.

id `instance-attribute` ¶

id

parent_account_id `instance-attribute` ¶

parent_account_id

type `instance-attribute` ¶

type

profile_id_type `module-attribute` ¶

profile_id_type = UUID