Skip to content

Api reference

components.onboarding.public.actions

cancel_person_onboarding

cancel_person_onboarding

cancel_person_onboarding(person_onboarding_id, commit=True)

Cancels the onboarding for the person onboarding. You can only cancel a non-completed onboarding

Source code in components/onboarding/internal/actions/cancel_person_onboarding.py
def cancel_person_onboarding(person_onboarding_id: UUID, commit: bool = True) -> None:
    """
    Cancels the onboarding for the person onboarding. You can only cancel a
    non-completed onboarding
    """
    person_onboarding = PersonOnboardingModelBroker.find_by_id(
        person_onboarding_id, with_cancelled=True
    ).one()

    if (
        person_onboarding.onboarding_status == PersonOnboardingStatus.completed
        or person_onboarding.is_cancelled
    ):
        raise ValueError(
            f"Cannot cancel a person onboarding that is already completed or "
            f"cancelled. Onboarding: {person_onboarding_id}",
        )

    person_onboarding.is_cancelled = True
    if commit:
        current_session.commit()
    else:
        current_session.flush()

    current_logger.info(
        "Cancelled person onboarding {person_onboarding_id}",
    )

create_or_update_retiree_onboarding

create_or_update_retiree_onboarding module-attribute

create_or_update_retiree_onboarding = (
    create_or_update_retiree_onboarding
)

customer_admin

cancel_customer_admin_invitation

cancel_customer_admin_invitation(invitation_id)

Remove a customer admin invitation by its ID.

todo @alexandre.dubreuil - Duplicates cancel_person_onboarding

Source code in components/onboarding/internal/actions/customer_admin.py
def cancel_customer_admin_invitation(invitation_id: UUID) -> None:
    """
    Remove a customer admin invitation by its ID.

    todo @alexandre.dubreuil - Duplicates cancel_person_onboarding
    """

    customer_admin_onboarding = CustomerAdminOnboardingModelBroker.find_by_id(
        invitation_id, with_cancelled=True
    ).one()

    if (
        customer_admin_onboarding.onboarding_status == PersonOnboardingStatus.completed
        or customer_admin_onboarding.is_cancelled
    ):
        raise ValueError(
            f"Cannot cancel a customer admin invitation that is already completed or cancelled. Onboarding: {invitation_id}",
        )

    # Remove the customer admin invitation
    customer_admin_onboarding.is_cancelled = True
    current_session.commit()

    current_logger.info(
        "Cancelled customer admin invitation",
        onboarding_component={"invitation_id": invitation_id},
    )

invite_customer_admin

invite_customer_admin(
    invitation_email,
    entities,
    *,
    first_name=None,
    last_name=None,
    external_employee_id=None,
    lang=None,
    **email_kwargs
)
Source code in components/onboarding/internal/actions/customer_admin.py
def invite_customer_admin(
    invitation_email: str,
    entities: list[CustomerAdminInvitationToEntity],
    *,
    first_name: str | None = None,
    last_name: str | None = None,
    external_employee_id: str | None = None,
    lang: Lang | None = None,
    **email_kwargs: str | bool | None,
) -> CustomerAdminOnboarding:
    # Create an invitation/onboarding for a customer admin, and sends an email

    # delay import to ensure app dependency has been defined for mailer
    from components.onboarding.internal.mailers.customer_admin_invitation_emails import (
        send_customer_admin_invitation_email,
    )

    # Only invite the customer admin for entities that are not already covered by an existing pending onboarding
    entities_to_invite_on = filter_entities_covered_by_existing_pending_onboardings(
        email=invitation_email,
        entities=entities,
    )

    # Ensure the admin is not already admin for the entities
    user_from_invite_email = get_app_dependency().get_user_by_email(invitation_email)
    if (
        user_from_invite_email is not None
        and get_app_dependency().user_can_admin_admined_entities(
            user_from_invite_email.id,
            admined_entities=[
                AdminedEntity(
                    type=(
                        AdminedEntityType.account
                        if entity.entity_type == EntityType.account
                        else (
                            AdminedEntityType.single_company
                            if entity.entity_type == EntityType.company
                            else AdminedEntityType.operational_scope
                        )
                    ),
                    id=entity.entity_id,
                )
                for entity in entities_to_invite_on
            ],
        )
    ):
        raise BaseErrorCode.company_admin_invitation_already_accepted(
            user_id=user_from_invite_email.id
        )

    customer_admin_onboarding = _create_customer_admin_onboarding(
        invitation_email,
        entities_to_invite_on,
        first_name=first_name,
        last_name=last_name,
        external_employee_id=external_employee_id,
        lang=lang,
    )

    send_customer_admin_invitation_email(
        person_onboarding_id=str(customer_admin_onboarding.id),
        **email_kwargs,
    )

    current_logger.info(
        "Invited customer admin",
        onboarding_component={
            "person_onboarding_id": customer_admin_onboarding.id,
        },
    )

    return customer_admin_onboarding

resend_customer_admin_invitation_email

resend_customer_admin_invitation_email(invitation_id)

Resend the invitation email for a customer admin onboarding.

Source code in components/onboarding/internal/actions/customer_admin.py
def resend_customer_admin_invitation_email(invitation_id: UUID) -> None:
    """
    Resend the invitation email for a customer admin onboarding.
    """
    # delay import to ensure app dependency has been defined for mailer
    from components.onboarding.internal.mailers.customer_admin_invitation_emails import (
        send_customer_admin_invitation_email,
    )

    customer_admin_onboarding = CustomerAdminOnboardingModelBroker.find_by_id(
        invitation_id, with_cancelled=True
    ).one()

    if (
        customer_admin_onboarding.onboarding_status == PersonOnboardingStatus.completed
        or customer_admin_onboarding.is_cancelled
    ):
        raise ValueError(
            f"Cannot resend a customer admin invitation email, when it's already completed, or cancelled. Onboarding: {invitation_id}",
        )

    # Resend the customer admin invitation
    send_customer_admin_invitation_email(
        person_onboarding_id=str(customer_admin_onboarding.id),
    )

    current_logger.info(
        "Resend customer admin invitation",
        onboarding_component={"invitation_id": invitation_id},
    )

employee_onboarding

update_employee_onboarding

update_employee_onboarding(
    person_onboarding_id,
    first_name=None,
    last_name=None,
    birthdate=None,
    street=None,
    gender=None,
    city=None,
    province=None,
    postal_code=None,
    country=None,
    billing_account_number=None,
    viewed_terms_at=None,
    commit=True,
)
Source code in components/onboarding/internal/actions/update_employee_onboarding.py
def update_employee_onboarding(
    person_onboarding_id: UUID,
    first_name: Optional[str] = None,
    last_name: Optional[str] = None,
    birthdate: Optional[date] = None,
    street: Optional[str] = None,
    gender: Optional[Gender] = None,
    city: Optional[str] = None,
    province: Optional[str] = None,
    postal_code: Optional[str] = None,
    country: Optional[str] = None,
    billing_account_number: Optional[str] = None,
    viewed_terms_at: Optional[datetime] = None,
    commit: bool = True,
) -> None:
    employee_onboarding_progress = (
        EmployeeOnboardingProgressModelBroker.find_by_person_onboarding_id(
            person_onboarding_id
        )
    ).one()

    person_onboarding = employee_onboarding_progress.person_onboarding

    if person_onboarding.onboarding_status == PersonOnboardingStatus.completed:
        raise BaseErrorCode.invalid_arguments(
            message="Updating a 'completed' onboarding is forbidden",
        )

    if first_name:
        person_onboarding.first_name = first_name.strip()

    if last_name:
        person_onboarding.last_name = last_name.strip()

    if birthdate:
        employee_onboarding_progress.birthdate = birthdate

    if gender:
        employee_onboarding_progress.gender = gender

    if street and city and postal_code and province and country:
        # Raises AddressInvalidException if the address doesn't have the right format, caught in the controller
        address = Address(
            street=street,
            unit=None,
            locality=city,
            administrative_area=province,
            postal_code=postal_code,
            country=country,
            sublocality=None,
            postal_town=None,
        )
        employee_onboarding_progress.street = address.street
        employee_onboarding_progress.postal_code = address.postal_code
        employee_onboarding_progress.province = address.administrative_area
        employee_onboarding_progress.city = address.locality

    if viewed_terms_at:
        employee_onboarding_progress.viewed_terms_at = viewed_terms_at

    if billing_account_number:
        validated_billing_account = (
            get_app_dependency().validates_billing_account_number(
                billing_account_number.strip()
            )
        )
        employee_onboarding_progress.billing_account_number = (
            validated_billing_account.account_number
        )
        employee_onboarding_progress.billing_account_type = (
            validated_billing_account.account_type
        )

    employee_onboarding_entity = get_employee_onboarding_entity(
        employee_onboarding_progress
    )
    if get_app_dependency().is_employee_onboarding_completed(
        employee_onboarding_entity
    ):
        person_onboarding.onboarding_status = PersonOnboardingStatus.completed
        # Call dependency so that they can updated themselves on completed (e.g. user.name, etc.)
        get_app_dependency().on_completed_onboarding(employee_onboarding_entity)
    else:
        person_onboarding.onboarding_status = PersonOnboardingStatus.in_progress

    if commit:
        current_session.commit()
    else:
        current_session.flush()

    current_logger.info(
        "Updated person onboarding",
        person_onboarding_id=person_onboarding.id,
        employee_onboarding_progress_id=employee_onboarding_progress.id,
    )

