Skip to content

Api reference

components.onboarding.public.actions

cancel_onboarding_invitation

cancel_onboarding_invitation 

cancel_onboarding_invitation(
    onboarding_invitation_id, commit=True
)

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

Source code in components/onboarding/internal/actions/cancel_onboarding_invitation.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def cancel_onboarding_invitation(
    onboarding_invitation_id: UUID, commit: bool = True
) -> None:
    """
    Cancels the onboarding invitation. You can only cancel a
    non-completed onboarding
    """
    if (
        onboarding_invitation := OnboardingInvitationModelBroker.find_by_id(
            onboarding_invitation_id,
        ).one_or_none()
    ) is None:
        raise ValueError(
            f"Onboarding invitation with ID {onboarding_invitation_id} does not exist.",
        )

    if (
        onboarding_invitation.onboarding_status == OnboardingInvitationStatus.completed
        or onboarding_invitation.is_cancelled
    ):
        raise ValueError(
            f"Cannot cancel an onboarding invitation that is already completed or "
            f"cancelled. Onboarding: {onboarding_invitation_id}",
        )

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

    current_logger.info(
        f"Cancelled onboarding invitation {onboarding_invitation_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_onboarding_invitation

Source code in components/onboarding/internal/actions/customer_admin.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def cancel_customer_admin_invitation(invitation_id: UUID) -> None:
    """
    Remove a customer admin invitation by its ID.

    todo @alexandre.dubreuil - Duplicates cancel_onboarding_invitation
    """

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

    if (
        customer_admin_onboarding.onboarding_status
        == OnboardingInvitationStatus.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 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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
        # TODO can be replaced by customer_admin public API once all countries are setup
        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 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
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
        == OnboardingInvitationStatus.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},
    )

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 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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 = OnboardingInvitationModelBroker.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

update_onboarding 

update_onboarding(
    onboarding_invitation_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,
    lang=None,
    email=None,
    commit=True,
)
Source code in components/onboarding/internal/actions/update_onboarding.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def update_onboarding(
    onboarding_invitation_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,
    lang: Optional[Lang] = None,
    email: Optional[str] = None,
    commit: bool = True,
) -> None:
    onboarding_progress = (
        OnboardingProgressBroker.find_by_onboarding_invitation_id(
            onboarding_invitation_id
        )
    ).one()

    onboarding_invitation = onboarding_progress.onboarding_invitation

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

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

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

    if birthdate:
        onboarding_progress.birthdate = birthdate

    if gender:
        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,
        )
        onboarding_progress.street = address.street
        onboarding_progress.postal_code = address.postal_code
        onboarding_progress.province = address.administrative_area
        onboarding_progress.city = address.locality

    if lang:
        onboarding_invitation.lang = lang

    if viewed_terms_at:
        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()
            )
        )
        onboarding_progress.billing_account_number = (
            validated_billing_account.account_number
        )
        onboarding_progress.billing_account_type = (
            validated_billing_account.account_type
        )
    if email:
        onboarding_progress.email = email

    onboarding_entity = get_onboarding_entity(onboarding_progress)
    if get_app_dependency().is_employee_onboarding_completed(onboarding_entity):
        onboarding_invitation.onboarding_status = OnboardingInvitationStatus.completed
        # Call dependency so that they can updated themselves on completed (e.g. user.name, etc.)
        get_app_dependency().on_completed_onboarding(onboarding_entity)
    else:
        onboarding_invitation.onboarding_status = OnboardingInvitationStatus.in_progress

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

    current_logger.info(
        "Updated onboarding invitation",
        onboarding_invitation_id=onboarding_invitation.id,
        onboarding_progress_id=onboarding_progress.id,
    )

onboarding_invitation_emails

resend_onboarding_invitation_email 