merge_users

merge_users_in_onboarding_component

merge_users_in_onboarding_component(
    source_user_id, target_user_id, commit=True
)

Support function for the "merge user" procedure: change all onboardings that target user source_user_id so that they target user target_user_id instead.

This will change ALL onboardings for the user, including cancelled ones.

Source code in components/onboarding/internal/actions/merge_users.py
def merge_users_in_onboarding_component(
    source_user_id: str,
    target_user_id: str,
    commit: bool = True,
) -> None:
    """
    Support function for the "merge user" procedure: change all onboardings that target
    user `source_user_id` so that they target user `target_user_id` instead.

    This will change ALL onboardings for the user, including cancelled ones.
    """
    onboardings = PersonOnboardingModelBroker.find_by_user_id_order_latest(
        source_user_id, with_cancelled=True
    )

    for onboarding in onboardings:
        current_logger.info(
            f"Moving employee onboarding {onboarding.id} to target user"
        )
        onboarding.user_id = target_user_id

    current_session.flush()

    if commit:
        current_session.commit()

onboarding_invitation_emails

resend_onboarding_invitation_email

resend_onboarding_invitation_email(person_onboarding_id)

Resend an onboarding invitation email if the onboarding is pending. Do nothing otherwise.

Source code in components/onboarding/internal/actions/resend_onboarding_invitation.py
def resend_onboarding_invitation_email(person_onboarding_id: UUID) -> None:
    """
    Resend an onboarding invitation email if the onboarding is pending.
    Do nothing otherwise.
    """

    from components.onboarding.internal.mailers.employee_onboarding_invitation_email import (
        send_employee_onboarding_email,
    )
    from components.onboarding.internal.mailers.retiree_onboarding_invitation_email import (
        send_retiree_onboarding_invitation_email,
    )

    person_onboarding = get_person_onboarding(
        person_onboarding_id=person_onboarding_id, with_cancelled=True
    )

    if (
        person_onboarding.is_cancelled
        or person_onboarding.onboarding_status == PersonOnboardingStatus.completed
    ):
        current_logger.info(
            f"Cannot resend an invitation email for {person_onboarding.onboarding_type}, "
            f"when it's already completed, cancelled, or not an employee onboarding. "
            f"Onboarding: {person_onboarding_id}"
        )
        return

    if person_onboarding.onboarding_type == OnboardingType.employee_onboarding:
        company_name = (
            get_app_dependency()
            .get_company_from_employed_user(
                user_id=mandatory(person_onboarding.user_id),
                on_date=mandatory(person_onboarding.start_date),
            )
            .name
        )
        send_employee_onboarding_email(
            company_name=company_name,
            employee_onboarding_id=str(person_onboarding_id),
        )
    elif person_onboarding.onboarding_type == OnboardingType.retiree_onboarding:
        send_retiree_onboarding_invitation_email(
            retiree_onboarding_id=str(person_onboarding_id)
        )
    else:
        raise NotImplementedError()

    current_logger.info(
        "Resend employee onboarding invitation",
        onboarding_component={"person_onboarding_id": person_onboarding_id},
    )

update_retiree_onboarding

update_retiree_onboarding module-attribute

update_retiree_onboarding = update_retiree_onboarding

update_retiree_onboarding_status module-attribute

update_retiree_onboarding_status = (
    update_retiree_onboarding_status
)

components.onboarding.public.api

complete_retiree_onboarding

complete_retiree_onboarding(
    person_onboarding_id, commit=False
)

Mark the retiree onboarding object as completed

Source code in components/onboarding/public/api.py
def complete_retiree_onboarding(
    person_onboarding_id: UUID, commit: bool = False
) -> None:
    """
    Mark the retiree onboarding object as completed
    """
    from components.onboarding.internal.actions.update_retiree_onboarding import (
        complete_retiree_onboarding as complete_retiree_onboarding_,
    )

    return complete_retiree_onboarding_(
        person_onboarding_id=person_onboarding_id, commit=commit
    )

get_first_employee_onboarding_by_email

get_first_employee_onboarding_by_email(invitation_email)
Source code in components/onboarding/public/api.py
def get_first_employee_onboarding_by_email(  # noqa: D103
    invitation_email: str,
) -> EmployeeOnboardingEntity | None:
    from components.onboarding.internal.queries.employee_onboarding import (
        get_employee_onboarding_by_email as get_employee_onboarding_by_email_,
    )

    return get_employee_onboarding_by_email_(invitation_email)

get_pending_onboardings

get_pending_onboardings(user_id)
Source code in components/onboarding/public/api.py
def get_pending_onboardings(user_id: str) -> PendingOnboardingsEntity:  # noqa: D103
    from components.onboarding.internal.queries.pending_onboardings import (
        get_pending_onboardings as get_pending_onboardings_,
    )

    return get_pending_onboardings_(user_id=user_id)

invite_employee

invite_employee(
    invitation_email,
    coverage_start_date,
    company_name,
    user_id=None,
    defer_side_effects=True,
    commit=True,
)
Source code in components/onboarding/public/api.py
def invite_employee(  # noqa: D103
    invitation_email: str,
    coverage_start_date: date,
    company_name: str,
    user_id: str | None = None,
    defer_side_effects: bool = True,
    commit: bool = True,
) -> EmployeeOnboardingEntity:
    from components.onboarding.internal.actions.invite_employee import (
        invite_employee as invite_employee_,
    )

    return invite_employee_(
        invitation_email=invitation_email,
        coverage_start_date=coverage_start_date,
        company_name=company_name,
        user_id=user_id,
        defer_side_effects=defer_side_effects,
        commit=commit,
    )

invite_partner

invite_partner(
    inviter_user_id,
    invitation_email,
    partner_profile_id,
    partner_first_name,
    partner_last_name,
    partner_user_id,
    lang=None,
    send_invite_email=True,
    commit=True,
)

Invites a user to onboard as a company admin.

Source code in components/onboarding/public/api.py
def invite_partner(
    inviter_user_id: str,
    invitation_email: str,
    partner_profile_id: str,
    partner_first_name: str,
    partner_last_name: str,
    partner_user_id: UUID,
    lang: Lang | None = None,
    send_invite_email: bool = True,
    commit: bool = True,
) -> UUID:
    """Invites a user to onboard as a company admin."""
    from components.onboarding.internal.actions.invite_partner import (
        invite_partner as _invite_partner,
    )

    return _invite_partner(
        inviter_user_id=inviter_user_id,
        invitation_email=invitation_email,
        partner_profile_id=partner_profile_id,
        partner_first_name=partner_first_name,
        partner_last_name=partner_last_name,
        partner_user_id=partner_user_id,
        lang=lang,
        send_invite_email=send_invite_email,
        commit=commit,
    )

invite_retiree

invite_retiree(
    inviter_user_id,
    invitation_email,
    coverage_start_date,
    contract_id,
    retiree_onboarding_type,
    retiree_user_id=None,
    first_name=None,
    last_name=None,
    identification_number=None,
    retirement_date=None,
    is_offline=False,
    alsace_moselle_eligibility_status=None,
    read_only_fields=None,
    lang=None,
    send_invite_email=True,
    on_duplicated_onboarding="Fail",
    defer_side_effects=False,
    commit=True,
)

Invites a user to onboard as retiree.

Source code in components/onboarding/public/api.py
def invite_retiree(
    inviter_user_id: str | None,
    invitation_email: str | None,
    coverage_start_date: date | None,
    contract_id: str,
    retiree_onboarding_type: RetireeOnboardingType,
    retiree_user_id: str | None = None,
    first_name: str | None = None,
    last_name: str | None = None,
    identification_number: str | None = None,
    retirement_date: date | None = None,
    is_offline: bool = False,
    alsace_moselle_eligibility_status: str | None = None,
    read_only_fields: list[str] | None = None,
    lang: Lang | None = None,
    send_invite_email: bool = True,
    on_duplicated_onboarding: Literal["Update", "Fail"] = "Fail",
    defer_side_effects: bool = False,
    commit: bool = True,
) -> UUID:
    """Invites a user to onboard as retiree."""
    from components.onboarding.internal.actions.invite_retiree import (
        invite_retiree as _invite_retiree,
    )

    return _invite_retiree(
        inviter_user_id=inviter_user_id,
        invitation_email=invitation_email,
        coverage_start_date=coverage_start_date,
        contract_id=contract_id,
        retiree_onboarding_type=retiree_onboarding_type,
        retiree_user_id=retiree_user_id,
        first_name=first_name,
        last_name=last_name,
        identification_number=identification_number,
        retirement_date=retirement_date,
        is_offline=is_offline,
        alsace_moselle_eligibility_status=alsace_moselle_eligibility_status,
        read_only_fields=read_only_fields,
        lang=lang,
        on_duplicated_onboarding=on_duplicated_onboarding,
        send_invite_email=send_invite_email,
        defer_side_effects=defer_side_effects,
        commit=commit,
    )

components.onboarding.public.commands

onboarding

generate_frontend_typescript_types_file

generate_frontend_typescript_types_file(dry_run)

Use flask generate_frontend_typescript_types_file --execute to generate the frontend typescript types.

  • File: "frontend/features/onboarding/queries/types.ts"
  • Types: all enums and entities
Source code in components/onboarding/public/commands/onboarding.py
@onboarding_commands.command()
@command_with_dry_run
def generate_frontend_typescript_types_file(dry_run: bool) -> None:
    """
    Use `flask generate_frontend_typescript_types_file --execute` to generate the frontend typescript types.

    - File: "frontend/features/onboarding/queries/types.ts"
    - Types: all enums and entities
    """
    from components.onboarding.public.entities.employee_onboarding import (
        EmployeeOnboardingEntity,
    )
    from components.onboarding.public.entities.person_onboarding import (
        PersonOnboardingEntity,
    )
    from components.onboarding.public.entities.retiree_onboarding import (
        RetireeOnboardingEntity,
    )
    from components.onboarding.public.entities.terms_document import (
        TermsDocumentEntity,
    )
    from components.onboarding.public.enums.account_type import AccountType
    from components.onboarding.public.enums.onboarding_type import OnboardingType
    from components.onboarding.public.enums.person_onboarding_status import (
        PersonOnboardingStatus,
    )
    from components.onboarding.public.enums.retiree_onboarding_type import (
        RetireeOnboardingType,
    )
    from components.onboarding.public.enums.terms_document_type import TermsDocumentType
    from shared.codegen.typescript import (
        generate_typescript_enums_and_interfaces as generate_typescript_enums_and_interfaces_,
    )
    from shared.models.enums.user_gender import UserGender

    generate_typescript_enums_and_interfaces_(
        target_filename_path_from_git_root="frontend/features/onboarding/queries/types.ts",
        types_to_generate=[
            # Enums
            OnboardingType,
            PersonOnboardingStatus,
            RetireeOnboardingType,
            AccountType,
            TermsDocumentType,
            UserGender,
            # Models
            PersonOnboardingEntity,
            EmployeeOnboardingEntity,
            RetireeOnboardingEntity,
            TermsDocumentEntity,
        ],
        dry_run=dry_run,
    )

invite_retiree

invite_retiree(
    invitation_email,
    start_date,
    contract_ref,
    retiree_user_id,
    retiree_onboarding_type,
    dry_run,
)

Use flask onboarding invite_retiree --invitation_email --start_date --contract_ref --retiree_user_id --retiree_onboarding_type --execute to invite a retiree.

Source code in components/onboarding/public/commands/onboarding.py
@onboarding_commands.command()
@click.option(
    "--invitation_email",
    type=str,
    required=True,
)
@click.option(
    "--start_date",
    type=click.DateTime(formats=["%Y-%m-%d"]),
    required=True,
)
@click.option(
    "--contract_ref",
    type=str,
    required=True,
)
@click.option(
    "--retiree_user_id",
    type=str,
    required=False,
)
@click.option(
    "--retiree_onboarding_type",
    type=str,
    required=True,
)
@command_with_dry_run
def invite_retiree(
    invitation_email: str,
    start_date: date,
    contract_ref: str,
    retiree_user_id: str | None,
    retiree_onboarding_type: str,
    dry_run: bool,
) -> None:
    """
    Use
    flask onboarding invite_retiree --invitation_email <email> --start_date <date> --contract_ref <contract_ref> --retiree_user_id <retiree_user_id> --retiree_onboarding_type <retiree_onboarding_type> --execute
    to invite a retiree.
    """
    from components.onboarding.public.api import invite_retiree as _invite_retiree
    from components.onboarding.public.enums.retiree_onboarding_type import (
        RetireeOnboardingType,
    )

    _invite_retiree(
        inviter_user_id=None,
        invitation_email=invitation_email,
        coverage_start_date=start_date,
        contract_id=contract_ref,
        retiree_user_id=retiree_user_id,
        retiree_onboarding_type=RetireeOnboardingType(retiree_onboarding_type),
        commit=not dry_run,
    )

onboarding_commands module-attribute

onboarding_commands = AppGroup(
    name="onboarding",
    help="Main command group for the onboarding component",
)

components.onboarding.public.constants

ONBOARDING_SCHEMA_NAME module-attribute

ONBOARDING_SCHEMA_NAME = 'onboarding'

components.onboarding.public.dependencies

COMPONENT_NAME module-attribute

COMPONENT_NAME = 'onboarding'

OnboardingDependency

Bases: ABC

can_invite_partner abstractmethod

can_invite_partner(user_id, partner_profile_id)

Implement can_invite_partner

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def can_invite_partner(
    self,
    user_id: str,
    partner_profile_id: str,
) -> bool:
    """Implement can_invite_partner"""

can_user_invite_company_admin abstractmethod

can_user_invite_company_admin(user_id, company_id)

Implement can_user_invite_company_admin

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def can_user_invite_company_admin(
    self,
    user_id: str,
    company_id: UUID,
) -> bool:
    """Implement can_user_invite_company_admin"""

can_user_invite_on_retiree_contract abstractmethod

can_user_invite_on_retiree_contract(
    user_id, contract_id, retiree_onboarding_type
)

Implement can_user_invite_on_contract

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def can_user_invite_on_retiree_contract(
    self,
    user_id: str,
    contract_id: str,
    retiree_onboarding_type: RetireeOnboardingType,
) -> bool:
    """Implement can_user_invite_on_contract"""

can_user_onboard_as_retiree abstractmethod

can_user_onboard_as_retiree(user_id, start_date)

Implement can_user_onboard_as_retiree

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def can_user_onboard_as_retiree(
    self,
    user_id: str,
    start_date: date | None,
) -> bool:
    """Implement can_user_onboard_as_retiree"""

create_or_update_user abstractmethod

create_or_update_user(user_id, email, prehashed_password)

Implement create_or_update_user

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def create_or_update_user(
    self,
    user_id: str | None,
    email: str,
    prehashed_password: str,
) -> BaseUser:
    """Implement create_or_update_user"""

create_retiree_proposal

create_retiree_proposal(onboarding)

Creates a proposal for a retiree onboarding. The flow will then be able to redirect the user to the approval request page of the proposal, so they can sign their contract.

Source code in components/onboarding/public/dependencies.py
def create_retiree_proposal(self, onboarding: RetireeOnboardingEntity) -> UUID:
    """Creates a proposal for a retiree onboarding. The flow will then be able to redirect the user to the
    approval request page of the proposal, so they can sign their contract.
    """
    raise NotImplementedError("create_retiree_proposal not implemented")

custom_mailer_params

custom_mailer_params(
    onboarding_id, onboarding_type, default_mailer_params
)

Allows dependencies to customize the email.

This allows dependencies to add attachments or even change the template itself.

Source code in components/onboarding/public/dependencies.py
def custom_mailer_params(
    self,
    onboarding_id: UUID,  # noqa: ARG002
    onboarding_type: OnboardingType,  # noqa: ARG002
    default_mailer_params: LangTemplateMailerParamsArgs,
) -> LangTemplateMailerParamsArgs:
    """
    Allows dependencies to customize the email.

    This allows dependencies to add attachments or even change the template itself.
    """
    return default_mailer_params