resend_onboarding_invitation_email(
    onboarding_invitation_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 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def resend_onboarding_invitation_email(onboarding_invitation_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,
    )

    onboarding_invitation = get_onboarding_invitation(
        onboarding_invitation_id=onboarding_invitation_id, with_cancelled=True
    )

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

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

    current_logger.info(
        "Resend employee onboarding invitation",
        onboarding_component={"onboarding_invitation_id": onboarding_invitation_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_onboarding_invitation_as_exempted 

complete_onboarding_invitation_as_exempted(
    onboarding_invitation_id,
)

Complete an onboarding invitation with outcome 'exempted'.

Source code in components/onboarding/public/api.py 
119
120
121
122
123
124
125
126
127
def complete_onboarding_invitation_as_exempted(onboarding_invitation_id: UUID) -> None:
    """Complete an onboarding invitation with outcome 'exempted'."""
    from components.onboarding.internal.actions.complete_onboarding_invitation_as_exempted import (
        complete_onboarding_invitation_as_exempted as complete_onboarding_invitation_as_exempted_,
    )

    return complete_onboarding_invitation_as_exempted_(
        onboarding_invitation_id=onboarding_invitation_id
    )

complete_retiree_onboarding 

complete_retiree_onboarding(
    onboarding_invitation_id, commit=False
)

Mark the retiree onboarding object as completed

Source code in components/onboarding/public/api.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
def complete_retiree_onboarding(
    onboarding_invitation_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_(
        onboarding_invitation_id=onboarding_invitation_id, commit=commit
    )

get_first_onboarding_by_email 

get_first_onboarding_by_email(invitation_email)
Source code in components/onboarding/public/api.py 
39
40
41
42
43
44
45
46
def get_first_onboarding_by_email(  # noqa: D103
    invitation_email: str,
) -> OnboardingEntity | None:
    from components.onboarding.internal.queries.onboarding import (
        get_onboarding_by_email as get_onboarding_by_email_,
    )

    return get_onboarding_by_email_(invitation_email)

get_pending_onboardings 

get_pending_onboardings(user_id)
Source code in components/onboarding/public/api.py 
 96
 97
 98
 99
100
101
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 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
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,
) -> OnboardingEntity:
    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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
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 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@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.onboarding import (
        OnboardingEntity,
    )
    from components.onboarding.public.entities.onboarding_invitation import (
        OnboardingInvitationEntity,
    )
    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_invitation_status import (
        OnboardingInvitationStatus,
    )
    from components.onboarding.public.enums.onboarding_type import OnboardingType
    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,
            OnboardingInvitationStatus,
            RetireeOnboardingType,
            AccountType,
            TermsDocumentType,
            UserGender,
            # Models
            OnboardingInvitationEntity,
            OnboardingEntity,
            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 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@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 
293
294
295
296
297
298
299
@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 
285
286
287
288
289
290
291
@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 
 97
 98
 99
100
101
102
103
104
@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 
110
111
112
113
114
115
116
@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 
68
69
70
71
72
73
74
75
@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 
245
246
247
248
249
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 
251
252
253
254
255
256
257
258
259
260
261
262
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 
85
86
87
@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 
93
94
95
@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 
47
48
49
50
51
52
53
54
55
56
57
58
@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 
89
90
91
@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 
77
78
79
@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 
81
82
83
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 
106
107
108
@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 
173
174
175
176
177
178
@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 
60
61
62
@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 
64
65
66
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 
180
181
182
183
184
185
186
def is_employee_onboarding_completed(self, onboarding: OnboardingEntity) -> bool:
    """Implement is_employee_onboarding_completed (override for specific completed rules)"""
    from components.onboarding.internal.actions.update_onboarding import (
        is_onboarding_completed,
    )

    return is_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 
188
189
190
191
192
193
194
195
196
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 
198
199
200
201
202
203
204
205
206
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 
208
209
210
211
212
213
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 
264
265
266
267
268
269
270
271
272
273
@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 
239
240
241
242
243
@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 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
@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 
233
234
235
236
237
@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 
277
278
279
280
281
282
283
@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 
118
119
120
121
122
123
124
125
126
@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 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
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 
128
129
130
131
132
133
134
135
136
@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,
    company_contract_start_date=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!

Source code in components/onboarding/public/dependencies.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def validates_start_date(
    self,
    coverage_start_date: date,  # noqa: ARG002
    start_date_is_read_only: bool,  # noqa: ARG002
    company_contract_start_date: date | None = None,  # 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 
310
311
312
313
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 
302
303
304
305
306
307
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, contract_start_date=None)

Bases: DataClassJsonMixin

Company entity.

contract_start_date class-attribute instance-attribute 
contract_start_date = None
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,
    target_ref=None,
    external_employee_id,
    entities
)

Bases: OnboardingInvitationEntity, DataClassJsonMixin

Dataclass representing a customer admin onboarding.

entities instance-attribute 
entities
external_employee_id instance-attribute 
external_employee_id

onboarding

OnboardingEntity dataclass 

OnboardingEntity(
    *,
    onboarding_invitation_id,
    onboarding_type,
    onboarding_status,
    invitation_email,
    authentication_token,
    user_id,
    target_ref,
    start_date=optional_isodate_field(),
    lang,
    is_offline,
    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(),
    email=None,
    created_at=isodatetime_field(),
    onboarding_outcome=None
)

Bases: BaseOnboardingProgress, DataClassJsonMixin

__post_init__ 
__post_init__()

Set employee_onboarding_progress_id for backward compatibility.

Source code in components/onboarding/public/entities/onboarding.py 
44
45
46
47
48
49
def __post_init__(self) -> None:
    """Set employee_onboarding_progress_id for backward compatibility."""
    super().__post_init__()
    object.__setattr__(
        self, "employee_onboarding_progress_id", self.onboarding_progress_id
    )
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()
email class-attribute instance-attribute 
email = None
employee_onboarding_progress_id class-attribute instance-attribute 
employee_onboarding_progress_id = field(init=False)
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
onboarding_outcome class-attribute instance-attribute 
onboarding_outcome = None
onboarding_progress_id instance-attribute 
onboarding_progress_id
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_invitation

OnboardingInvitationEntity dataclass 

OnboardingInvitationEntity(
    *,
    id,
    onboarding_type,
    onboarding_status,
    invitation_email,
    authentication_token,
    user_id,
    first_name,
    last_name,
    lang,
    is_offline,
    start_date,
    is_cancelled,
    target_ref=None
)

Bases: DataClassJsonMixin

Dataclass representing an onboarding invitation.

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
target_ref class-attribute instance-attribute 
target_ref = None
user_id instance-attribute 
user_id

onboarding_progress

BaseOnboardingProgress dataclass 

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

Bases: DataClassJsonMixin

Base onboarding progress entity.

__post_init__ 
__post_init__()

Set person_onboarding_id to same value as onboarding_invitation_id for backward compatibility.

Source code in components/onboarding/public/entities/onboarding_progress.py 
33
34
35
def __post_init__(self) -> None:
    """Set person_onboarding_id to same value as onboarding_invitation_id for backward compatibility."""
    object.__setattr__(self, "person_onboarding_id", self.onboarding_invitation_id)
authentication_token instance-attribute 
authentication_token
invitation_email instance-attribute 
invitation_email
is_offline instance-attribute 
is_offline
lang instance-attribute 
lang
onboarding_invitation_id instance-attribute 
onboarding_invitation_id
onboarding_status instance-attribute 
onboarding_status
onboarding_type instance-attribute 
onboarding_type
person_onboarding_id class-attribute instance-attribute 
person_onboarding_id = field(init=False)
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(
    *,
    onboarding_invitation_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

PendingOnboardingEntity dataclass 

PendingOnboardingEntity(employment_id)

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

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,
    contract_id,
)

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
contract_id instance-attribute 
contract_id
id instance-attribute 
id
is_offline instance-attribute 
is_offline
onboarding_token instance-attribute 
onboarding_token
onboarding_url instance-attribute 
onboarding_url
retirement_date instance-attribute 
retirement_date

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 
 94
 95
 96
 97
 98
 99
100
101
102
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(
    *,
    onboarding_invitation_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,
    company_contract_start_date=optional_isodate_field(),
    contract_ref,
    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_contract_start_date class-attribute instance-attribute 
company_contract_start_date = optional_isodate_field()
company_id instance-attribute 
company_id
company_name instance-attribute 
company_name
contract_ref instance-attribute 
contract_ref
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_invitation_outcome

OnboardingInvitationOutcome

Bases: AlanBaseEnum

Outcome of an onboarding invitation.

covered class-attribute instance-attribute 
covered = 'covered'

The person is covered by an insurance plan.

exempted class-attribute instance-attribute 
exempted = 'exempted'

The person is exempted from an insurance plan.

onboarding_invitation_status

OnboardingInvitationStatus

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'

onboarding_type

EmployeeOnboardingTypes module-attribute 

EmployeeOnboardingTypes = [
    employee_onboarding,
    ca_employee_onboarding,
]

OnboardingType

Bases: AlanBaseEnum

ca_employee_onboarding class-attribute instance-attribute 
ca_employee_onboarding = 'ca_employee_onboarding'
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'

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 
7
8
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 
12
13
def __init__(self) -> None:
    super().__init__("This member is already covered")

AlreadyUsedOnboardingException 

AlreadyUsedOnboardingException()

Bases: Exception

Source code in components/onboarding/public/errors/exceptions.py 
2
3
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 
17
18
def __init__(self) -> None:
    super().__init__("Invited partner should not have a user email.")

components.onboarding.public.exceptions

BadContractIdError 

BadContractIdError(message='Generic onboarding error')

Bases: OnboardingError

Source code in components/onboarding/internal/exceptions/exceptions.py 
16
17
18
19
20
21
22
23
24
25
def __init__(
    self,
    message: str = "Generic onboarding error",
):
    super().__init__(
        message=message,
        code=self.__class__.code,
        type=self.type,
        description=self.__class__.description,
    )

code class-attribute instance-attribute 

code = 1104

description class-attribute instance-attribute 

description = 'The provided contract id is not right'

key class-attribute instance-attribute 

key = ONBOARDING_BAD_CONTRACT_ID_ERROR

BankAccountValidationError 

BankAccountValidationError(
    message="Generic onboarding error",
)

Bases: OnboardingError

Source code in components/onboarding/internal/exceptions/exceptions.py 
16
17
18
19
20
21
22
23
24
25
def __init__(
    self,
    message: str = "Generic onboarding error",
):
    super().__init__(
        message=message,
        code=self.__class__.code,
        type=self.type,
        description=self.__class__.description,
    )

code class-attribute instance-attribute 

code = 912

description class-attribute instance-attribute 

description = "Could not validate the provided bank account information"

key class-attribute instance-attribute 

key = ONBOARDING_BANK_ACCOUNT_VALIDATION_ERROR

LegacyContractDelayValidationError 

LegacyContractDelayValidationError(
    message="Generic onboarding error",
)

Bases: OnboardingError

Source code in components/onboarding/internal/exceptions/exceptions.py 
16
17
18
19
20
21
22
23
24
25
def __init__(
    self,
    message: str = "Generic onboarding error",
):
    super().__init__(
        message=message,
        code=self.__class__.code,
        type=self.type,
        description=self.__class__.description,
    )

code class-attribute instance-attribute 

code = 1101

description class-attribute instance-attribute 

description = "The coverage start date must be long enough after legacy contract start date"

key class-attribute instance-attribute 

key = ONBOARDING_LEGACY_CONTRACT_DELAY_ERROR

OnboardingAlreadyPendingError 

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

Bases: OnboardingError

Source code in components/onboarding/internal/exceptions/exceptions.py 
70
71
72
73
74
75
76
77
78
79
def __init__(
    self,
    existing_onboarding_invitation_id: UUID,
    message: str = "A pending onboarding already exists",
):
    super().__init__(
        message=message,
    )
    self.message = message
    self.existing_onboarding_invitation_id = existing_onboarding_invitation_id

code class-attribute instance-attribute 

code = 1105

description class-attribute instance-attribute 

description = 'A pending onboarding already exists'

existing_onboarding_invitation_id instance-attribute 

existing_onboarding_invitation_id = (
    existing_onboarding_invitation_id
)

key class-attribute instance-attribute 

key = ONBOARDING_ALREADY_PENDING_ERROR

message instance-attribute 

message = message

OnboardingMissingInvitationEmailError 

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

Bases: OnboardingError

Source code in components/onboarding/internal/exceptions/exceptions.py 
88
89
90
91
92
93
94
95
96
97
def __init__(
    self,
    onboarding_invitation_id: UUID,
    message: str = "Missing email for onboarding",
):
    super().__init__(
        message=message,
    )
    self.message = message
    self.onboarding_invitation_id = onboarding_invitation_id

code class-attribute instance-attribute 

code = 1106

description class-attribute instance-attribute 

description = 'Missing email for onboarding'

key class-attribute instance-attribute 

key = ONBOARDING_MISSING_INVITATION_ERROR

message instance-attribute 

message = message

onboarding_invitation_id instance-attribute 

onboarding_invitation_id = onboarding_invitation_id

StartDateBeforeContractError 

StartDateBeforeContractError(
    message="Generic onboarding error",
)

Bases: OnboardingError

Source code in components/onboarding/internal/exceptions/exceptions.py 
16
17
18
19
20
21
22
23
24
25
def __init__(
    self,
    message: str = "Generic onboarding error",
):
    super().__init__(
        message=message,
        code=self.__class__.code,
        type=self.type,
        description=self.__class__.description,
    )

code class-attribute instance-attribute 

code = 1103

description class-attribute instance-attribute 

description = "The provided onboarding start date must not be before contract start date"

key class-attribute instance-attribute 

key = ONBOARDING_START_DATE_BEFORE_CONTRACT_ERROR

StartDateInThePastError 

StartDateInThePastError(message='Generic onboarding error')

Bases: OnboardingError

Source code in components/onboarding/internal/exceptions/exceptions.py 
16
17
18
19
20
21
22
23
24
25
def __init__(
    self,
    message: str = "Generic onboarding error",
):
    super().__init__(
        message=message,
        code=self.__class__.code,
        type=self.type,
        description=self.__class__.description,
    )

code class-attribute instance-attribute 

code = 1102

description class-attribute instance-attribute 

description = (
    "The onboarding start date cannot be in the past"
)

key class-attribute instance-attribute 

key = ONBOARDING_START_DATE_IN_THE_PAST_ERROR

components.onboarding.public.queries

get_customer_admin_onboarding

get_customer_admin_onboarding 

get_customer_admin_onboarding(onboarding_invitation_id)

Get customer admin onboarding by onboarding invitation ID.

Source code in components/onboarding/internal/queries/customer_admin.py 
33
34
35
36
37
38
39
40
41
42
def get_customer_admin_onboarding(
    onboarding_invitation_id: UUID,
) -> CustomerAdminOnboarding:
    """
    Get customer admin onboarding by onboarding invitation ID.
    """
    customer_admin_onboarding: OnboardingInvitation = (
        CustomerAdminOnboardingModelBroker.find_by_id(onboarding_invitation_id).one()
    )
    return get_customer_admin_onboarding_from_model(customer_admin_onboarding)

get_customer_admin_onboardings 

get_customer_admin_onboardings(onboarding_ids)

Get customer admin onboardings by onboarding invitation IDs.

Source code in components/onboarding/internal/queries/customer_admin.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def get_customer_admin_onboardings(
    onboarding_ids: Iterable[UUID],
) -> list[CustomerAdminOnboarding]:
    """
    Get customer admin onboardings by onboarding invitation IDs.
    """
    customer_admin_onboardings = current_session.scalars(
        CustomerAdminOnboardingModelBroker.select().filter(
            OnboardingInvitation.id.in_(onboarding_ids)
        )
    ).all()
    return [
        get_customer_admin_onboarding_from_model(customer_admin_onboarding)
        for customer_admin_onboarding in customer_admin_onboardings
    ]

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 
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
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
    """
    user_id_to_onboarding = _get_onboardings_for_users_and_entities(user_id_to_entities)
    return {
        user_id: onboarding.invitation_email if onboarding else None
        for user_id, onboarding in user_id_to_onboarding.items()
    }

get_onboarding_id_for_users_and_entities 

get_onboarding_id_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 onboarding id for user and entities if any

Source code in components/onboarding/internal/queries/customer_admin.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
def get_onboarding_id_for_users_and_entities(
    user_id_to_entities: Mapping[str, set[tuple[str, EntityType]]],
) -> dict[str, UUID | 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 onboarding id for user and entities if any
    """
    user_id_to_onboarding = _get_onboardings_for_users_and_entities(user_id_to_entities)
    return {
        user_id: onboarding.id if onboarding else None
        for user_id, onboarding in user_id_to_onboarding.items()
    }

get_pending_customer_admin_onboarding 

get_pending_customer_admin_onboarding(
    onboarding_invitation_id,
)

Get customer admin onboarding by onboarding invitation ID.

Source code in components/onboarding/internal/queries/customer_admin.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def get_pending_customer_admin_onboarding(
    onboarding_invitation_id: UUID,
) -> CustomerAdminOnboarding:
    """
    Get customer admin onboarding by onboarding invitation ID.
    """
    customer_admin_onboarding: OnboardingInvitation = (
        CustomerAdminOnboardingModelBroker.find_by_id(onboarding_invitation_id)
        .filter(
            OnboardingInvitation.onboarding_status
            != OnboardingInvitationStatus.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 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
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(
            OnboardingInvitation.onboarding_status
            != OnboardingInvitationStatus.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_onboarding_invitation

get_onboarding_invitation 

get_onboarding_invitation(
    onboarding_invitation_id, with_cancelled=False
)
Source code in components/onboarding/internal/queries/onboarding_invitation.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
def get_onboarding_invitation(
    onboarding_invitation_id: UUID, with_cancelled: bool = False
) -> OnboardingInvitationEntity:
    onboarding_invitation = OnboardingInvitationModelBroker.find_by_id(
        onboarding_invitation_id, with_cancelled=True
    ).one_or_none()

    if onboarding_invitation is None:
        raise BaseErrorCode.missing_resource(message="Onboarding invitation not found")
    elif onboarding_invitation.is_cancelled and not with_cancelled:
        raise BaseErrorCode.employee_onboarding_was_cancelled()

    return get_onboarding_invitation_entity(onboarding_invitation)

get_onboarding_invitation_by_invitation_email 

get_onboarding_invitation_by_invitation_email(
    invitation_email,
)
Source code in components/onboarding/internal/queries/onboarding_invitation.py 
 97
 98
 99
100
101
102
103
104
105
106
107
def get_onboarding_invitation_by_invitation_email(
    invitation_email: str,
) -> Optional[OnboardingInvitationEntity]:
    onboarding_invitation = OnboardingInvitationModelBroker.find_by_invitation_email(
        invitation_email
    ).one_or_none()
    return (
        get_onboarding_invitation_entity(onboarding_invitation)
        if onboarding_invitation
        else None
    )

get_onboarding_invitation_by_user 

get_onboarding_invitation_by_user(user_id)
Source code in components/onboarding/internal/queries/onboarding_invitation.py 
58
59
60
61
62
63
64
65
66
67
68
def get_onboarding_invitation_by_user(
    user_id: str,
) -> Optional[OnboardingInvitationEntity]:
    onboarding_invitation = OnboardingInvitationModelBroker.find_by_user_id(
        user_id
    ).one_or_none()
    return (
        get_onboarding_invitation_entity(onboarding_invitation)
        if onboarding_invitation
        else None
    )

get_onboarding_invitation_entity 

get_onboarding_invitation_entity(onboarding_invitation)
Source code in components/onboarding/internal/queries/onboarding_invitation.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def get_onboarding_invitation_entity(
    onboarding_invitation: OnboardingInvitation,
) -> OnboardingInvitationEntity:
    return OnboardingInvitationEntity(
        id=onboarding_invitation.id,
        onboarding_type=onboarding_invitation.onboarding_type,
        onboarding_status=onboarding_invitation.onboarding_status,
        invitation_email=onboarding_invitation.invitation_email,
        authentication_token=onboarding_invitation.authentication_token,
        user_id=onboarding_invitation.user_id,
        first_name=onboarding_invitation.first_name,
        last_name=onboarding_invitation.last_name,
        lang=onboarding_invitation.lang,
        is_offline=onboarding_invitation.is_offline,
        start_date=onboarding_invitation.start_date,
        is_cancelled=onboarding_invitation.is_cancelled,
        target_ref=onboarding_invitation.target_ref,
    )

get_onboarding_invitation_id_from_token 

get_onboarding_invitation_id_from_token(token)
Source code in components/onboarding/internal/queries/onboarding_invitation.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def get_onboarding_invitation_id_from_token(token: str) -> UUID | None:
    from components.onboarding.internal.controllers.onboarding_invitation import (
        OnboardingInvitationToken,
    )

    onboarding_invitation_id, error = OnboardingInvitationToken.get_from_token(
        id=None,
        token=token,
    )

    if error or onboarding_invitation_id is None:
        current_logger.warning(
            "Error verifying token",
            error=error,
            onboarding_invitation_id=onboarding_invitation_id,
        )
        return None

    return onboarding_invitation_id

get_onboarding_invitations_by_user 

get_onboarding_invitations_by_user(
    user_id, with_cancelled=False
)
Source code in components/onboarding/internal/queries/onboarding_invitation.py 
71
72
73
74
75
76
77
78
79
80
81
82
def get_onboarding_invitations_by_user(
    user_id: str, with_cancelled: bool = False
) -> list[OnboardingInvitationEntity]:
    onboarding_invitations = (
        OnboardingInvitationModelBroker.find_by_user_id_order_latest(
            user_id, with_cancelled=with_cancelled
        ).all()
    )
    return [
        get_onboarding_invitation_entity(onboarding_invitation)
        for onboarding_invitation in onboarding_invitations
    ]

search_pending_onboarding_invitations_by_email 

search_pending_onboarding_invitations_by_email(email)

Search for pending OnboardingInvitations by email (exact match).

Source code in components/onboarding/internal/queries/onboarding_invitation.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def search_pending_onboarding_invitations_by_email(
    email: str,
) -> list[OnboardingInvitation]:
    """Search for pending OnboardingInvitations by email (exact match)."""
    onboarding_invitations = (
        OnboardingInvitationModelBroker.find_by_invitation_email(email)
        .filter(
            OnboardingInvitation.onboarding_status
            != OnboardingInvitationStatus.completed,
        )
        .options(selectinload(OnboardingInvitation.retiree_onboarding_progress))
        .all()
    )
    return onboarding_invitations

get_onboardings

get_first_onboarding_by_user 

get_first_onboarding_by_user(user_id)

Get onboarding for a given user id.

Source code in components/onboarding/public/queries/get_onboardings.py 
22
23
24
25
26
27
28
29
30
31
32
def get_first_onboarding_by_user(
    user_id: str,
) -> Optional[OnboardingEntity]:
    """
    Get onboarding for a given user id.
    """
    from components.onboarding.internal.queries.onboarding import (
        get_first_onboarding_by_user,
    )

    return get_first_onboarding_by_user(user_id)

get_onboarding 

get_onboarding(onboarding_invitation_id)

Get onboarding for a given onboarding_invitation_id.

Source code in components/onboarding/public/queries/get_onboardings.py 
 9
10
11
12
13
14
15
16
17
18
19
def get_onboarding(
    onboarding_invitation_id: UUID,
) -> OnboardingEntity:
    """
    Get onboarding for a given onboarding_invitation_id.
    """
    from components.onboarding.internal.queries.onboarding import (
        get_onboarding,
    )

    return get_onboarding(onboarding_invitation_id)

get_onboardings 

get_onboardings(user_ids)

Get onboarding for each user ids provided.

Source code in components/onboarding/public/queries/get_onboardings.py 
35
36
37
38
39
40
41
42
43
def get_onboardings(user_ids: list[str]) -> list[OnboardingEntity]:
    """
    Get onboarding for each user ids provided.
    """
    from components.onboarding.internal.queries.onboarding import (
        get_onboardings_by_user_ids,
    )

    return get_onboardings_by_user_ids(user_ids)

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 
 8
 9
10
11
12
13
14
15
16
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 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def get_retiree_onboarding_from_token(token: str) -> RetireeOnboardingEntity | None:
    """
    Get retiree onboarding with given token
    """
    from sqlalchemy.exc import NoResultFound

    from components.onboarding.internal.controllers.onboarding_invitation import (
        OnboardingInvitationToken,
    )
    from components.onboarding.internal.queries.retiree_onboarding import (
        get_retiree_onboarding as _get_retiree_onboarding,
    )
    from shared.helpers.logging.logger import current_logger

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

    if person_onboarding_id is None:
        return None

    try:
        return _get_retiree_onboarding(person_onboarding_id)
    except NoResultFound:
        current_logger.warn(
            "Retiree onboarding not found", person_onboarding_id=person_onboarding_id
        )
        return None

get_onboarding_landing_url 

get_onboarding_landing_url(
    onboarding_invitation_id, with_cancelled=False
)

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

See FRONT_END_KEY_MAP_LANDING_PAGE above

e.g. https://alan.com/ca/employee-onboarding/[token ⧉...]

Source code in components/onboarding/internal/queries/invitation_link.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_onboarding_landing_url(
    onboarding_invitation_id: UUID,
    with_cancelled: bool = False,
) -> str | None:
    """
    Returns the landing page URL for this onboarding invitation, based on its type

    See FRONT_END_KEY_MAP_LANDING_PAGE above

    e.g. https://alan.com/ca/employee-onboarding/[token...]
    """
    onboarding_invitation = OnboardingInvitationModelBroker.find_by_id(
        onboarding_invitation_id, with_cancelled
    ).one_or_none()
    if not onboarding_invitation:
        return None

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

    if onboarding_invitation.onboarding_type in (
        OnboardingType.employee_onboarding,
        OnboardingType.partner_onboarding,
    ):
        return get_frontend_url().build_url(
            front_end_key,
            base_url=current_config["FR_FRONT_END_BASE_URL"],
            additional_path=onboarding_invitation.authentication_token,
        )

    return get_frontend_url().build_url(
        front_end_key,
        query_args={"token": onboarding_invitation.authentication_token},
    )

get_onboarding_mobile_deeplink(
    onboarding_invitation_id, with_cancelled=False
)

Returns the mobile deeplink for this onboarding invitation, 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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def get_onboarding_mobile_deeplink(
    onboarding_invitation_id: UUID,
    with_cancelled: bool = False,
) -> str | None:
    """
    Returns the mobile deeplink for this onboarding invitation, based on its type

    See FRONT_END_KEY_MAP_DEEPLINK above

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

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

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