get_account abstractmethod

get_account(company_id)

Implement get_account

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def get_account(self, company_id: str) -> Account:
    """Implement get_account"""

get_account_org_tree_query_api abstractmethod

get_account_org_tree_query_api()

Implement get_account_org_tree_query_api

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def get_account_org_tree_query_api(self) -> "AccountOrgTreeQueryApi":
    """Implement get_account_org_tree_query_api"""

get_async_mailer abstractmethod

get_async_mailer(
    category,
    priority,
    email_queue_name,
    delivery=DEFAULT_DELIVERY,
    job_timeout=None,
    enqueue=True,
    redacted_args=None,
)

Implements getting the local async_mailer

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def get_async_mailer(
    self,
    category: EmailCategory,
    priority: EmailPriority,
    email_queue_name: Optional[str],
    delivery: Union[CustomerIODelivery, SendgridDelivery] = DEFAULT_DELIVERY,
    job_timeout: Optional[int] = None,
    enqueue: bool = True,
    redacted_args: set[str] | None = None,
) -> Callable[[_MailerCallable], _MailerCallable]:
    """Implements getting the local async_mailer"""

get_company abstractmethod

get_company(company_id)

Implement get_company

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def get_company(self, company_id: str) -> Company:
    """Implement get_company"""

get_company_from_contract abstractmethod

get_company_from_contract(contract_id)

Implement get_company_from_contract

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def get_company_from_contract(self, contract_id: str) -> Company:
    """Implement get_company_from_contract"""

get_company_from_employed_user

get_company_from_employed_user(user_id, on_date)

Implement company_from_employed_user

Source code in components/onboarding/public/dependencies.py
def get_company_from_employed_user(self, user_id: str, on_date: date) -> Company:
    """Implement company_from_employed_user"""
    raise NotImplementedError()
get_retiree_onboarding_link(token, lang)

Implement get_retiree_onboarding_link

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def get_retiree_onboarding_link(self, token: str, lang: str | None) -> str:
    """Implement get_retiree_onboarding_link"""

get_terms_documents abstractmethod

get_terms_documents(user_id)

Implement get_terms_documents

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def get_terms_documents(
    self,
    user_id: UUID,
) -> list[TermsDocumentEntity]:
    """Implement get_terms_documents"""

get_user_by_email abstractmethod

get_user_by_email(email)

Implement get_user_by_email

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def get_user_by_email(self, email: str) -> BaseUser | None:
    """Implement get_user_by_email"""

get_user_info

get_user_info(user_id)

Override get_user_info to provide more information

Source code in components/onboarding/public/dependencies.py
def get_user_info(self, user_id: str | None) -> UserInfo:  # noqa: ARG002
    """Override get_user_info to provide more information"""
    return UserInfo()

is_employee_onboarding_completed

is_employee_onboarding_completed(onboarding)

Implement is_employee_onboarding_completed (override for specific completed rules)

Source code in components/onboarding/public/dependencies.py
def is_employee_onboarding_completed(
    self, onboarding: EmployeeOnboardingEntity
) -> bool:
    """Implement is_employee_onboarding_completed (override for specific completed rules)"""
    from components.onboarding.internal.actions.update_employee_onboarding import (
        is_employee_onboarding_completed as is_employee_onboarding_completed_,
    )

    return is_employee_onboarding_completed_(onboarding)

is_partner_onboarding_completed

is_partner_onboarding_completed(onboarding)

Implement is_employee_onboarding_completed (override for specific completed rules)

Source code in components/onboarding/public/dependencies.py
def is_partner_onboarding_completed(
    self, onboarding: PartnerOnboardingEntity
) -> bool:
    """Implement is_employee_onboarding_completed (override for specific completed rules)"""
    from components.onboarding.internal.actions.update_partner_onboarding import (
        is_partner_onboarding_completed as is_partner_onboarding_completed_,
    )

    return is_partner_onboarding_completed_(onboarding)

is_retiree_onboarding_completed

is_retiree_onboarding_completed(onboarding)

Implement is_retiree_onboarding_completed (override for specific completed rules)

Source code in components/onboarding/public/dependencies.py
def is_retiree_onboarding_completed(
    self, onboarding: RetireeOnboardingEntity
) -> bool:
    """Implement is_retiree_onboarding_completed (override for specific completed rules)"""
    from components.onboarding.internal.queries.retiree_onboarding import (
        is_retiree_onboarding_completed as is_retiree_onboarding_completed_,
    )

    return is_retiree_onboarding_completed_(onboarding)

on_completed_onboarding

on_completed_onboarding(onboarding)

Implement on_completed_employee_onboarding (optional)

Source code in components/onboarding/public/dependencies.py
def on_completed_onboarding(
    self,
    onboarding: BaseOnboardingProgress,  # noqa: ARG002
) -> None:
    """Implement on_completed_employee_onboarding (optional)"""
    return None

on_onboarding_completed abstractmethod

on_onboarding_completed(onboarding_id, onboarding_type)

Called when an onboarding is completed. This allows dependencies to perform additional actions when an onboarding is completed.

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def on_onboarding_completed(
    self,
    onboarding_id: UUID,
    onboarding_type: OnboardingType,
) -> None:
    """
    Called when an onboarding is completed.
    This allows dependencies to perform additional actions when an onboarding is completed.
    """

update_user_billing_account_number abstractmethod

update_user_billing_account_number(user_id, account_number)

Update user billing account number

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def update_user_billing_account_number(
    self, user_id: str, account_number: str
) -> None:
    """Update user billing account number"""

update_user_info abstractmethod

update_user_info(
    user_id,
    first_name,
    last_name,
    birth_date,
    gender,
    identification_number,
    street,
    postal_code,
    city,
    province,
    country,
    lang,
)

Update basic member info

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def update_user_info(
    self,
    user_id: str,
    first_name: str,
    last_name: str,
    birth_date: date,
    gender: UserGender,
    identification_number: str | None,
    street: str,
    postal_code: str,
    city: str,
    province: str | None,
    country: str,
    lang: str,
) -> None:
    """Update basic member info"""

update_user_settlement_account_number abstractmethod

update_user_settlement_account_number(
    user_id, account_number
)

Update user settlement account number

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def update_user_settlement_account_number(
    self, user_id: str, account_number: str
) -> None:
    """Update user settlement account number"""

user_can_admin_admined_entities abstractmethod

user_can_admin_admined_entities(user_id, admined_entities)

Implement user_can_admin_admined_entities

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def user_can_admin_admined_entities(
    self,
    user_id: int | UUID,
    admined_entities: list[AdminedEntity],
) -> bool:
    """Implement user_can_admin_admined_entities"""

validates_billing_account_number abstractmethod

validates_billing_account_number(billing_account_number)

Implement validates_billing_account_number If failing to validate provided account number, should raise BankAccountValidationError

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def validates_billing_account_number(
    self,
    billing_account_number: str,
) -> BankAccount:
    """
    Implement validates_billing_account_number
    If failing to validate provided account number, should raise BankAccountValidationError
    """

validates_legacy_contract_delay

validates_legacy_contract_delay(
    coverage_start_date,
    legacy_contract_start_date,
    on_date=None,
)

Implement validates_legacy_contract_delay If failing to validate provided dates, should raise LegacyContractDelayValidationError

Source code in components/onboarding/public/dependencies.py
def validates_legacy_contract_delay(
    self,
    coverage_start_date: date,  # noqa: ARG002
    legacy_contract_start_date: date,  # noqa: ARG002
    on_date: date | None = None,  # Helps with testing  # noqa: ARG002
) -> None:
    """
    Implement validates_legacy_contract_delay
    If failing to validate provided dates, should raise LegacyContractDelayValidationError
    """
    # NOTE: A default implementation is provided so it's not mandatory to override in subclasses.
    from components.onboarding.public.exceptions import (
        LegacyContractDelayValidationError,
    )

    raise LegacyContractDelayValidationError(
        f"Legacy contract validation isn't implemented for {self.__class__.__name__}"
    )

validates_settlement_account_number abstractmethod

validates_settlement_account_number(
    settlement_account_number,
)

Implement validates_settlement_account_number If failing to validate provided account number, should raise BankAccountValidationError

Source code in components/onboarding/public/dependencies.py
@abstractmethod
def validates_settlement_account_number(
    self,
    settlement_account_number: str,
) -> BankAccount:
    """
    Implement validates_settlement_account_number
    If failing to validate provided account number, should raise BankAccountValidationError
    """

validates_start_date

validates_start_date(
    coverage_start_date, start_date_is_read_only
)

Depending on the flow, the start date might be customized by members. In that case, we want to validates that date. We might want to force it to be in the future, and so on.

By default it doesn't validates anything, but one could check!

Source code in components/onboarding/public/dependencies.py
def validates_start_date(
    self,
    coverage_start_date: date,  # noqa: ARG002
    start_date_is_read_only: bool,  # noqa: ARG002
) -> None:
    """
    Depending on the flow, the start date might be customized by members. In that case,
    we want to validates that date. We might want to force it to be in the future, and so on.

    By default it doesn't validates anything, but one could check!
    """
    # NOTE: This might get merged with/absorb validates_legacy_contract_delay.
    return

get_app_dependency

get_app_dependency()
Source code in components/onboarding/public/dependencies.py
def get_app_dependency() -> OnboardingDependency:  # noqa: D103
    from flask import current_app

    return cast("CustomFlask", current_app).get_component_dependency(COMPONENT_NAME)  # type: ignore[no-any-return]

set_app_dependency

set_app_dependency(dependency)
Source code in components/onboarding/public/dependencies.py
def set_app_dependency(dependency: OnboardingDependency) -> None:  # noqa: D103
    from flask import current_app

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

components.onboarding.public.entities

bank_account

BankAccount dataclass

BankAccount(*, account_number, account_type)

Bases: DataClassJsonMixin

account_number instance-attribute
account_number
account_type instance-attribute
account_type

company

Account dataclass

Account(*, id, name)

Bases: DataClassJsonMixin

Account entity.

id instance-attribute
id
name instance-attribute
name

Company dataclass

Company(*, id, name)

Bases: DataClassJsonMixin

Company entity.

id instance-attribute
id
name instance-attribute
name

customer_admin_onboarding

CustomerAdminInvitationToEntity dataclass

CustomerAdminInvitationToEntity(
    *,
    entity_id,
    entity_type,
    invite_as_contract_manager=None,
    invite_as_people_manager=None,
    invite_as_payroll_manager=None,
    invite_as_invoice_manager=None,
    invite_as_wellbeing_referent=None
)

Bases: DataClassJsonMixin

Dataclass representing a customer admin invitation to an entity.

entity_id instance-attribute
entity_id
entity_type instance-attribute
entity_type
invite_as_contract_manager class-attribute instance-attribute
invite_as_contract_manager = None
invite_as_invoice_manager class-attribute instance-attribute
invite_as_invoice_manager = None
invite_as_payroll_manager class-attribute instance-attribute
invite_as_payroll_manager = None
invite_as_people_manager class-attribute instance-attribute
invite_as_people_manager = None
invite_as_wellbeing_referent class-attribute instance-attribute
invite_as_wellbeing_referent = None

CustomerAdminOnboarding dataclass

CustomerAdminOnboarding(
    *,
    id,
    onboarding_type,
    onboarding_status,
    invitation_email,
    authentication_token,
    user_id,
    first_name,
    last_name,
    lang,
    is_offline,
    start_date,
    is_cancelled,
    external_employee_id,
    entities
)

Bases: PersonOnboardingEntity, DataClassJsonMixin

Dataclass representing a customer admin onboarding.

entities instance-attribute
entities
external_employee_id instance-attribute
external_employee_id

employee_onboarding

EmployeeOnboardingEntity dataclass

EmployeeOnboardingEntity(
    *,
    person_onboarding_id,
    onboarding_type,
    onboarding_status,
    invitation_email,
    authentication_token,
    user_id,
    target_ref,
    start_date=optional_isodate_field(),
    lang,
    is_offline,
    employee_onboarding_progress_id,
    first_name=None,
    last_name=None,
    birthdate=optional_isodate_field(),
    gender=None,
    street=None,
    city=None,
    province=None,
    postal_code=None,
    billing_account_number=None,
    billing_account_type=None,
    viewed_terms_at=optional_isodatetime_field(),
    created_at=isodatetime_field()
)

Bases: BaseOnboardingProgress, DataClassJsonMixin

billing_account_number class-attribute instance-attribute
billing_account_number = None
billing_account_type class-attribute instance-attribute
billing_account_type = None
birthdate class-attribute instance-attribute
birthdate = optional_isodate_field()
city class-attribute instance-attribute
city = None
created_at class-attribute instance-attribute
created_at = isodatetime_field()
employee_onboarding_progress_id instance-attribute
employee_onboarding_progress_id
first_name class-attribute instance-attribute
first_name = None
gender class-attribute instance-attribute
gender = None
last_name class-attribute instance-attribute
last_name = None
postal_code class-attribute instance-attribute
postal_code = None
province class-attribute instance-attribute
province = None
street class-attribute instance-attribute
street = None
viewed_terms_at class-attribute instance-attribute
viewed_terms_at = optional_isodatetime_field()

onboarding_progress

BaseOnboardingProgress dataclass

BaseOnboardingProgress(
    *,
    person_onboarding_id,
    onboarding_type,
    onboarding_status,
    invitation_email,
    authentication_token,
    user_id,
    target_ref,
    start_date=optional_isodate_field(),
    lang,
    is_offline
)

Bases: DataClassJsonMixin

authentication_token instance-attribute
authentication_token
invitation_email instance-attribute
invitation_email
is_offline instance-attribute
is_offline
lang instance-attribute
lang
onboarding_status instance-attribute
onboarding_status
onboarding_type instance-attribute
onboarding_type
person_onboarding_id instance-attribute
person_onboarding_id
start_date class-attribute instance-attribute
start_date = optional_isodate_field()
target_ref instance-attribute
target_ref
user_id instance-attribute
user_id

partner_onboarding

PartnerOnboardingEntity dataclass

PartnerOnboardingEntity(
    *,
    person_onboarding_id,
    onboarding_type,
    onboarding_status,
    invitation_email,
    authentication_token,
    user_id,
    target_ref,
    start_date=optional_isodate_field(),
    lang,
    is_offline,
    first_name=None,
    last_name=None
)

Bases: BaseOnboardingProgress, DataClassJsonMixin

first_name class-attribute instance-attribute
first_name = None
last_name class-attribute instance-attribute
last_name = None

pending_onboardings

PendingEmployeeOnboardingEntity dataclass

PendingEmployeeOnboardingEntity(
    employment_id, onboarding_status
)

Bases: DataClassJsonMixin

Represents a pending employee onboarding status

  • duplicates backend/components/fr/internal/business_logic/user/queries/user.py - keep return type aligned
employment_id instance-attribute
employment_id
onboarding_status instance-attribute
onboarding_status

PendingOnboardingsEntity dataclass

PendingOnboardingsEntity(
    pending_employee_onboarding, pending_retiree_onboarding
)

Bases: DataClassJsonMixin

Represents pending onboardings

  • duplicates backend/components/fr/internal/business_logic/user/queries/user.py - keep return type aligned
pending_employee_onboarding instance-attribute
pending_employee_onboarding
pending_retiree_onboarding instance-attribute
pending_retiree_onboarding

PendingRetireeOnboardingEntity dataclass

PendingRetireeOnboardingEntity(
    id,
    retirement_date,
    is_offline,
    alsace_moselle_eligibility_status,
    onboarding_token,
    onboarding_url,
    onboarding_status,
)

Bases: DataClassJsonMixin

Represents a pending retiree onboarding status

  • new compared to backend/components/fr/internal/business_logic/user/queries/user.py - keep return type aligned
alsace_moselle_eligibility_status instance-attribute
alsace_moselle_eligibility_status
id instance-attribute
id
is_offline instance-attribute
is_offline
onboarding_status instance-attribute
onboarding_status
onboarding_token instance-attribute
onboarding_token
onboarding_url instance-attribute
onboarding_url
retirement_date instance-attribute
retirement_date

person_onboarding

PersonOnboardingEntity dataclass

PersonOnboardingEntity(
    *,
    id,
    onboarding_type,
    onboarding_status,
    invitation_email,
    authentication_token,
    user_id,
    first_name,
    last_name,
    lang,
    is_offline,
    start_date,
    is_cancelled
)

Bases: DataClassJsonMixin

Dataclass representing a person onboarding.

authentication_token instance-attribute
authentication_token
first_name instance-attribute
first_name
id instance-attribute
id
invitation_email instance-attribute
invitation_email
is_cancelled instance-attribute
is_cancelled
is_offline instance-attribute
is_offline
lang instance-attribute
lang
last_name instance-attribute
last_name
onboarding_status instance-attribute
onboarding_status
onboarding_type instance-attribute
onboarding_type
start_date instance-attribute
start_date
user_id instance-attribute
user_id

retiree_onboarding

OnboardingDocumentEntity dataclass

OnboardingDocumentEntity(*, name, presigned_s3_uri, uri)

Bases: DataClassJsonMixin

Represents a document that is part of the onboarding process.

It's usually a proof of eligibility that's gonna be processed afterward.

name instance-attribute
name
presigned_s3_uri instance-attribute
presigned_s3_uri
uri instance-attribute
uri

OnboardingDocumentToAdd dataclass

OnboardingDocumentToAdd(*, filename, base64_content)

Bases: DataClassJsonMixin, DataClassJsonAlanMixin

Represent a file from the frontend, one that can be included in JSON bodies.

base64_content instance-attribute
base64_content
filename instance-attribute
filename
get_file
get_file()

Return a file-like object from the base64 content.

Source code in components/onboarding/public/entities/retiree_onboarding.py
def get_file(self) -> IO[bytes]:
    """Return a file-like object from the base64 content."""
    import base64

    file = NamedTemporaryFile()
    file.write(base64.b64decode(self.base64_content))
    file.seek(0)

    return file

OnboardingDocumentUpdates module-attribute

OnboardingDocumentUpdates = list[
    tuple[str, Literal["set", "delete"], IO[bytes] | None]
]

RetireeOnboardingEntity dataclass

RetireeOnboardingEntity(
    *,
    person_onboarding_id,
    onboarding_type,
    onboarding_status,
    invitation_email,
    authentication_token,
    user_id,
    target_ref,
    start_date=optional_isodate_field(),
    lang,
    is_offline,
    company_id,
    company_name,
    retiree_onboarding_progress_id,
    retiree_onboarding_type,
    first_name=None,
    last_name=None,
    birthdate=optional_isodate_field(),
    gender=None,
    identification_number=None,
    street=None,
    postal_code=None,
    city=None,
    province=None,
    country=None,
    retirement_date=optional_isodate_field(),
    legacy_insurer_ref=None,
    legacy_contract_ref=None,
    legacy_contract_start_date=optional_isodate_field(),
    billing_account_number=None,
    billing_account_type=None,
    settlement_account_number=None,
    settlement_account_type=None,
    approval_link=None,
    alsace_moselle_eligibility_status=None,
    onboarding_documents=list(),
    read_only_fields=list(),
    proposal_ref=None
)

Bases: BaseOnboardingProgress, DataClassJsonMixin

alsace_moselle_eligibility_status class-attribute instance-attribute
alsace_moselle_eligibility_status = None
approval_link = None
billing_account_number class-attribute instance-attribute
billing_account_number = None
billing_account_type class-attribute instance-attribute
billing_account_type = None
birthdate class-attribute instance-attribute
birthdate = optional_isodate_field()
city class-attribute instance-attribute
city = None
company_id instance-attribute
company_id
company_name instance-attribute
company_name
country class-attribute instance-attribute
country = None
first_name class-attribute instance-attribute
first_name = None
gender class-attribute instance-attribute
gender = None
identification_number class-attribute instance-attribute
identification_number = None
last_name class-attribute instance-attribute
last_name = None
legacy_contract_ref class-attribute instance-attribute
legacy_contract_ref = None
legacy_contract_start_date class-attribute instance-attribute
legacy_contract_start_date = optional_isodate_field()
legacy_insurer_ref class-attribute instance-attribute
legacy_insurer_ref = None
onboarding_documents class-attribute instance-attribute
onboarding_documents = field(default_factory=list)
postal_code class-attribute instance-attribute
postal_code = None
proposal_ref class-attribute instance-attribute
proposal_ref = None
province class-attribute instance-attribute
province = None
read_only_fields class-attribute instance-attribute
read_only_fields = field(default_factory=list)
retiree_onboarding_progress_id instance-attribute
retiree_onboarding_progress_id
retiree_onboarding_type instance-attribute
retiree_onboarding_type
retirement_date class-attribute instance-attribute
retirement_date = optional_isodate_field()
settlement_account_number class-attribute instance-attribute
settlement_account_number = None
settlement_account_type class-attribute instance-attribute
settlement_account_type = None
street class-attribute instance-attribute
street = None

RetireeOnboardingStatusUpdate

Bases: AlanBaseEnum

  • waiting_for_approval is used by the frontend when we consider we are ready to create a proposal.
  • cancelled is used to cancel the onboarding itself.
cancelled class-attribute instance-attribute
cancelled = 'cancelled'
waiting_for_approval class-attribute instance-attribute
waiting_for_approval = 'waiting_for_approval'

UpdateRetireeOnboardingFields dataclass

UpdateRetireeOnboardingFields(
    *,
    first_name=NOT_SET,
    last_name=NOT_SET,
    gender=NOT_SET,
    birthdate=NOT_SET,
    identification_number=NOT_SET,
    street=NOT_SET,
    postal_code=NOT_SET,
    city=NOT_SET,
    province=NOT_SET,
    country=NOT_SET,
    start_date=NOT_SET,
    retirement_date=NOT_SET,
    legacy_insurer_ref=NOT_SET,
    legacy_contract_ref=NOT_SET,
    legacy_contract_start_date=NOT_SET,
    billing_account_number=NOT_SET,
    settlement_account_number=NOT_SET,
    onboarding_documents=NOT_SET,
    is_offline=NOT_SET
)

See update_retiree_onboarding, which takes such updates.

Having those updates in the same dataclass help sharing them across different functions.

billing_account_number class-attribute instance-attribute
billing_account_number = NOT_SET
birthdate class-attribute instance-attribute
birthdate = NOT_SET
city class-attribute instance-attribute
city = NOT_SET
country class-attribute instance-attribute
country = NOT_SET
first_name class-attribute instance-attribute
first_name = NOT_SET
gender class-attribute instance-attribute
gender = NOT_SET
identification_number class-attribute instance-attribute
identification_number = NOT_SET
is_offline class-attribute instance-attribute
is_offline = NOT_SET
last_name class-attribute instance-attribute
last_name = NOT_SET
legacy_contract_ref class-attribute instance-attribute
legacy_contract_ref = NOT_SET
legacy_contract_start_date class-attribute instance-attribute
legacy_contract_start_date = NOT_SET
legacy_insurer_ref class-attribute instance-attribute
legacy_insurer_ref = NOT_SET
onboarding_documents class-attribute instance-attribute
onboarding_documents = NOT_SET
postal_code class-attribute instance-attribute
postal_code = NOT_SET
province class-attribute instance-attribute
province = NOT_SET
retirement_date class-attribute instance-attribute
retirement_date = NOT_SET
settlement_account_number class-attribute instance-attribute
settlement_account_number = NOT_SET
start_date class-attribute instance-attribute
start_date = NOT_SET
street class-attribute instance-attribute
street = NOT_SET

terms_document

TermsDocumentEntity dataclass

TermsDocumentEntity(type, uri)

Bases: DataClassJsonMixin

Represents an onboarding terms document to be acknowledged by the user

type instance-attribute
type
uri instance-attribute
uri

user_info

UserInfo dataclass

UserInfo(
    first_name=None,
    last_name=None,
    birthdate=None,
    gender=None,
    identification_number=None,
    street=None,
    postal_code=None,
    city=None,
    province=None,
    country=None,
    settlement_account_number=None,
    settlement_account_type=None,
    billing_account_number=None,
    billing_account_type=None,
)

Bases: DataClassJsonMixin

Dataclass containing user information used to fill the onboarding fields

billing_account_number class-attribute instance-attribute
billing_account_number = None
billing_account_type class-attribute instance-attribute
billing_account_type = None
birthdate class-attribute instance-attribute
birthdate = None
city class-attribute instance-attribute
city = None
country class-attribute instance-attribute
country = None
first_name class-attribute instance-attribute
first_name = None
gender class-attribute instance-attribute
gender = None
identification_number class-attribute instance-attribute
identification_number = None
last_name class-attribute instance-attribute
last_name = None
postal_code class-attribute instance-attribute
postal_code = None
province class-attribute instance-attribute
province = None
settlement_account_number class-attribute instance-attribute
settlement_account_number = None
settlement_account_type class-attribute instance-attribute
settlement_account_type = None
street class-attribute instance-attribute
street = None

components.onboarding.public.enums

account_type

AccountType

Bases: AlanBaseEnum

ca_account class-attribute instance-attribute
ca_account = 'ca_account'
iban class-attribute instance-attribute
iban = 'iban'

onboarding_type

OnboardingType

Bases: AlanBaseEnum

company_admin_onboarding class-attribute instance-attribute
company_admin_onboarding = 'company_admin_onboarding'

DEPRECATED: Replaced by customer_admin_onboarding, keeping for backward compatibility

customer_admin_onboarding class-attribute instance-attribute
customer_admin_onboarding = 'customer_admin_onboarding'
dependent_onboarding class-attribute instance-attribute
dependent_onboarding = 'dependent_onboarding'
employee_onboarding class-attribute instance-attribute
employee_onboarding = 'employee_onboarding'
partner_onboarding class-attribute instance-attribute
partner_onboarding = 'partner_onboarding'
retiree_onboarding class-attribute instance-attribute
retiree_onboarding = 'retiree_onboarding'

person_onboarding_status

PersonOnboardingStatus

Bases: AlanBaseEnum

completed class-attribute instance-attribute
completed = 'completed'
in_progress class-attribute instance-attribute
in_progress = 'in_progress'
not_started class-attribute instance-attribute
not_started = 'not_started'

retiree_onboarding_type

RetireeOnboardingType

Bases: AlanBaseEnum

Specifies the type of retiree onboarding.

be class-attribute instance-attribute
be = 'be'
fr class-attribute instance-attribute
fr = 'fr'
fr_mtect class-attribute instance-attribute
fr_mtect = 'fr_mtect'

terms_document_type

TermsDocumentType

Bases: AlanBaseEnum

employee_booklet class-attribute instance-attribute
employee_booklet = 'employee_booklet'
life_and_disability_booklet class-attribute instance-attribute
life_and_disability_booklet = 'life_and_disability_booklet'
privacy_policy class-attribute instance-attribute
privacy_policy = 'privacy_policy'
terms_of_use class-attribute instance-attribute
terms_of_use = 'terms_of_use'

components.onboarding.public.errors

exceptions

AlreadyCompletedOnboardingException

AlreadyCompletedOnboardingException()

Bases: Exception

Source code in components/onboarding/public/errors/exceptions.py
def __init__(self) -> None:
    super().__init__("This onboarding has already been completed")

AlreadyCoveredException

AlreadyCoveredException()

Bases: Exception

Source code in components/onboarding/public/errors/exceptions.py
def __init__(self) -> None:
    super().__init__("This member is already covered")

AlreadyUsedOnboardingException

AlreadyUsedOnboardingException()

Bases: Exception

Source code in components/onboarding/public/errors/exceptions.py
def __init__(self) -> None:
    super().__init__("This onboarding has already been used")

InvalidPartnerInvite

InvalidPartnerInvite()

Bases: Exception

Source code in components/onboarding/public/errors/exceptions.py
def __init__(self) -> None:
    super().__init__("Invited partner should not have a user email.")

components.onboarding.public.exceptions

BankAccountValidationError

Bases: Exception

LegacyContractDelayValidationError

Bases: Exception

OnboardingAlreadyPendingError

OnboardingAlreadyPendingError(
    existing_person_onboarding_id,
    message="A pending onboarding already exists",
)

Bases: Exception

Source code in components/onboarding/public/exceptions.py
def __init__(
    self,
    existing_person_onboarding_id: UUID,
    message: str = "A pending onboarding already exists",
):
    super().__init__(message)
    self.message = message
    self.existing_person_onboarding_id = existing_person_onboarding_id

existing_person_onboarding_id instance-attribute

existing_person_onboarding_id = (
    existing_person_onboarding_id
)

message instance-attribute

message = message

OnboardingMissingInvitationEmailError

OnboardingMissingInvitationEmailError(
    person_onboarding_id,
    message="Missing email for onboarding",
)

Bases: Exception

Source code in components/onboarding/public/exceptions.py
def __init__(
    self,
    person_onboarding_id: UUID,
    message: str = "Missing email for onboarding",
):
    super().__init__(message)
    self.message = message
    self.person_onboarding_id = person_onboarding_id

existing_person_onboarding_id instance-attribute

existing_person_onboarding_id

message instance-attribute

message = message

person_onboarding_id instance-attribute

person_onboarding_id = person_onboarding_id

StartDateInThePastError

Bases: Exception

components.onboarding.public.queries

get_customer_admin_onboarding

get_customer_admin_onboarding

get_customer_admin_onboarding(person_onboarding_id)

Get customer admin onboarding by person onboarding ID.

Source code in components/onboarding/internal/queries/customer_admin.py
def get_customer_admin_onboarding(
    person_onboarding_id: UUID,
) -> CustomerAdminOnboarding:
    """
    Get customer admin onboarding by person onboarding ID.
    """
    customer_admin_onboarding: PersonOnboarding = (
        CustomerAdminOnboardingModelBroker.find_by_id(person_onboarding_id).one()
    )
    return get_customer_admin_onboarding_from_model(customer_admin_onboarding)

get_invitation_email_for_users_and_entities

get_invitation_email_for_users_and_entities(
    user_id_to_entities,
)

:param user_id_to_entities: dict used as input to nudge towards performant data loading patterns. In practice this means calling this query function once instead of once per user. :return: most recent invitation email if any for each user and entities

Source code in components/onboarding/internal/queries/customer_admin.py
def get_invitation_email_for_users_and_entities(
    user_id_to_entities: Mapping[str, set[tuple[str, EntityType]]],
) -> dict[str, str | None]:
    """

    :param user_id_to_entities: dict used as input to nudge towards
        performant data loading patterns.
        In practice this means calling this query function once
        instead of once per user.
    :return: most recent invitation email if any for each user and entities
    """

    select_statement = (
        CustomerAdminOnboardingModelBroker()
        .select()
        .join(
            CustomerAdminInvitationDetails,
            PersonOnboarding.id == CustomerAdminInvitationDetails.person_onboarding_id,
        )
        .join(
            CustomerAdminInvitationToEntityModel,
            CustomerAdminInvitationDetails.id
            == CustomerAdminInvitationToEntityModel.customer_admin_invitation_details_id,
        )
        .filter(
            PersonOnboarding.is_cancelled.is_(False),
            or_(
                *[
                    and_(
                        PersonOnboarding.user_id == user_id,
                        PersonOnboarding.invitation_email.isnot(None),
                        or_(
                            *[
                                and_(
                                    CustomerAdminInvitationToEntityModel.entity_type
                                    == entity_type,
                                    CustomerAdminInvitationToEntityModel.entity_id
                                    == entity_id,
                                )
                                for entity_id, entity_type in entities
                            ]
                        ),
                    )
                    for user_id, entities in user_id_to_entities.items()
                ]
            ),
        )
        .order_by(PersonOnboarding.user_id, PersonOnboarding.created_at.desc())
        # Using DISTINCT ON to take the last onboarding for a given user
        .distinct(PersonOnboarding.user_id)
    )

    customer_admin_onboardings = current_session.scalars(select_statement).all()

    # Preset default None value for all user ids to have a key for each user id
    # while keeping the code simple.
    # Only onboardings with an invitation_email are fetched,
    # and some users might not have any onboarding.
    # This can happen if an existing employee was promoted as admin for example.
    user_id_to_invitation_email = dict.fromkeys(user_id_to_entities.keys())
    for customer_admin_onboarding in customer_admin_onboardings:
        user_id_to_invitation_email[customer_admin_onboarding.user_id] = (
            customer_admin_onboarding.invitation_email
        )

    return user_id_to_invitation_email

get_pending_customer_admin_onboarding

get_pending_customer_admin_onboarding(person_onboarding_id)

Get customer admin onboarding by person onboarding ID.

Source code in components/onboarding/internal/queries/customer_admin.py
def get_pending_customer_admin_onboarding(
    person_onboarding_id: UUID,
) -> CustomerAdminOnboarding:
    """
    Get customer admin onboarding by person onboarding ID.
    """
    customer_admin_onboarding: PersonOnboarding = (
        CustomerAdminOnboardingModelBroker.find_by_id(person_onboarding_id)
        .filter(PersonOnboarding.onboarding_status != PersonOnboardingStatus.completed)
        .one()
    )
    return get_customer_admin_onboarding_from_model(customer_admin_onboarding)

get_pending_customer_admin_onboarding_for_entities

get_pending_customer_admin_onboarding_for_entities(
    entities,
    sort_filter,
    sort_direction,
    cursor=None,
    limit=None,
)

Get all customer admin onboardings targeting the specified entities that are not completed.

Source code in components/onboarding/internal/queries/customer_admin.py
def get_pending_customer_admin_onboarding_for_entities(
    entities: set[tuple[str, EntityType]],
    sort_filter: AlanBaseEnum,
    sort_direction: PaginationSortDirectionType,
    cursor: int | None = None,
    limit: int | None = None,
) -> list[CustomerAdminOnboarding]:
    """
    Get all customer admin onboardings targeting the specified entities that are not completed.
    """
    select_statement = (
        CustomerAdminOnboardingModelBroker()
        .get_invitations_for_entities(entities)
        .filter(
            PersonOnboarding.onboarding_status != PersonOnboardingStatus.completed,
        )
    )

    paginated_query = paginate_query(
        query=select_statement,  # type: ignore[arg-type]
        sort_filter=sort_filter,
        sort_direction=sort_direction,
        cursor=cursor,
        limit=limit,
        sort_model=CustomerAdminOnboardingModelBroker.model,
    )

    result = current_session.execute(paginated_query).scalars()
    return [
        get_customer_admin_onboarding_from_model(onboarding) for onboarding in result
    ]

get_employee_onboardings

get_employee_onboardings

get_employee_onboardings(user_ids)

Get employee onboarding for each user ids provided.

Source code in components/onboarding/public/queries/get_employee_onboardings.py
def get_employee_onboardings(user_ids: list[str]) -> list[EmployeeOnboardingEntity]:
    """
    Get employee onboarding for each user ids provided.
    """
    from components.onboarding.internal.queries.employee_onboarding import (
        get_employee_onboardings_by_user_ids as _get_employee_onboardings_by_user_ids,
    )

    return _get_employee_onboardings_by_user_ids(user_ids)

get_first_employee_onboarding_by_user

get_first_employee_onboarding_by_user(user_id)

Get employee onboarding for a given user id.

Source code in components/onboarding/public/queries/get_employee_onboardings.py
def get_first_employee_onboarding_by_user(
    user_id: str,
) -> Optional[EmployeeOnboardingEntity]:
    """
    Get employee onboarding for a given user id.
    """
    from components.onboarding.internal.queries.employee_onboarding import (
        get_first_employee_onboarding_by_user as _get_first_employee_onboarding_by_user,
    )

    return _get_first_employee_onboarding_by_user(user_id)

get_person_onboarding

get_person_onboarding

get_person_onboarding(
    person_onboarding_id, with_cancelled=False
)
Source code in components/onboarding/internal/queries/person_onboarding.py
def get_person_onboarding(
    person_onboarding_id: UUID, with_cancelled: bool = False
) -> PersonOnboardingEntity:
    person_onboarding = PersonOnboardingModelBroker.find_by_id(
        person_onboarding_id, with_cancelled=True
    ).one_or_none()

    if person_onboarding is None:
        raise BaseErrorCode.missing_resource(message="Person onboarding not found")
    elif person_onboarding.is_cancelled and not with_cancelled:
        raise BaseErrorCode.employee_onboarding_was_cancelled()

    return get_person_onboarding_entity(person_onboarding)

get_person_onboarding_by_invitation_email

get_person_onboarding_by_invitation_email(invitation_email)
Source code in components/onboarding/internal/queries/person_onboarding.py
def get_person_onboarding_by_invitation_email(
    invitation_email: str,
) -> Optional[PersonOnboardingEntity]:
    person_onboarding = PersonOnboardingModelBroker.find_by_invitation_email(
        invitation_email
    ).one_or_none()
    return (
        get_person_onboarding_entity(person_onboarding) if person_onboarding else None
    )

get_person_onboarding_by_user

get_person_onboarding_by_user(user_id)
Source code in components/onboarding/internal/queries/person_onboarding.py
def get_person_onboarding_by_user(
    user_id: str,
) -> Optional[PersonOnboardingEntity]:
    person_onboarding = PersonOnboardingModelBroker.find_by_user_id(
        user_id
    ).one_or_none()
    return (
        get_person_onboarding_entity(person_onboarding) if person_onboarding else None
    )

get_person_onboardings_by_user

get_person_onboardings_by_user(
    user_id, with_cancelled=False
)
Source code in components/onboarding/internal/queries/person_onboarding.py
def get_person_onboardings_by_user(
    user_id: str, with_cancelled: bool = False
) -> list[PersonOnboardingEntity]:
    person_onboardings = PersonOnboardingModelBroker.find_by_user_id_order_latest(
        user_id, with_cancelled=with_cancelled
    ).all()
    return [
        get_person_onboarding_entity(person_onboarding)
        for person_onboarding in person_onboardings
    ]

get_retiree_onboarding

get_retiree_onboarding

get_retiree_onboarding(retiree_onboarding_id)

Get retiree onboarding with given person_onboarding_id

Source code in components/onboarding/public/queries/get_retiree_onboarding.py
def get_retiree_onboarding(retiree_onboarding_id: UUID) -> RetireeOnboardingEntity:
    """
    Get retiree onboarding with given person_onboarding_id
    """
    from components.onboarding.internal.queries.retiree_onboarding import (
        get_retiree_onboarding as _get_retiree_onboarding,
    )

    return _get_retiree_onboarding(retiree_onboarding_id)

get_retiree_onboarding_from_token

get_retiree_onboarding_from_token(token)

Get retiree onboarding with given token

Source code in components/onboarding/public/queries/get_retiree_onboarding.py
def get_retiree_onboarding_from_token(token: str) -> RetireeOnboardingEntity | None:
    """
    Get retiree onboarding with given token
    """
    from components.onboarding.internal.controllers.person_onboarding import (
        PersonOnboardingToken,
    )
    from components.onboarding.internal.queries.retiree_onboarding import (
        get_retiree_onboarding as _get_retiree_onboarding,
    )

    person_onboarding_id, _error = PersonOnboardingToken.get_from_token(
        id=None,
        token=token,
    )

    if person_onboarding_id is None:
        return None

    return _get_retiree_onboarding(person_onboarding_id)

get_onboarding_landing_url

get_onboarding_landing_url(
    person_onboarding_id, with_cancelled=False
)

Returns the landing page URL for this person onboarding, based on its type

See FRONT_END_KEY_MAP_LANDING_PAGE above

e.g. https://alan.com/en-ca/onboarding/customer-admin-onboarding?token= ⧉...

Source code in components/onboarding/internal/queries/invitation_link.py
def get_onboarding_landing_url(
    person_onboarding_id: UUID,
    with_cancelled: bool = False,
) -> str | None:
    """
    Returns the landing page URL for this person onboarding, based on its type

    See FRONT_END_KEY_MAP_LANDING_PAGE above

    e.g. https://alan.com/en-ca/onboarding/customer-admin-onboarding?token=...
    """
    person_onboarding = PersonOnboardingModelBroker.find_by_id(
        person_onboarding_id, with_cancelled
    ).one_or_none()
    if not person_onboarding:
        return None

    front_end_key = FRONT_END_KEY_MAP_LANDING_PAGE.get(
        person_onboarding.onboarding_type
    )
    if not front_end_key:
        return None

    return get_frontend_url().build_url(
        front_end_key,
        query_args={"token": person_onboarding.authentication_token},
    )
get_onboarding_mobile_deeplink(
    person_onboarding_id, with_cancelled=False
)

Returns the mobile deeplink for this person onboarding, based on its type

See FRONT_END_KEY_MAP_DEEPLINK above

e.g. alanmobile://employee_onboarding?country=ca&token=...

Source code in components/onboarding/internal/queries/invitation_link.py
def get_onboarding_mobile_deeplink(
    person_onboarding_id: UUID,
    with_cancelled: bool = False,
) -> str | None:
    """
    Returns the mobile deeplink for this person onboarding, based on its type

    See FRONT_END_KEY_MAP_DEEPLINK above

    e.g. alanmobile://employee_onboarding?country=ca&token=...
    """
    person_onboarding = PersonOnboardingModelBroker.find_by_id(
        person_onboarding_id, with_cancelled
    ).one_or_none()
    if not person_onboarding:
        return None

    front_end_key = FRONT_END_KEY_MAP_DEEPLINK.get(person_onboarding.onboarding_type)
    if not front_end_key:
        return None

    return get_frontend_url().build_deep_link(
        front_end_key,
        query_args={"token": person_onboarding.authentication_token},
    )