Skip to content

Api reference

components.fr.public.admins

queries

get_account_admins_profile_ids 

get_account_admins_profile_ids(account_id)

Get the global_profile_ids of the admins of an account.

Parameters:

Name Type Description Default
account_id UUID

The id of the account.

 required
Source code in components/fr/public/admins/queries.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def get_account_admins_profile_ids(account_id: uuid.UUID) -> set[uuid.UUID]:
    """
    Get the global_profile_ids of the admins of an account.

    Args:
        account_id: The id of the account.
    """
    profile_ids = (
        current_session.execute(
            select(User.profile_id)
            .select_from(CompanyAdmin)
            .join(User, User.id == CompanyAdmin.user_id)
            .filter(CompanyAdmin.account_id == account_id)
        )
        .scalars()
        .all()
    )

    return set(profile_ids)

get_company_admins_profile_ids 

get_company_admins_profile_ids(company_id)

Get the global_profile_ids of the admins of a company.

Parameters:

Name Type Description Default
company_id int

The id of the company.

 required
Source code in components/fr/public/admins/queries.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def get_company_admins_profile_ids(company_id: int) -> set[uuid.UUID]:
    """
    Get the global_profile_ids of the admins of a company.

    Args:
        company_id: The id of the company.
    """
    profile_ids = (
        current_session.execute(
            select(User.profile_id)
            .select_from(CompanyAdmin)
            .join(User, User.id == CompanyAdmin.user_id)
            .filter(CompanyAdmin.company_id == company_id)
        )
        .scalars()
        .all()
    )

    return set(profile_ids)

get_user_can_add_and_refer_company 

get_user_can_add_and_refer_company(
    admined_entity_id, admined_entity_type
)
Source code in components/fr/public/admins/queries.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
48
def get_user_can_add_and_refer_company(  # noqa: D103
    admined_entity_id: str, admined_entity_type: AdminedEntityType
) -> bool:
    from components.customer_product_configuration.public.queries.account_setting import (
        has_account_setting,
    )

    if admined_entity_type == AdminedEntityType.account:
        account_id = admined_entity_id
    elif admined_entity_type == AdminedEntityType.single_company:
        account_id = get_company_id_to_account_id(company_ids=[admined_entity_id])[
            admined_entity_id
        ]
    elif (
        admined_entity_type == AdminedEntityType.operational_scope
        or admined_entity_type == AdminedEntityType.company_for_operational_scope
    ):
        # Operational scope admins can't refer companies
        return False
    else:
        raise NotImplementedError(
            f"admined_entity_type {admined_entity_type} is not supported"
        )

    is_civil_servant = has_account_setting(
        account_id=uuid.UUID(account_id),
        feature_name=AccountFeature.civil_servant,
    )

    return not is_civil_servant

components.fr.public.auth

authorization

AuthorizationStrategies

alaner_admin class-attribute instance-attribute 
alaner_admin = FrAlanerAdminStrategy
authenticated class-attribute instance-attribute 
authenticated = FrAuthenticatedStrategy
authenticated_with_custom_authorization class-attribute instance-attribute 
authenticated_with_custom_authorization = (
    FrAuthenticatedWithCustomAuthorizationStrategy
)
authenticated_with_token_fallback class-attribute instance-attribute 
authenticated_with_token_fallback = (
    FrAuthenticatedOrTokenStrategy
)
authenticated_with_token_fallback_with_custom_authorization class-attribute instance-attribute 
authenticated_with_token_fallback_with_custom_authorization = (
    FrAuthenticatedOrTokenWithCustomAuthorizationStrategy
)
open class-attribute instance-attribute 
open = FrOpenStrategy
owner_only class-attribute instance-attribute 
owner_only = FrOwnerOnlyStrategy

FrAlanerAdminStrategy 

FrAlanerAdminStrategy(permitted_for=None)

Bases: AlanerAdminStrategy

Source code in components/fr/public/auth/authorization.py 
51
52
def __init__(self, permitted_for: set[EmployeePermission] | None = None) -> None:
    super().__init__(permitted_for=permitted_for)

FrAuthenticatedOrTokenStrategy 

FrAuthenticatedOrTokenStrategy(
    token_extractor=None,
    token_param_name="token",
    schemes=None,
    allow_deep_link=False,
)

Bases: FrAuthenticatedStrategy

Support either one pair of param name and token extractor or a scheme of multiple pairs of param name and token extractor.

Source code in components/fr/public/auth/authorization.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def __init__(
    self,
    token_extractor: TokenExtractor | None = None,
    token_param_name: str = "token",  # noqa: S107
    # Mapping of param name to TokenExtractor
    schemes: dict[str, TokenExtractor] | None = None,
    allow_deep_link: bool = False,
) -> None:
    """
    Support either one pair of param name and token extractor or a scheme of multiple
    pairs of param name and token extractor.
    """
    super().__init__(allow_deep_link=allow_deep_link)

    if schemes is not None:
        self.schemes = schemes
    elif token_extractor is not None:
        self.schemes = {token_param_name: token_extractor}
    else:
        raise ValueError("Either token_extractor or schemes must be provided")
authentication_required class-attribute instance-attribute 
authentication_required = False
authenticator 
authenticator()
Source code in components/fr/public/auth/authorization.py 
 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
def authenticator(self) -> Callable:  # type: ignore[type-arg] # noqa: D102
    default_authenticator_decorator = super().authenticator

    def endpoint_decorator(endpoint_fn):  # type: ignore[no-untyped-def]
        @wraps(endpoint_fn)
        @default_authenticator_decorator()
        def decorated_endpoint(*args, **kwargs):  # type: ignore[no-untyped-def]
            matching_token_param_name = next(
                (
                    token_param_name
                    for token_param_name in self.schemes.keys()
                    if token_param_name in request.values
                ),
                None,
            )

            if matching_token_param_name is None:
                return endpoint_fn(*args, **kwargs)

            try:
                token = request.values[matching_token_param_name]
                token_extractor = self.schemes[matching_token_param_name]
                token_content = token_extractor(token=token)
            except BadSignature:
                current_logger.warning("Failed to unsign a authentication token")
                raise BaseErrorCode.authorization_error(
                    message="Invalid token signature"
                )

            # injecting the token payload in controller function's arguments
            for key, value in token_content.items():
                if key in signature(endpoint_fn).parameters:
                    kwargs[key] = value
            return endpoint_fn(*args, **kwargs)

        return decorated_endpoint

    return endpoint_decorator
schemes instance-attribute 
schemes = schemes

FrAuthenticatedOrTokenWithCustomAuthorizationStrategy 

FrAuthenticatedOrTokenWithCustomAuthorizationStrategy(
    token_extractor=None,
    token_param_name="token",
    schemes=None,
    allow_deep_link=False,
)

Bases: FrAuthenticatedOrTokenStrategy

Source code in components/fr/public/auth/authorization.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def __init__(
    self,
    token_extractor: TokenExtractor | None = None,
    token_param_name: str = "token",  # noqa: S107
    # Mapping of param name to TokenExtractor
    schemes: dict[str, TokenExtractor] | None = None,
    allow_deep_link: bool = False,
) -> None:
    """
    Support either one pair of param name and token extractor or a scheme of multiple
    pairs of param name and token extractor.
    """
    super().__init__(allow_deep_link=allow_deep_link)

    if schemes is not None:
        self.schemes = schemes
    elif token_extractor is not None:
        self.schemes = {token_param_name: token_extractor}
    else:
        raise ValueError("Either token_extractor or schemes must be provided")
ensure_custom_authorization class-attribute instance-attribute 
ensure_custom_authorization = True

FrAuthenticatedStrategy 

FrAuthenticatedStrategy(allow_deep_link=False)

Bases: AuthenticatedStrategy

Source code in components/fr/public/auth/authorization.py 
27
28
def __init__(self, allow_deep_link: bool = False) -> None:
    super().__init__(allow_deep_link=allow_deep_link)

FrAuthenticatedWithCustomAuthorizationStrategy 

FrAuthenticatedWithCustomAuthorizationStrategy(
    allow_deep_link=False,
)

Bases: AuthenticatedWithCustomAuthorizationStrategy

Source code in components/fr/public/auth/authorization.py 
34
35
def __init__(self, allow_deep_link: bool = False) -> None:
    super().__init__(allow_deep_link=allow_deep_link)

FrOpenStrategy 

FrOpenStrategy(allow_deep_link=False)

Bases: OpenStrategy

Source code in components/fr/public/auth/authorization.py 
22
23
def __init__(self, allow_deep_link: bool = False) -> None:
    super().__init__(allow_deep_link=allow_deep_link)

FrOwnerOnlyStrategy 

FrOwnerOnlyStrategy(
    owner_bypass_permitted_for=None, allow_deep_link=False
)

Bases: OwnerOnlyStrategy

Source code in components/fr/public/auth/authorization.py 
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    owner_bypass_permitted_for: set[EmployeePermission] | None = None,
    allow_deep_link: bool = False,
) -> None:
    super().__init__(
        owner_bypass_permitted_for=owner_bypass_permitted_for,
        allow_deep_link=allow_deep_link,
    )

TokenExtractor

Bases: Protocol

__call__ 
__call__(token)
Source code in components/fr/public/auth/authorization.py 
56
def __call__(self, token: str) -> dict[str, Any]: ...  # noqa: D102

queries

mfa

is_mfa_required_for_account_admins 
is_mfa_required_for_account_admins(account_id)
Source code in components/fr/internal/auth/business_logic/queries/mfa.py 
50
51
52
53
54
55
56
57
58
59
def is_mfa_required_for_account_admins(account_id: str) -> bool:
    from components.fr.internal.business_logic.account.queries.account import (
        get_company_ids_from_account,
    )

    company_ids = get_company_ids_from_account(uuid.UUID(account_id))
    return any(
        is_mfa_required_for_company_admins(company_id=company_id)
        for company_id in company_ids
    )
is_mfa_required_for_company_admins 
is_mfa_required_for_company_admins(company_id)

This function checks if a french company forced MFA activation for its admins.

Source code in components/fr/internal/auth/business_logic/queries/mfa.py 
39
40
41
42
43
44
45
46
47
def is_mfa_required_for_company_admins(company_id: int) -> bool:
    """This function checks if a french company forced MFA activation for its admins."""
    from components.global_services.public.queries.mfa import (
        is_mfa_required_for_company_admins,
    )

    return is_mfa_required_for_company_admins(
        app_name=AppName.ALAN_FR, company_id=company_id
    )
is_mfa_required_for_company_employees 
is_mfa_required_for_company_employees(company_id)

This function checks if a french company forced MFA activation for its employees.

Source code in components/fr/internal/auth/business_logic/queries/mfa.py 
62
63
64
65
66
67
68
69
70
def is_mfa_required_for_company_employees(company_id: int) -> bool:
    """This function checks if a french company forced MFA activation for its employees."""
    from components.global_services.public.queries.mfa import (
        is_mfa_required_for_company_employees,
    )

    return is_mfa_required_for_company_employees(
        app_name=AppName.ALAN_FR, company_id=company_id
    )

components.fr.public.beneficiary

actions

add_beneficiary_form

create_or_update_add_beneficiary_form 
create_or_update_add_beneficiary_form(
    policy_id,
    enrollment_type,
    first_name,
    last_name,
    birth_date,
    ssn,
    settlement_iban,
    start_date,
    referent_ssns,
    lang,
)
Source code in components/fr/internal/beneficiary/actions/add_beneficiary_form.py 
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
def create_or_update_add_beneficiary_form(
    policy_id: int,
    enrollment_type: BeneficiaryEnrollmentType,
    first_name: str,
    last_name: str,
    birth_date: date,
    ssn: str | None,
    settlement_iban: str | None,
    start_date: date,
    referent_ssns: list[str] | None,
    lang: str | None,
) -> AddBeneficiaryFormEntity:
    check_can_be_added_as_beneficiary(
        policy_id=policy_id,
        first_name=first_name,
        birth_date=birth_date,
        ssn=ssn,
        start_date=start_date,
        enrollment_type=EnrollmentType(enrollment_type),
    )
    existing_add_beneficiary_form = get_last_add_beneficiary_form_from_type(
        policy_id=policy_id, enrollment_type=EnrollmentType(enrollment_type)
    )
    add_beneficiary_form_params = AddBeneficiaryFormParams(
        policy_id=policy_id,
        enrollment_type=enrollment_type,
        first_name=first_name,
        last_name=last_name,
        birth_date=birth_date,
        ssn=ssn,
        settlement_iban=settlement_iban,
        start_date=start_date,
        referent_ssns=referent_ssns,
        lang=lang,
    )

    if not existing_add_beneficiary_form:
        # if no existing add_beneficiary_form for the given enrollment_type, a new form is created
        add_beneficiary_form = _create_add_beneficiary_form(
            add_beneficiary_form_params=add_beneficiary_form_params
        )
    else:
        # if an existing add_beneficiary_form for the given enrollment_type, we need to check if a creation or an update needs to be performed
        add_beneficiary_form = _create_or_update_existing_add_beneficiary_form(
            existing_add_beneficiary_form=existing_add_beneficiary_form,
            add_beneficiary_form_params=add_beneficiary_form_params,
        )

    return AddBeneficiaryFormEntity.from_model(add_beneficiary_form)  # type: ignore[no-any-return]

basic_member_attributes

beneficiary_age module-attribute 

beneficiary_age = MemberAttributeDefinition[int](
    name="beneficiary_age",
    display_name="Beneficiary age",
    description="Age of the beneficiary in years",
    getter=_get_beneficiary_age,
    raw_type=int,
)

beneficiary_birth_date module-attribute 

beneficiary_birth_date = MemberAttributeDefinition[date](
    name="beneficiary_birth_date",
    display_name="Beneficiary birth date",
    description="Birth date of the beneficiary",
    getter=_get_beneficiary_birth_date,
    raw_type=date,
    formatter=_format_birth_date,
)

beneficiary_enrollment_type module-attribute 

beneficiary_enrollment_type = MemberAttributeDefinition[
    EnrollmentType
](
    name="beneficiary_enrollment_type",
    display_name="Beneficiary enrollment type",
    description="Enrollment type of the beneficiary (primary, partner, child)",
    getter=_get_beneficiary_enrollment_type,
    raw_type=EnrollmentType,
    formatter=_format_enrollment_type,
)

beneficiary_full_name module-attribute 

beneficiary_full_name = MemberAttributeDefinition[str](
    name="beneficiary_full_name",
    display_name="Beneficiary full name",
    description="Full name of the beneficiary",
    getter=_get_beneficiary_full_name,
    raw_type=str,
)

enrollment_end_date module-attribute 

enrollment_end_date = MemberAttributeDefinition[date](
    name="enrollment_end_date",
    display_name="Enrollment end date",
    description="End date of the enrollment",
    getter=_get_enrollment_end_date,
    raw_type=date,
)

enrollment_start_date module-attribute 

enrollment_start_date = MemberAttributeDefinition[date](
    name="enrollment_start_date",
    display_name="Enrollment start date",
    description="Start date of the enrollment",
    getter=_get_enrollment_start_date,
    raw_type=date,
)

enrollment_termination_type module-attribute 

enrollment_termination_type = MemberAttributeDefinition[
    EnrollmentTerminationType
](
    name="enrollment_termination_type",
    display_name="Enrollment termination type",
    description="Reason for enrollment termination",
    getter=_get_enrollment_termination_type,
    raw_type=EnrollmentTerminationType,
    formatter=_format_enrollment_termination_type,
)

member_attributes

beneficiary_eligibility_extension module-attribute 

beneficiary_eligibility_extension = MemberAttributeDefinition[
    BeneficiaryEligibilityExtension
](
    name="beneficiary_eligibility_extension",
    display_name="Beneficiary eligibility extension",
    description="Beneficiary eligibility extension for child enrollments",
    getter=_get_beneficiary_eligibility_extension,
    raw_type=BeneficiaryEligibilityExtension,
    formatter=_format_beneficiary_eligibility_extension,
)

contract_regime module-attribute 

contract_regime = MemberAttributeDefinition[ContractRegime](
    name="contract_regime",
    display_name="Contract regime",
    description="Contract regime (general or Alsace-Moselle)",
    getter=_get_contract_regime,
    raw_type=ContractRegime,
    formatter=_format_contract_regime,
)

is_alan_on_surco module-attribute 

is_alan_on_surco = MemberAttributeDefinition[bool](
    name="is_alan_on_surco",
    display_name="Is Alan on surco",
    description="Whether Alan is on surcomplementaire",
    getter=_get_is_alan_on_surco,
    raw_type=bool,
)

is_attestation_de_droits_uploaded module-attribute 

is_attestation_de_droits_uploaded = MemberAttributeDefinition[
    bool
](
    name="is_attestation_de_droits_uploaded",
    display_name="Is attestation de droits uploaded",
    description="Whether attestation de droits has been uploaded",
    getter=_get_is_attestation_de_droits_uploaded,
    raw_type=bool,
)

last_date_of_attestation_de_droits module-attribute 

last_date_of_attestation_de_droits = MemberAttributeDefinition[
    date
](
    name="last_date_of_attestation_de_droits",
    display_name="Last date of attestation de droits",
    description="Date of latest attestation de droits upload",
    getter=_get_last_date_of_attestation_de_droits,
    raw_type=date,
)

last_date_of_noemie_connection_request module-attribute 

last_date_of_noemie_connection_request = MemberAttributeDefinition[
    date
](
    name="last_date_of_noemie_connection_request",
    display_name="Last date of noemie connection request",
    description="Date of most recent noemie connection request",
    getter=_get_last_date_of_noemie_connection_request,
    raw_type=date,
)

reimbursement_iban module-attribute 

reimbursement_iban = MemberAttributeDefinition[str](
    name="reimbursement_iban",
    display_name="Reimbursement IBAN",
    description="IBAN used for reimbursement",
    getter=_get_reimbursement_iban,
    raw_type=str,
)

surcomplementaire_period module-attribute 

surcomplementaire_period = MemberAttributeDefinition[
    SurcomplementairePeriodInfo
](
    name="surcomplementaire_period",
    display_name="Surcomplementaire period",
    description="Surcomplementaire period info",
    getter=_get_surcomplementaire_period,
    raw_type=SurcomplementairePeriodInfo,
    formatter=_format_surcomplementaire_period,
)

components.fr.public.billing

business_logic

actions

Public billing actions for the FR component.

mark_invoices_as_sent 
mark_invoices_as_sent(invoice_ids, commit=False)

Query FR invoices by IDs and mark each as sent.

Source code in components/fr/public/billing/business_logic/actions.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def mark_invoices_as_sent(invoice_ids: list[int], commit: bool = False) -> None:
    """Query FR invoices by IDs and mark each as sent."""
    from sqlalchemy import select

    from components.fr.internal.billing.models.invoice import Invoice  # noqa: ALN043

    invoices = (
        current_session.execute(select(Invoice).where(Invoice.id.in_(invoice_ids)))
        .scalars()
        .all()
    )
    for invoice in invoices:
        mark_invoice_as_sent(invoice)
    if commit:
        current_session.commit()

queries

get_employee_debt_status 
get_employee_debt_status(user_id)
Source code in components/fr/internal/billing/business_logic/queries/employee_contract.py 
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
def get_employee_debt_status(user_id: int) -> DebtStatus | None:
    check_resource_exists_or_raise(User, user_id)
    ani_contracts = PolicyAniContractBroker.get_all_uncancelled_for_user_id(
        user_id=user_id,
    )

    all_non_cancelled_employee_contracts = (
        get_all_non_cancelled_option_contract(user_id)
        + get_all_non_cancelled_unpaid_leave_contracts(user_id)
        + get_all_non_cancelled_direct_billing_contract(user_id)
        + ani_contracts
    )
    # Sort by start date, most recent first
    all_non_cancelled_employee_contracts.sort(
        key=attrgetter("start_date"), reverse=True
    )
    for contract in all_non_cancelled_employee_contracts:
        balance = get_balance_for_recovery(
            contract_identifier=contract.contract_identifier
        )
        if balance > 0:
            _, _, unpaid_invoices = get_latest_unpaid_period(
                contract_identifier=contract.contract_identifier
            )
            return DebtStatus(
                contract_type=contract.contract_identifier.contract_type,
                contract_id=contract.contract_identifier.contract_id,
                balance=balance,
                invoices=(
                    [invoice.to_dict() for invoice in unpaid_invoices]  # type: ignore[alan-deprecated-serialisation]
                    if unpaid_invoices
                    else None
                ),
            )
    return None

member_attributes

billing_balance module-attribute 

billing_balance = MemberAttributeDefinition[int](
    name="billing_balance",
    display_name="Billing balance",
    description="Contract billing balance in euro cents",
    getter=_get_contract_billing_balance_in_euro_cents,
    raw_type=int,
    formatter=_format_billing_balance,
)

billing_information module-attribute 

billing_information = MemberAttributeDefinition(
    name="billing_information",
    display_name="Billing information",
    description="Detailed billing information including invoices and payments",
    getter=_get_billing_information,
    raw_type=dict,
    formatter=_format_billing_information,
)

components.fr.public.blueprint

fr module-attribute 

fr = create_blueprint(
    "fr", import_name=__name__, template_folder="templates"
)

components.fr.public.claim_management

api

schedule_update_insurance_profile_informations_cache 

schedule_update_insurance_profile_informations_cache(
    insurance_profile_id=None,
    insurance_profile_ids=None,
    priority=None,
)

Schedule the update of the insurance profile information cache for the given insurance profile

Source code in components/fr/public/claim_management/api.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def schedule_update_insurance_profile_informations_cache(
    insurance_profile_id: int | None = None,  # DEPRECATED, use insurance_profile_ids
    insurance_profile_ids: Iterable[int] | None = None,
    priority: JobPriority | None = None,
) -> None:
    """
    Schedule the update of the insurance profile information cache for the given insurance profile
    """
    from components.fr.internal.claim_management.public.insurance_profile_informations_cache.api import (
        schedule_update_insurance_profile_informations_caches,
    )

    # backward compatibility
    if insurance_profile_ids is None and insurance_profile_id is not None:
        insurance_profile_ids = [insurance_profile_id]

    return schedule_update_insurance_profile_informations_caches(
        insurance_profile_ids=mandatory(insurance_profile_ids),
        priority=priority,
    )

upload_alan_therapy_invoice_as_insurance_document 

upload_alan_therapy_invoice_as_insurance_document(
    file,
    start_date,
    end_date,
    operator_comment,
    user_id,
    paid_amount,
    executant_number,
    health_professional,
    short_code,
    excluded_company_ids=None,
)

Upload internally an insurance document for Alan Therapy

Source code in components/fr/public/claim_management/api.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
 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
123
124
125
126
127
128
129
def upload_alan_therapy_invoice_as_insurance_document(
    file: IO,  # type: ignore[type-arg]
    start_date: date,
    end_date: date,
    operator_comment: str,
    user_id: int,
    paid_amount: float | None,
    executant_number: str,
    health_professional: str,
    short_code: str,
    excluded_company_ids: set[int] | None = None,
) -> bool:
    """
    Upload internally an insurance document for Alan Therapy
    """
    from components.fr.internal.claim_management.claim_engine.steps.parsing.entities.in_memory_parsed_document_content import (  # noqa: ALN039
        InMemoryParsedDocumentContent,
        InvoiceCareAct,
        InvoiceContent,
        InvoiceOverallInformation,
    )
    from components.fr.internal.claim_management.internal.misc.business_logic.remaining_usage import (  # noqa: ALN039
        get_health_coverage_on_date,
        get_remaining_usages,
    )
    from components.fr.internal.models.enums.insurance_document_type import (
        ClaimInsuranceDocumentCategory,
    )
    from components.fr.internal.models.enums.practitioner import (
        Practitioner,
    )
    from components.fr.internal.models.user import User
    from shared.helpers.db import current_session

    user = current_session.get(User, user_id)
    if not user:
        raise Exception("User is unknown")

    insurance_profile = user.insurance_profile
    if not insurance_profile:
        raise Exception("Insurance Profile is unknown")

    policy = insurance_profile.policy_on(start_date)
    if not policy:
        raise Exception("Policy is unknown")

    company_id = policy.contract.company_id
    if excluded_company_ids and company_id in excluded_company_ids:
        raise Exception("Company is excluded")

    # Checking coverage and remaining usage

    health_coverage = get_health_coverage_on_date(
        insurance_profile=insurance_profile, reference_date=start_date
    )
    if not health_coverage:
        raise Exception("No health coverage")

    mapping = next(
        (
            m
            for m in health_coverage.internal_care_type_guarantee_mappings
            if m.internal_care_type.short_code == short_code
        ),
        None,
    )
    if not mapping:
        raise Exception("No internal care type mapping")

    remaining_usages = get_remaining_usages(
        insurance_profile=insurance_profile,
        reference_date=start_date,
        internal_health_guarantees_to_consider={mapping.internal_health_guarantee},
    )
    if len(remaining_usages) != 1:
        raise Exception("No remaining usages")

    count_remaining = remaining_usages[0].count_remaining or 0
    if count_remaining < 1:
        raise Exception("No remaining count")

    # Uploading invoice as insurance document
    pre_parsed_content = InMemoryParsedDocumentContent(
        content=InvoiceContent(
            care_acts=[
                InvoiceCareAct(
                    care_code=short_code,
                    start_date=start_date,
                    end_date=end_date,
                    total_spend=paid_amount or 0,
                    tp_amount=0,
                    brss=0,
                    secu_reimbursed_amount=0,
                    ss_coverage_rate=0,
                )
            ],
            overall_information=InvoiceOverallInformation(
                beneficiary=user.insurance_profile.id,  # type: ignore[arg-type,union-attr]
                executant_number=executant_number,
                health_professional=health_professional,
                practitioner=Practitioner.other,
            ),
        ),
        context=ParsingContext().with_no_internal_control(),
        document_category=ClaimInsuranceDocumentCategory.invoice,
    )
    upload_insurance_document_internally(
        file=file,
        user_id=user.id,
        policy_id=policy.id,
        operator_comment=operator_comment,
        pre_parsed_content=pre_parsed_content,
    )
    current_session.commit()
    return True

are_policy_reimbursements_blocked

are_policy_reimbursements_blocked_at_care_date 

are_policy_reimbursements_blocked_at_care_date(
    insurance_profile_id, care_date
)
Source code in components/fr/internal/business_logic/insurance_profile.py 
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
def are_policy_reimbursements_blocked_at_care_date(
    insurance_profile_id: int, care_date: date
) -> bool:
    insurance_profile = get_or_raise_missing_resource(
        InsuranceProfile, insurance_profile_id
    )
    # When we block reimbursements, we want them to be still blocked if the care date happened after the policy
    policy = insurance_profile.policy_on(care_date) or insurance_profile.last_policy

    # We first check for any KYC blocker from the new table
    if policy:
        siren = get_siren_for_policy(policy)
        if siren is None:
            current_logger.warning(
                f"No siren found for policy {policy.id}, this should not happen"
            )
        if is_siren_blocked_at_date(siren, care_date):  # type: ignore[arg-type]
            current_logger.warning(
                f"Active blocker found for siren {siren} on policy {policy.id} at date {care_date}, blocking reimbursement"
            )
            return True

    # If no new KYC blocker, we still apply the legacy check
    return (
        policy is not None
        and care_date is not None
        and policy.block_reimbursements_from is not None
        and care_date >= policy.block_reimbursements_from
    )

member_attributes

care_event_details_from_entry_point module-attribute 

care_event_details_from_entry_point = MemberAttributeDefinition[
    CareEventDetailsForSupportAiAssistant
](
    name="care_event_details_from_entry_point",
    display_name="Care event details from entry point",
    description="Detailed care event info from the conversation entry point",
    getter=_get_care_event_details,
    raw_type=CareEventDetailsForSupportAiAssistant,
    formatter=_format_care_event_details,
)

contact_lenses_forfait_remaining_and_limit module-attribute 

contact_lenses_forfait_remaining_and_limit = MemberAttributeDefinition[
    dict[str, RemainingAmountUsage | RemainingCountUsage]
](
    name="contact_lenses_forfait_remaining_and_limit",
    display_name="Contact lenses forfait remaining and limit",
    description="Remaining contact lenses forfait usage and limit",
    getter=_get_contact_lenses_forfait_remaining_and_limit,
    raw_type=dict,
    formatter=_format_remaining_usage_contact_lenses,
)

contact_lenses_pec_requests module-attribute 

contact_lenses_pec_requests = MemberAttributeDefinition[
    list[PecRequestAttribute]
](
    name="contact_lenses_pec_requests",
    display_name="Contact lenses PEC requests",
    description="List of contact lenses tiers-payant PEC requests",
    getter=_get_contact_lenses_pec_requests,
    raw_type=list,
    formatter=_format_contact_lenses_pec_requests,
)

contact_lenses_prescription_dates module-attribute 

contact_lenses_prescription_dates = MemberAttributeDefinition[
    list[PrescriptionAttribute]
](
    name="contact_lenses_prescription_dates",
    display_name="Contact lenses prescription dates",
    description="List of contact lenses prescription attributes",
    getter=_get_contact_lenses_prescription_dates,
    raw_type=list,
    formatter=_format_prescription_attributes,
)

format_pec_request 

format_pec_request(pec_request, on_date)

Format a PEC request for display.

Source code in components/fr/public/claim_management/member_attributes.py 
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
def format_pec_request(pec_request: PecRequestAttribute, on_date: date) -> str:
    """Format a PEC request for display."""
    from components.member_attributes.public.formatters import (
        format_date_with_relative_time,
    )

    pec_id_part = f" {pec_request.pec_id}" if pec_request.pec_id else ""
    request_type_part = (
        f" ({pec_request.request_type.value.upper()})"
        if pec_request.request_type
        else ""
    )
    result = (
        f"PEC{pec_id_part}{request_type_part} {pec_request.status} on "
        f"{format_date_with_relative_time(pec_request.pec_date, today=on_date)}"
    )
    if pec_request.health_professional_name:
        result += f" at {pec_request.health_professional_name}"
    if pec_request.is_fraud_confirmed:
        result += " (rejection reason: Almerys needs investigation)"
    elif pec_request.rejection_reason_code:
        rejection_reason = _PEC_RESPONSE_REJECTION_CODE_TO_LABEL.get(
            pec_request.rejection_reason_code, "unknown"
        )
        result += f" (rejection reason: {rejection_reason})"
    return result

format_remaining_usage 

format_remaining_usage(remaining_usages, unit='')

Format remaining usages dict — single entry returns string, multiple returns dict.

Source code in components/fr/public/claim_management/member_attributes.py 
438
439
440
441
442
443
444
445
446
447
448
449
450
451
def format_remaining_usage(
    remaining_usages: dict[str, RemainingAmountUsage | RemainingCountUsage],
    unit: str = "",
) -> dict[str, str] | str:
    """Format remaining usages dict — single entry returns string, multiple returns dict."""
    if len(remaining_usages.keys()) == 1:
        return _format_single_remaining_usage(
            remaining_usages[list(remaining_usages.keys())[0]], unit
        )
    else:
        return {
            code: _format_single_remaining_usage(remaining_usage, unit)
            for code, remaining_usage in remaining_usages.items()
        }

glasses_forfait_remaining_and_limit module-attribute 

glasses_forfait_remaining_and_limit = MemberAttributeDefinition[
    dict[str, RemainingAmountUsage | RemainingCountUsage]
](
    name="glasses_forfait_remaining_and_limit",
    display_name="Glasses forfait remaining and limit",
    description="Remaining glasses forfait usage and limit",
    getter=_get_glasses_forfait_remaining_and_limit,
    raw_type=dict,
    formatter=_format_remaining_usage_glasses,
)

glasses_pec_requests module-attribute 

glasses_pec_requests = MemberAttributeDefinition[
    list[PecRequestAttribute]
](
    name="glasses_pec_requests",
    display_name="Glasses PEC requests",
    description="List of glasses tiers-payant PEC requests",
    getter=_get_glasses_pec_requests,
    raw_type=list,
    formatter=_format_glasses_pec_requests,
)

glasses_prescription_dates module-attribute 

glasses_prescription_dates = MemberAttributeDefinition[
    list[PrescriptionAttribute]
](
    name="glasses_prescription_dates",
    display_name="Glasses prescription dates",
    description="List of glasses prescription attributes",
    getter=_get_glasses_prescription_dates,
    raw_type=list,
    formatter=_format_prescription_attributes,
)

insurance_document_details_from_entry_point module-attribute 

insurance_document_details_from_entry_point = MemberAttributeDefinition[
    ParsedDocumentContentForSupportAiAssistant
](
    name="insurance_document_details_from_entry_point",
    display_name="Insurance document details from entry point",
    description="Parsed insurance document details from the conversation entry point",
    getter=_get_insurance_document_details,
    raw_type=ParsedDocumentContentForSupportAiAssistant,
    formatter=_format_insurance_document_details,
)

is_enterprise_overconsumption_pilot_member module-attribute 

is_enterprise_overconsumption_pilot_member = MemberAttributeDefinition[
    bool
](
    name="is_enterprise_overconsumption_pilot_member",
    display_name="Is enterprise overconsumption pilot member",
    description="Whether the member is part of the enterprise overconsumption pilot program",
    getter=_get_is_enterprise_overconsumption_pilot_member,
    raw_type=bool,
)

last_care_date module-attribute 

last_care_date = MemberAttributeDefinition[date](
    name="last_care_date",
    display_name="Last care date",
    description="Date of the most recent care event",
    getter=_get_last_care_date,
    raw_type=date,
)

last_date_of_contact_lenses_purchase module-attribute 

last_date_of_contact_lenses_purchase = MemberAttributeDefinition[
    date
](
    name="last_date_of_contact_lenses_purchase",
    display_name="Last date of contact lenses purchase",
    description="Date of last contact lenses purchase via TP",
    getter=_get_last_date_of_contact_lenses_purchase,
    raw_type=date,
)

last_date_of_glasses_purchase module-attribute 

last_date_of_glasses_purchase = MemberAttributeDefinition[
    date
](
    name="last_date_of_glasses_purchase",
    display_name="Last date of glasses purchase",
    description="Date of last glasses purchase via TP",
    getter=_get_last_date_of_glasses_purchase,
    raw_type=date,
)

previous_glasses_purchase_date_from_previous_insurer module-attribute 

previous_glasses_purchase_date_from_previous_insurer = MemberAttributeDefinition[
    date
](
    name="previous_glasses_purchase_date_from_previous_insurer",
    display_name="Previous glasses purchase date from previous insurer",
    description="Date of last glasses purchase recorded from previous insurer form",
    getter=_get_previous_glasses_purchase_date_from_previous_insurer,
    raw_type=date,
)

quote_document_details_from_entry_point module-attribute 

quote_document_details_from_entry_point = MemberAttributeDefinition[
    QuoteDocumentInfo
](
    name="quote_document_details_from_entry_point",
    display_name="Quote document details from entry point",
    description="Quote document details from the conversation entry point",
    getter=_get_quote_document_details,
    raw_type=QuoteDocumentInfo,
)

teletransmission_status module-attribute 

teletransmission_status = MemberAttributeDefinition[
    TeletransmissionStatusInfo
    | list[TeletransmissionStatusInfo]
](
    name="teletransmission_status",
    display_name="Teletransmission status",
    description="Teletransmission status for the enrollment",
    getter=_get_teletransmission_status,
    raw_type=TeletransmissionStatusInfo,
    formatter=_format_teletransmission_status,
)

components.fr.public.clinic

adapter

FrClinicAdapter

Bases: ClinicAdapter

clinic_consent_ai_publish_date = datetime(2023, 12, 6)
create_external_user 
create_external_user(
    onboarding_data, profile_service, authentication_service
)

Create an external teleconsultation user profile with onboarding data for FR.

Source code in components/fr/public/clinic/adapter.py 
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
@inject_profile_service
@inject_authentication_service
def create_external_user(
    self,
    onboarding_data: ExternalOnboardingUserData,
    profile_service: ProfileService,
    authentication_service: AuthenticationService,
) -> tuple[User, RefreshTokenType]:
    """Create an external teleconsultation user profile with onboarding data for FR."""
    # Validate and parse gender
    if onboarding_data.gender_str not in ["male", "female"]:
        raise BaseErrorCode.invalid_arguments(
            description=f"Invalid gender value: {onboarding_data.gender_str}. Must be 'male' or 'female'"
        )
    gender = (
        UserGender.male
        if onboarding_data.gender_str == "male"
        else UserGender.female
    )

    # Validate SSN
    ssn = SSNValidator.validates_ssn(onboarding_data.social_security_number)

    # Create user profile with all available fields
    user = create_or_assign_profile_with_authenticatable_user(
        first_name=onboarding_data.first_name,
        last_name=onboarding_data.last_name,
        birth_date=onboarding_data.date_of_birth,
        language=Lang.french,
        phone_number=onboarding_data.phone_number,
        gender=gender,
        prehashed_password=onboarding_data.prehashed_password,
        password=onboarding_data.password,
        email=onboarding_data.email,
        email_verified=False,
    )
    set_ssn_ntt_on_user(int(user.id), ssn)

    current_session.add(user)
    current_session.commit()

    # Send verification email if client_id provided
    if onboarding_data.client_id:
        identity = authentication_service.get_keycloak_identity_by_profile_id(
            user.profile_id
        )
        if identity:
            redirect_uri = f"{current_config['FRONT_END_BASE_URL']}/stlc-booking"
            authentication_service.send_verification_email(
                identity_id=identity.id,
                client_id=onboarding_data.client_id,
                redirect_uri=redirect_uri,
            )
        else:
            # Identity should always exist after create_or_assign_profile_with_authenticatable_user
            # If missing, log error - user will be stuck on verification screen
            current_logger.error(
                "Failed to send verification email: identity not found for external user",
                user_id=str(user.id),
                profile_id=str(user.profile_id),
            )

    # Set place of birth (mandatory field)
    place_of_birth = PlaceOfBirth(
        name=f"{onboarding_data.place_of_birth_name} {onboarding_data.place_of_birth_postal_code}",
    )
    profile_service.edit_birth_information(
        profile_id=user.profile_id,
        place_of_birth=place_of_birth,
    )

    return user, onboarding_data.refresh_token_type
get_app_base_user_data 
get_app_base_user_data(app_user_id)
Source code in components/fr/public/clinic/adapter.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def get_app_base_user_data(self, app_user_id: str) -> BaseUserData:  # noqa: D102
    from components.fr.internal.models.user import User
    from shared.helpers.string import normalize_name

    result = (
        current_session.query(User.first_name, User.last_name)  # type: ignore[call-overload] # noqa: ALN085
        .filter(User.id == app_user_id)
        .one_or_none()
    )

    if not result:
        raise BaseErrorCode.missing_resource(message="User not found")

    first_name, last_name = result

    return BaseUserData(
        first_name=normalize_name(first_name),
        last_name=normalize_name(last_name),
    )
get_app_user_available_health_services 
get_app_user_available_health_services(
    profile_service, app_user_id
)
Source code in components/fr/public/clinic/adapter.py 
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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
@inject_profile_service
def get_app_user_available_health_services(  # noqa: D102
    self, profile_service: ProfileService, app_user_id: str
) -> list[AvailableHealthService]:
    from components.fr.internal.models.user import User
    from components.fr.public.clinic.internal_teleconsultation import (
        app_user_has_access_to_internal_teleconsultation,
    )
    from components.fr.public.clinic.livi.queries import (
        is_user_eligible_for_telemedicine_with_livi,
    )

    fr_user = get_or_raise_missing_resource(User, app_user_id)
    user_profile = profile_service.get_profile(fr_user.profile_id)

    user_id = int(app_user_id)

    # All FR users have access to therapy sessions and orientation calls (even if they are not included) and the DATO content library
    available_health_services = [
        AvailableHealthService(
            name=AvailableHealthServiceName.THERAPY_SESSION,
        ),
        AvailableHealthService(
            name=AvailableHealthServiceName.ORIENTATION_CALL,
        ),
        AvailableHealthService(name=AvailableHealthServiceName.DATO_CONTENT),
    ]

    if is_user_eligible_for_telemedicine_with_livi(user_id=user_id):
        available_health_services.append(
            AvailableHealthService(
                name=AvailableHealthServiceName.VIDEO_CONSULTATION_LIVI,
            ),
        )

    if app_user_has_access_to_internal_teleconsultation(fr_user):
        available_health_services.append(
            AvailableHealthService(
                name=AvailableHealthServiceName.VIDEO_CONSULTATION,
                is_recommended=True,
                has_upcoming_availability=app_user_has_upcoming_availability_for_alan_consultations(
                    member_app_user=FeatureUser(
                        app_id=AppName.ALAN_FR,
                        app_user_id=app_user_id,
                    ),
                    availability_threshold_in_hours=24,
                    locale=user_profile.preferred_language
                    if user_profile
                    else "fr",
                ),
            )
        )

    available_health_services.append(
        AvailableHealthService(name=AvailableHealthServiceName.HEALTH_PROGRAM)
    )

    return available_health_services
get_app_user_data 
get_app_user_data(
    app_user_id, compute_key_account_info=False
)
Source code in components/fr/public/clinic/adapter.py 
 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
123
124
125
126
127
128
129
130
131
132
def get_app_user_data(  # noqa: D102
    self, app_user_id: str, compute_key_account_info: bool = False
) -> UserData:
    from components.fr.public.user.queries import get_user_for_clinic

    user = get_user_for_clinic(int(app_user_id))

    is_key_account_or_large_company_and_not_alaner = False
    if compute_key_account_info:
        is_key_account_or_large_company_and_not_alaner = not user.is_alaner and any(
            user_company.key_account or user_company.n_employees > 100
            for user_company in user.companies
        )

    dependents = (
        [
            Dependent(
                app_user_id=str(insurance_profile.user.id),
                first_name=insurance_profile.user.normalized_first_name,
                last_name=insurance_profile.user.normalized_last_name,
                age=int(insurance_profile.user.age)
                if insurance_profile.user.age
                else None,
                gender=insurance_profile.user.guess_gender,
                birth_date=insurance_profile.user.birth_date,
                dependent_type=(
                    DependentType.PARTNER
                    if insurance_profile.current_enrollment_type
                    == EnrollmentType.partner
                    else DependentType.CHILD
                    if insurance_profile.current_enrollment_type
                    == EnrollmentType.child
                    else None
                ),
            )
            for insurance_profile in user.insurance_profile.get_current_policy_insurance_profiles()
            if insurance_profile.user != user
        ]
        if user.insurance_profile is not None
        else []
    )

    return UserData(
        first_name=user.normalized_first_name,
        last_name=user.normalized_last_name,
        gender=user.guess_gender,
        email=user.email,
        profile_id=user.profile_id,
        birth_date=user.birth_date,
        phone=user.phone,
        country=user.address.country if user.address else None,
        address=user.address.as_one_line if user.address else None,
        ssn=user.insurance_profile.ssn if user.insurance_profile else None,
        lang=user.lang,
        is_key_account_or_large_company_and_not_alaner=is_key_account_or_large_company_and_not_alaner,
        is_alaner=user.is_alaner,
        dependents=dependents,
    )
get_booking_session_package 
get_booking_session_package(app_user_id, session_type)
Source code in components/fr/public/clinic/adapter.py 
376
377
378
379
380
381
382
383
384
385
386
387
def get_booking_session_package(  # noqa: D102
    self,
    app_user_id: str,
    session_type: TherapistBookingSessionType,
) -> BookingSessionPackage | None:
    from components.fr.public.clinic.guarantee import (
        get_therapy_sessions_guarantee_package,
    )

    if session_type == TherapistBookingSessionType.therapy:
        return get_therapy_sessions_guarantee_package(app_user_id=int(app_user_id))
    return None
get_coverage_status 
get_coverage_status(app_user_id)
Source code in components/fr/public/clinic/adapter.py 
253
254
255
256
257
258
def get_coverage_status(self, app_user_id: str) -> CoverageStatus | None:  # noqa: D102
    from components.fr.public.clinic.clinic_eligibility import (
        get_coverage_status,
    )

    return get_coverage_status(user_id=int(app_user_id))
get_inactive_profile_ids_since 
get_inactive_profile_ids_since(period_in_years)

Return profile IDs inactive for the given period in FR.

Source code in components/fr/public/clinic/adapter.py 
486
487
488
489
490
491
492
def get_inactive_profile_ids_since(self, period_in_years: int) -> list[UUID]:
    """Return profile IDs inactive for the given period in FR."""
    from components.fr.public.data_retention.queries import (
        get_inactive_profile_ids_since,
    )

    return get_inactive_profile_ids_since(inactive_number_of_years=period_in_years)
get_inactive_user_ids_since 
get_inactive_user_ids_since(period_in_years)

Return app user IDs inactive for the given period in FR.

Source code in components/fr/public/clinic/adapter.py 
494
495
496
497
498
499
500
def get_inactive_user_ids_since(self, period_in_years: int) -> list[str]:
    """Return app user IDs inactive for the given period in FR."""
    from components.fr.public.data_retention.queries import (
        get_inactive_user_ids_since,
    )

    return get_inactive_user_ids_since(inactive_number_of_years=period_in_years)
get_last_active_id_verification_request_for_user 
get_last_active_id_verification_request_for_user(
    app_user_id,
)

FR implementation of getting the last active ID verification request for a user.

Source code in components/fr/public/clinic/adapter.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
def get_last_active_id_verification_request_for_user(
    self, app_user_id: str
) -> IDVerificationRequest | None:
    """
    FR implementation of getting the last active ID verification request for a user.
    """
    from components.id_verification.public.business_logic.queries.id_verification import (
        get_last_active_id_verification_request_for_user,
    )

    return get_last_active_id_verification_request_for_user(
        user_id=int(app_user_id)
    )
get_orientation_survey_url 
get_orientation_survey_url(lang)
Source code in components/fr/public/clinic/adapter.py 
72
73
def get_orientation_survey_url(self, lang: str) -> str:  # noqa: D102
    raise NotImplementedError("FR doesn't support orientation surveys.")
has_access_to_orientation_call 
has_access_to_orientation_call(_)
Source code in components/fr/public/clinic/adapter.py 
178
179
180
181
182
def has_access_to_orientation_call(  # noqa: D102
    self, _: str
) -> bool:
    # In France, all users have access to 1 orientation call, even though they are not required to book it
    return True
has_app_user_permission 
has_app_user_permission(app_user_id, permission)
Source code in components/fr/public/clinic/adapter.py 
154
155
156
157
158
159
160
def has_app_user_permission(  # noqa: D102
    self, app_user_id: str, permission: EmployeePermission
) -> bool:
    from components.fr.internal.models.user import User

    user: User = get_or_raise_missing_resource(User, app_user_id)
    return has_permission(user, permission)
is_app_user_admin_of_company 
is_app_user_admin_of_company(app_user_id, app_company_id)
Source code in components/fr/public/clinic/adapter.py 
243
244
245
246
247
248
249
250
251
def is_app_user_admin_of_company(  # noqa: D102
    self, app_user_id: str, app_company_id: str
) -> bool:
    from components.fr.internal.models.user import User

    user: User = get_or_raise_missing_resource(User, app_user_id)
    return app_company_id in [
        str(admined_company.id) for admined_company in user.admined_companies
    ]
is_orientation_session_mandatory 
is_orientation_session_mandatory()
Source code in components/fr/public/clinic/adapter.py 
404
405
406
407
def is_orientation_session_mandatory(self) -> bool:  # noqa: D102
    # In France, orientation sessions are not mandatory by default
    # This can be customized based on specific business requirements
    return False
release_date_of_conversations_created_for_therapy_sessions class-attribute instance-attribute 
release_date_of_conversations_created_for_therapy_sessions = datetime(
    2025, 2, 17
)
request_id_verification_request_for_user 
request_id_verification_request_for_user(
    app_user_id, user_info, commit=True
)

FR implementation of getting or requesting ID verification for a user.

Source code in components/fr/public/clinic/adapter.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
def request_id_verification_request_for_user(
    self,
    app_user_id: str,
    user_info: ClinicUserDataForIdVerification,
    commit: bool = True,
) -> IDVerificationRequest:
    """
    FR implementation of getting or requesting ID verification for a user.
    """
    from components.id_verification.public.business_logic.queries.id_verification import (
        get_or_request_id_verification,
    )
    from components.id_verification.public.entities.id_verification_request import (
        IDVerificationReason,
    )

    return get_or_request_id_verification(
        user_id=int(app_user_id),
        company_id=None,
        reason=IDVerificationReason.clinic_teleconsultation,
        user_info=convert_user_data_id_verification_request_user_info(user_info),
        commit=commit,
    )
should_request_id_verification_for_user 
should_request_id_verification_for_user(app_user_id)

FR implementation of checking if ID verification should be requested for a user. We request ID verification for internal consultation if: - The feature flag "request_id_verification_for_internal_consultation" is enabled for the user - The user does not have an active ID verification request - The user's identity is not validated

Source code in components/fr/public/clinic/adapter.py 
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
def should_request_id_verification_for_user(
    self,
    app_user_id: str,
) -> bool:
    """
    FR implementation of checking if ID verification should be requested for a user.
    We request ID verification for internal consultation if:
    - The feature flag "request_id_verification_for_internal_consultation" is enabled for the user
    - The user does not have an active ID verification request
    - The user's identity is not validated
    """
    from components.clinic.public.business_logic.insi_identity import (
        is_identity_verified_for_user,
    )
    from components.id_verification.public.business_logic.queries.id_verification import (
        get_last_active_id_verification_request_for_user,
    )

    id_verification_request = get_last_active_id_verification_request_for_user(
        user_id=int(app_user_id)
    )

    return (
        # Kill switch of the whole automated id verification flow
        # (as per requirement 37)
        bool_feature_flag(
            feature_flag_key="killswitch-id-verification-for-clinic-teleconsultation",
            default_value=True,
        )
        and bool_feature_flag(
            feature_flag_key="request-id-verification-for-internal-consultation",
            default_value=False,
            context_data=get_user_context_data(app_user_id, {}),
        )
        and id_verification_request is None
        and not is_identity_verified_for_user(
            app_user=FeatureUser(app_user_id=app_user_id, app_id=AppName.ALAN_FR)
        )
    )
therapy_refundable_cancellation_limit_in_hours 
therapy_refundable_cancellation_limit_in_hours()
Source code in components/fr/public/clinic/adapter.py 
409
410
411
def therapy_refundable_cancellation_limit_in_hours(self) -> int | None:  # noqa: D102
    # 48 hours is the cancellation limit for therapy sessions in FR
    return 48
update_app_user_phone 
update_app_user_phone(profile_service, app_user_id, phone)
Source code in components/fr/public/clinic/adapter.py 
162
163
164
165
166
167
168
169
170
@inject_profile_service
def update_app_user_phone(  # noqa: D102
    self, profile_service: ProfileService, app_user_id: str, phone: str | None
) -> None:
    from components.fr.internal.models.user import User

    fr_user = get_or_raise_missing_resource(User, app_user_id)
    profile_service.change_phone_number(fr_user.profile_id, phone_number=phone)
    current_session.commit()
update_app_user_ssn 
update_app_user_ssn(app_user_id, ssn, commit=False)
Source code in components/fr/public/clinic/adapter.py 
172
173
174
175
176
def update_app_user_ssn(  # noqa: D102
    self, app_user_id: str, ssn: str | None, commit: bool = False
) -> None:
    # No SSN update implementation for FR
    pass
upload_invoice_as_insurance_document 
upload_invoice_as_insurance_document(
    file, app_user_id, upload_invoice_data
)
Source code in components/fr/public/clinic/adapter.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
def upload_invoice_as_insurance_document(  # noqa: D102
    self,
    file: IO,  # type: ignore[type-arg]
    app_user_id: str,
    upload_invoice_data: UploadInvoiceData,
) -> bool:
    from components.fr.public.claim_management.api import (
        upload_alan_therapy_invoice_as_insurance_document,
    )

    try:
        current_logger.info(
            "Checking if session invoice can be registered as insurance document",
            session_id=upload_invoice_data.session_id,
        )

        return upload_alan_therapy_invoice_as_insurance_document(
            file=file,
            operator_comment=f"This is an Alan Therapist Booking Session invoice ({upload_invoice_data.session_id}). It is not supposed to be reparsed by Tessi.",
            start_date=upload_invoice_data.starts_at,
            end_date=upload_invoice_data.ends_at,
            user_id=int(app_user_id),
            paid_amount=upload_invoice_data.paid_amount,
            short_code="MED_DOUCE_PSYCH",
            executant_number=(
                upload_invoice_data.medical_admin_identifier.replace(" ", "")
                if upload_invoice_data.medical_admin_identifier
                else ""
            ),
            health_professional=upload_invoice_data.medical_admin_name,
        )
    except Exception as e:
        current_logger.info(
            f"Cannot register therapist session invoice as insurance document: {e}",
            session_id=upload_invoice_data.session_id,
        )
        return False
user_has_24_hour_response_guarantee 
user_has_24_hour_response_guarantee(app_user_id)
Source code in components/fr/public/clinic/adapter.py 
389
390
391
392
393
394
395
def user_has_24_hour_response_guarantee(  # noqa: D102
    self,
    app_user_id: str,
) -> bool:
    from components.fr.internal.models.queries.user import is_user_civil_servant

    return is_user_civil_servant(user_id=int(app_user_id))
validate_session_duration 
validate_session_duration(session_duration)
Source code in components/fr/public/clinic/adapter.py 
397
398
399
400
401
402
def validate_session_duration(  # noqa: D102
    self,
    session_duration: int,
) -> None:
    # No banned session durations in FR
    pass

clinic_adapter module-attribute 

clinic_adapter = FrClinicAdapter()

clinic_eligibility

This module contains the query to get the current or upcoming period of eligibility to the clinic restricted services.

NOTE: the logic could be reused by other services than the Clinic, provided the country-specific rules are the same. If yes, feel free to rename the file and query to a more generic name.

get_coverage_status 

get_coverage_status(user_id)

Return the start and optionally the end date of the current or upcoming period of eligibility to the clinic restricted services.

Source code in components/fr/public/clinic/clinic_eligibility.py 
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
def get_coverage_status(user_id: int) -> CoverageStatus | None:
    """
    Return the start and optionally the end date of the current or upcoming period of eligibility to the clinic restricted services.
    """
    from components.fr.internal.business_logic.user.queries.user_states import (
        get_user_states_ever_active_between,
    )

    user_states = get_user_states_ever_active_between(
        user_id=user_id,
        period_start=utctoday(),
        period_end=None,
    )

    # Sort by earliest start date first
    user_states.sort(
        key=lambda state: state.start_date if state.start_date else date.max
    )

    current_logger.debug(
        f"Clinic restricted service eligibility: {len(user_states)} state(s) found.",
        user_id=user_id,
        state=[state.to_dict() for state in user_states],
    )

    for state in user_states:
        from components.fr.internal.business_logic.user.data.user_state import (
            UserStateInsured,
        )

        # This de facto excludes states like UserStateInvited/UserStateInvitedRetiree, UserStateNotInsuredUnpaidLeave or UserStateExempted
        if isinstance(state, UserStateInsured):
            current_logger.debug(
                f"Clinic restricted service eligibility: {state} is eligible.",
                user_id=user_id,
                state=state.to_dict(),
            )

            return CoverageStatus(
                # Assumption: UserStateInsured always has a start date
                start_date=mandatory(
                    state.start_date,
                    "Found an unexpected user state with no start date.",
                ),
                end_date=state.end_date,
            )

    return None

guarantee

THERAPY_SESSION_COMPANY_PRICE_IN_CENTS module-attribute 

THERAPY_SESSION_COMPANY_PRICE_IN_CENTS = 7000

THERAPY_SESSION_TNS_PRICE_IN_CENTS module-attribute 

THERAPY_SESSION_TNS_PRICE_IN_CENTS = 8000

get_therapy_sessions_guarantee_package 

get_therapy_sessions_guarantee_package(app_user_id)

Get the therapy sessions guarantee data for a user

Parameters:

Name Type Description Default
app_user_id int

The user ID

 required

Returns:

Type Description
BookingSessionPackage

The therapy sessions package
Source code in components/fr/public/clinic/guarantee.py 
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
def get_therapy_sessions_guarantee_package(
    app_user_id: int,
) -> BookingSessionPackage:
    """Get the therapy sessions guarantee data for a user

    Arguments:
        app_user_id: The user ID

    Returns:
        The therapy sessions package
    """
    from components.fr.public.company.queries import (
        get_default_company_id_for_user,
    )

    company_id = get_default_company_id_for_user(user_id=app_user_id)

    user_coverage = None
    insurance_profile_id = None

    user_coverage_payload = get_user_coverage_payload(app_user_id)

    if user_coverage_payload is not None:
        user_coverage, insurance_profile_id = user_coverage_payload

    session_count = (
        _get_therapy_session_count(user_coverage.guarantees) if user_coverage else None
    )

    price_in_cents = (
        THERAPY_SESSION_TNS_PRICE_IN_CENTS
        if company_id is None
        else THERAPY_SESSION_COMPANY_PRICE_IN_CENTS
    )

    package_session = None

    if user_coverage and insurance_profile_id:
        reimbursed_package = get_therapy_sessions_reimbursed_data(
            guarantees=user_coverage.guarantees if user_coverage else [],
            insurance_profile_id=insurance_profile_id,
            app_user_id=str(app_user_id),
        )

        if reimbursed_package:
            package_session, max_reimbursement_per_care = reimbursed_package

            # Dynamic pricing is only applied if users do not have legacy forfait
            if (
                not session_count
                and max_reimbursement_per_care is not None
                and max_reimbursement_per_care < price_in_cents
                and package_session.count_remaining
                and package_session.count_remaining > 0
            ):
                price_in_cents = max_reimbursement_per_care

    return BookingSessionPackage(
        price_in_cents=price_in_cents,
        included=(
            BookingSessionPackageCount(count_limit=session_count)
            if session_count
            else None
        ),
        reimbursed=package_session,
    )

get_therapy_sessions_reimbursed_data 

get_therapy_sessions_reimbursed_data(
    guarantees, insurance_profile_id, app_user_id
)

Get the therapy sessions reimbursed for a user

Parameters:

Name Type Description Default
guarantees list[CoverageGuarantee]

The user guarantees

 required
insurance_profile_id int

The insurance profile ID

 required
app_user_id str

The user ID

 required
Source code in components/fr/public/clinic/guarantee.py 
 99
100
101
102
103
104
105
106
107
108
109
110
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
143
144
145
146
147
148
149
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
178
179
180
181
182
183
184
def get_therapy_sessions_reimbursed_data(
    guarantees: list[CoverageGuarantee],
    insurance_profile_id: int,
    app_user_id: str,
) -> tuple[BookingSessionPackageCount, int | None] | None:
    """
    Get the therapy sessions reimbursed for a user

    Arguments:
        guarantees: The user guarantees
        insurance_profile_id: The insurance profile ID
        app_user_id: The user ID
    """
    from components.clinic.public.business_logic.therapist_booking_session import (
        get_past_sessions_that_need_invoice_generation_count_for_user,
        get_upcoming_reimbursed_therapist_booking_session_count_for_user,
    )
    from components.fr.internal.claim_management.enums.count_and_amount_usage import (
        RemainingUsageCode,
    )
    from components.fr.internal.claim_management.internal.misc.business_logic.remaining_usage import (
        get_remaining_usages_for_insurance_profile,
    )

    psycho_guarantees_with_remaining_code: list[RemainingUsageCode] = [
        RemainingUsageCode.ALTERNATIVE_MEDICINE,
        RemainingUsageCode.PSYCHOLOGY_NON_REIMBURSED,
    ]

    for guarantees_with_remaining_code in psycho_guarantees_with_remaining_code:
        guarantee_data = _find_guarantee_by_name(
            guarantees, guarantees_with_remaining_code.value
        )
        if not guarantee_data:
            continue

        care_type_mappings = guarantee_data.care_type_mappings
        if not care_type_mappings:
            continue

        selector = care_type_mappings[0].selector
        if "MED_DOUCE_PSYCH" not in selector.include_care_types:
            continue

        remaining_usages = get_remaining_usages_for_insurance_profile(
            insurance_profile_id=insurance_profile_id,
            remaining_usage_codes=[guarantees_with_remaining_code],
        )
        if not remaining_usages:
            continue

        remaining_usage: RemainingUsage = remaining_usages[0]
        count_limit = remaining_usage.count_limit

        if count_limit is not None and remaining_usage.count_remaining is not None:
            feature_user = FeatureUser(app_user_id=app_user_id, app_id=AppName.ALAN_FR)
            # We want to deduce the number of upcoming sessions that are going to be reimbursed
            # (the claims engine does not count them as the care acts do not exist yet)
            count_upcoming_reimbursed_sessions = (
                get_upcoming_reimbursed_therapist_booking_session_count_for_user(
                    feature_user=feature_user,
                    session_type=TherapistBookingSessionType.therapy,
                )
            )
            count_past_reimbursed_sessions_that_needs_invoice_generation = (
                get_past_sessions_that_need_invoice_generation_count_for_user(
                    feature_user=feature_user,
                    session_type=TherapistBookingSessionType.therapy,
                )
            )
            # If the member reached the limit of their forfait, we return 0
            count_remaining = max(
                remaining_usage.count_remaining
                - count_upcoming_reimbursed_sessions
                - count_past_reimbursed_sessions_that_needs_invoice_generation,
                0,
            )
            return (
                BookingSessionPackageCount(
                    count_limit=count_limit,
                    count_remaining=count_remaining,
                ),
                remaining_usage.max_reimbursement_per_care,
            )

    return None

get_user_coverage_payload 

get_user_coverage_payload(app_user_id)
Source code in components/fr/public/clinic/guarantee.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def get_user_coverage_payload(  # noqa: D103
    app_user_id: int,
) -> tuple[CoveragePayload, int] | None:
    from components.fr.internal.claim_management.public.coverage.api import (
        get_latest_coverage_payload,
    )

    user = get_or_raise_missing_resource(
        User,
        app_user_id,
        options=[
            selectinload(User.address),
            selectinload(User.companies),
            selectinload(User.insurance_profile).options(
                selectinload(InsuranceProfile.policies),  # type: ignore[arg-type]
            ),
        ],
    )

    insurance_profile = user.insurance_profile

    # Exempted?
    if insurance_profile is None:
        return None

    insurance_policy = insurance_profile.current_policy

    # Exempted?
    if insurance_policy is None:
        return None

    health_coverage = insurance_policy.current_health_coverage
    latest_coverage_payload = get_latest_coverage_payload(health_coverage.name)

    if latest_coverage_payload is None:
        return None

    return latest_coverage_payload, insurance_profile.id

id_verification

convert_user_data_id_verification_request_user_info 

convert_user_data_id_verification_request_user_info(
    user_data,
)

Converts UserData to IDVerificationRequestUserInfo for FR.

Source code in components/fr/public/clinic/id_verification.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def convert_user_data_id_verification_request_user_info(
    user_data: ClinicUserDataForIdVerification,
) -> IDVerificationRequestUserInfo:
    """
    Converts UserData to IDVerificationRequestUserInfo for FR.
    """
    return IDVerificationRequestUserInfo(
        first_name=create_first_name_with_additional_first_names(
            first_name=user_data.first_name,
            additional_first_names=user_data.additional_first_names,
        ),
        last_name=user_data.last_name,
        email=user_data.email,
        additional_properties=IDVerificationRequestUserInfoAdditionalProperties(
            date_of_birth=user_data.date_of_birth,
            place_of_birth=user_data.place_of_birth,
            birth_last_name=user_data.birth_last_name,
            gender=user_data.gender,
            session_id=user_data.session_id,
        ),
    )

create_first_name_with_additional_first_names 

create_first_name_with_additional_first_names(
    first_name, additional_first_names
)

Creates a first name string that includes additional first names if provided.

Source code in components/fr/public/clinic/id_verification.py 
 8
 9
10
11
12
13
14
15
16
17
18
def create_first_name_with_additional_first_names(
    first_name: str, additional_first_names: list[str] | None
) -> str:
    """
    Creates a first name string that includes additional first names if provided.
    """
    if not additional_first_names:
        return first_name

    all_names = [first_name] + additional_first_names
    return " ".join(all_names)

internal_teleconsultation

app_user_has_access_to_internal_teleconsultation 

app_user_has_access_to_internal_teleconsultation(user)

Check if the app user has access to internal teleconsultation.

Source code in components/fr/public/clinic/internal_teleconsultation.py 
 5
 6
 7
 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
33
34
35
36
37
38
39
40
def app_user_has_access_to_internal_teleconsultation(user: User) -> bool:
    """Check if the app user has access to internal teleconsultation."""
    try:
        from components.feature_flags_context.public.api import (
            get_feature_flags_context_for_user,
        )
        from components.feature_flags_context.public.dependencies import (
            get_app_dependency as get_feature_flags_context_data_accessors,
        )
        from shared.feature_flags.client import (
            bool_feature_flag,
            build_multi_context,
            get_anonymous_user_context_data,
            get_user_context_data,
        )

        context_data = build_multi_context(
            user_context_data=get_user_context_data(
                user_id=str(user.id),
                additional_context=get_feature_flags_context_for_user(
                    user, data_accessors=get_feature_flags_context_data_accessors()
                ),
            ),
            anonymous_user_context_data=get_anonymous_user_context_data(),
        )
        return bool_feature_flag(
            feature_flag_key="clinic-internal-teleconsultation",
            default_value=False,
            context_data=context_data,
        )
    except Exception:
        current_logger.exception(
            "Error checking if app user has access to internal teleconsultation",
            user_id=user.id,
        )
        return False

livi

actions

enable_telemedicine_with_livi 
enable_telemedicine_with_livi(user_id)

Enables telemedicine (with Livi) for a user by adding them to the Livi system. :param user_id: :return: True or False indicating success

Source code in components/fr/public/clinic/livi/actions.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
def enable_telemedicine_with_livi(user_id: int) -> bool:
    """
    Enables telemedicine (with Livi) for a user by adding them to the Livi system.
    :param user_id:
    :return: True or False indicating success
    """
    from components.fr.public.clinic.livi.queries import (
        is_user_eligible_for_telemedicine_with_livi,
    )

    if not is_user_eligible_for_telemedicine_with_livi(user_id=user_id):
        raise ErrorCode.invalid_arguments(
            message="An SSN is required to sign up for telemedicine with LIVI"
        )

    user = get_or_raise_missing_resource(User, user_id)
    livi_client = LiviClient()
    success = livi_client.add_user(user)

    if success:
        user.has_consented_ssn_for_telemedicine_partnership = True
        current_session.commit()

    return success
forcefully_reenable_telemedicine_with_livi 
forcefully_reenable_telemedicine_with_livi(user_id)

Forcefully re-enables telemedicine (with Livi) for a user by adding them to the Livi system. This is only allowed if the user has previously opted in for telemedicine. :param user_id: :return: True or False indicating success

Source code in components/fr/public/clinic/livi/actions.py 
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
def forcefully_reenable_telemedicine_with_livi(user_id: int) -> bool:
    """
    Forcefully re-enables telemedicine (with Livi) for a user by adding them to the Livi system.
    This is only allowed if the user has previously opted in for telemedicine.
    :param user_id:
    :return: True or False indicating success
    """
    from components.fr.public.clinic.livi.queries import (
        is_user_eligible_for_telemedicine_with_livi,
    )

    if not is_user_eligible_for_telemedicine_with_livi(user_id=user_id):
        raise ErrorCode.invalid_arguments(
            message="An SSN is required to reenable telemedicine with LIVI"
        )

    user = get_or_raise_missing_resource(User, user_id)
    # We only reenable the telemedecine if we're sure the user has actively opted
    # in for it _beforehand_
    if not user.has_consented_ssn_for_telemedicine_partnership:
        raise ErrorCode.forbidden(
            message="You can only re-enable telemedicine forcefully for a user that has not opted-in for it"
        )

    livi_client = LiviClient()
    success = livi_client.add_user(user)

    return success

commands

clinic_livi module-attribute 
clinic_livi = AppGroup(
    name="clinic_livi",
    help="All commands related to Livi in the scope of the clinic (billing livi, user management, etc)",
)
remove_uncovered_users_from_livi 
remove_uncovered_users_from_livi(dry_run)

Removes users who have opted in to Livi telemedicine partnership but do not have active insurance coverage.

Source code in components/fr/public/clinic/livi/commands.py 
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
68
69
70
71
72
73
74
@clinic_livi.command()
@command_with_dry_run
def remove_uncovered_users_from_livi(dry_run: bool) -> None:
    """
    Removes users who have opted in to Livi telemedicine partnership but do not have active insurance coverage.
    """
    from components.fr.public.clinic.livi.queries import (
        get_opted_in_users_without_active_enrollments,
    )

    livi_client = LiviClient()

    with transaction(propagation=Propagation.REQUIRES_NEW) as session:
        opted_in_users = get_opted_in_users_without_active_enrollments(session)

        total_opted_in_users_without_active_enrollments = len(opted_in_users)

        current_logger.info(
            f"Found {total_opted_in_users_without_active_enrollments} Livi opted-in users without active enrollments"
        )

        count = 0
        error_count = 0
        for user in opted_in_users:
            if user.insurance_profile is None or user.insurance_profile.ssn is None:
                current_logger.error(
                    f"User {user.id} has no insurance profile or no SSN, skipping"
                )
                error_count += 1
                continue
            if dry_run:
                count += 1
                current_logger.info(f"Would remove user {user.id} from Livi")
            else:
                try:
                    livi_client.remove_user(user.insurance_profile.ssn)
                    count += 1
                    user.has_consented_ssn_for_telemedicine_partnership = False
                    current_logger.info(f"Removed user {user.id} from Livi")
                except LiviUserNotFoundError:
                    # If the user is not found on Livi, we can consider them already removed and just update our local consent status.
                    count += 1
                    user.has_consented_ssn_for_telemedicine_partnership = False
                    current_logger.warning(
                        f"User {user.id} not found on Livi, opted out locally"
                    )
                except Exception:
                    error_count += 1
                    current_logger.exception(
                        f"Could not remove user {user.id} from Livi"
                    )

        if dry_run:
            current_logger.info(
                f"Would have removed and opted-out {count} users from Livi"
            )
        else:
            current_logger.info(f"Removed and opted-out {count} users from Livi")
            current_logger.info(f"Could not remove {error_count} users from Livi")

queries

get_opted_in_users_without_active_enrollments 
get_opted_in_users_without_active_enrollments(session)

Return Livi opted-in users who do not have any active enrollments as of today.

Ordering: by User.id ascending.

Source code in components/fr/public/clinic/livi/queries.py 
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
def get_opted_in_users_without_active_enrollments(session: Session) -> list[User]:
    """
    Return Livi opted-in users who do not have any active enrollments as of today.

    Ordering: by User.id ascending.
    """
    result = (
        session.execute(
            select(User)
            .options(
                load_only(
                    User.id,
                    User.has_consented_ssn_for_telemedicine_partnership,
                ),
                selectinload(User.insurance_profile).selectinload(
                    InsuranceProfile.enrollments
                ),
            )
            .join(User.insurance_profile)
            .outerjoin(
                # We make an outer join first, so we can filter on None later
                Enrollment,
                and_(
                    Enrollment.insurance_profile_id == InsuranceProfile.id,
                    Enrollment.is_active_on(utctoday()),
                ),
            )
            .filter(
                User.has_consented_ssn_for_telemedicine_partnership.is_(True),
                Enrollment.id.is_(None),  # No matching active enrollment
            )
            .order_by(User.id.asc())
        )
        .scalars()
        .unique()
        .all()
    )
    return list(result)
is_user_eligible_for_telemedicine_with_livi 
is_user_eligible_for_telemedicine_with_livi(user_id)

Checks if a user is eligible for telemedicine with Livi based on the presence of an SSN in their insurance profile. :param user_id: :return: True or False

Source code in components/fr/public/clinic/livi/queries.py 
11
12
13
14
15
16
17
18
def is_user_eligible_for_telemedicine_with_livi(user_id: int) -> bool:
    """
    Checks if a user is eligible for telemedicine with Livi based on the presence of an SSN in their insurance profile.
    :param user_id:
    :return: True or False
    """
    user = get_or_raise_missing_resource(User, user_id)
    return user.insurance_profile is not None and user.insurance_profile.ssn is not None

components.fr.public.company

queries

get_account_id_for_fr_company 

get_account_id_for_fr_company(company_id)

Return the account_id for a given FR company, raises the company doesn't exist.

Source code in components/fr/internal/business_logic/company/queries/company.py 
936
937
938
939
def get_account_id_for_fr_company(company_id: int) -> UUID:
    """Return the account_id for a given FR company, raises the company doesn't exist."""
    company = get_or_raise_missing_resource(Company, company_id)
    return company.account_id

get_all_franchises 

get_all_franchises()
Source code in components/fr/internal/business_logic/company/queries/franchise.py 
38
39
40
def get_all_franchises() -> list[FranchiseEntity]:
    franchises = current_session.query(Franchise).all()  # noqa: ALN085
    return [_convert_model_to_entity(franchise) for franchise in franchises]

get_company_ids_from_account 

get_company_ids_from_account(account_id)
Source code in components/fr/internal/business_logic/account/queries/account.py 
44
45
46
47
48
def get_company_ids_from_account(account_id: uuid.UUID) -> list[int]:
    company_ids = (
        current_session.query(Company.id).filter(Company.account_id == account_id).all()  # noqa: ALN085
    )
    return [company_id for (company_id,) in company_ids]

get_default_company_id_for_user 

get_default_company_id_for_user(user_id)
Source code in components/fr/internal/business_logic/company/queries/company.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
@tracer.wrap()
def get_default_company_id_for_user(user_id: int) -> int | None:
    from components.fr.internal.models.queries.user import (
        get_active_or_last_employment,
    )

    # Pre-load relationships.
    # NOTE: We are using `joinedload` even for one-to-many relationships, because
    # we expect the cardinality to be low (only a few employments/policies per user).
    # This saves some round-trips to the database.
    user = (
        current_session.query(User)  # noqa: ALN085
        .filter(User.id == user_id)
        .options(
            joinedload(User.alan_employee),
            joinedload(User.employments),
            joinedload(User.insurance_profile).options(
                joinedload(InsuranceProfile.enrollments)
                .joinedload(Enrollment.policy)
                .joinedload(Policy.contract),
                joinedload(InsuranceProfile.policies).options(  # type: ignore[arg-type]
                    joinedload(Policy.contract),
                    joinedload(Policy.enrollments),
                ),
            ),
        )
        .one_or_none()
    )
    if not user:
        raise ErrorCode.missing_resource()

    # User is currently on a policy from an individual contract
    if (
        user.insurance_profile
        and user.insurance_profile.current_policy
        and user.insurance_profile.current_policy.contract.contractee_type
        == ContracteeType.individual
    ):
        return None

    # User is a beneficiary
    if (
        user.insurance_profile
        and user.insurance_profile.current_policy
        and user.insurance_profile
        != user.insurance_profile.current_policy.primary_profile
        and user.insurance_profile.current_policy.contract.contractee_type
        == ContracteeType.company
    ):
        return user.insurance_profile.current_policy.contract.company_id

    # User is currently insured
    if (
        user.insurance_profile
        and user.insurance_profile.active_enrollment
        and user.insurance_profile.active_enrollment.policy.contract.contractee_type
        == ContracteeType.company
    ):
        return user.insurance_profile.active_enrollment.policy.contract.company_id

    # User is invited or exempted or non-currently insured
    active_or_last_employment = get_active_or_last_employment(user.id)
    if active_or_last_employment:
        return active_or_last_employment.company_id

    return None

get_franchise 

get_franchise(id)
Source code in components/fr/internal/business_logic/company/queries/franchise.py 
18
19
20
def get_franchise(id: uuid.UUID) -> FranchiseEntity:
    franchise = get_or_raise_missing_resource(Franchise, id)
    return _convert_model_to_entity(franchise)

components.fr.public.contract

member_attributes

company_name module-attribute 

company_name = MemberAttributeDefinition[str](
    name="company_name",
    display_name="Company name",
    description="Company contractee name",
    getter=_get_company_name,
    raw_type=str,
)

contract_end_date module-attribute 

contract_end_date = MemberAttributeDefinition[date](
    name="contract_end_date",
    display_name="Contract end date",
    description="Health contract end date",
    getter=_get_contract_end_date,
    raw_type=date,
)

contract_has_madelin_attestation_available module-attribute 

contract_has_madelin_attestation_available = MemberAttributeDefinition[
    bool
](
    name="contract_has_madelin_attestation_available",
    display_name="Madelin attestation available",
    description="Whether madelin attestation is available for previous year",
    getter=_get_contract_has_madelin_attestation_available,
    raw_type=bool,
)

contract_start_date module-attribute 

contract_start_date = MemberAttributeDefinition[date](
    name="contract_start_date",
    display_name="Contract start date",
    description="Health contract start date",
    getter=_get_contract_start_date,
    raw_type=date,
)

contract_type module-attribute 

contract_type = MemberAttributeDefinition[ContractPlanInfo](
    name="contract_type",
    display_name="Contract type",
    description="Contract plan type and segment",
    getter=_get_contract_type,
    raw_type=ContractPlanInfo,
    formatter=_format_contract_type,
)

has_income_based_children_discount_rules module-attribute 

has_income_based_children_discount_rules = MemberAttributeDefinition[
    bool
](
    name="has_income_based_children_discount_rules",
    display_name="Has income-based children discount rules",
    description="Whether contract has income-based children discount rules",
    getter=_get_has_income_based_children_discount_rules,
    raw_type=bool,
)

health_contract_coverage module-attribute 

health_contract_coverage = MemberAttributeDefinition[
    HealthContractCoverageInfo
](
    name="health_contract_coverage",
    display_name="Health contract coverage",
    description="Health contract coverage participation info",
    getter=_get_health_contract_coverage,
    raw_type=HealthContractCoverageInfo,
    formatter=_format_health_contract_coverage,
)

health_contract_status module-attribute 

health_contract_status = MemberAttributeDefinition[
    HealthContractStatusInfo
](
    name="health_contract_status",
    display_name="Health contract status",
    description="Health contract status with dates",
    getter=_get_health_contract_status,
    raw_type=HealthContractStatusInfo,
    formatter=_format_health_contract_status,
)

individual_plan_name module-attribute 

individual_plan_name = MemberAttributeDefinition[str](
    name="individual_plan_name",
    display_name="Individual plan name",
    description="Individual plan display name",
    getter=_get_individual_plan_name,
    raw_type=str,
)

is_direct_billing module-attribute 

is_direct_billing = MemberAttributeDefinition[bool](
    name="is_direct_billing",
    display_name="Is direct billing",
    description="Whether contract uses direct billing",
    getter=_get_is_direct_billing,
    raw_type=bool,
)

is_direct_billing_of_options module-attribute 

is_direct_billing_of_options = MemberAttributeDefinition[
    bool
](
    name="is_direct_billing_of_options",
    display_name="Is direct billing of options",
    description="Whether contract options use direct billing",
    getter=_get_is_direct_billing_of_options,
    raw_type=bool,
)

plan_number_of_options module-attribute 

plan_number_of_options = MemberAttributeDefinition[int](
    name="plan_number_of_options",
    display_name="Plan number of options",
    description="Number of options in current plan",
    getter=_get_plan_number_of_options,
    raw_type=int,
)

prevoyance_contract_start_date module-attribute 

prevoyance_contract_start_date = MemberAttributeDefinition[
    date
](
    name="prevoyance_contract_start_date",
    display_name="Prevoyance contract start date",
    description="Start date of first active prevoyance contract",
    getter=_get_prevoyance_contract_start_date,
    raw_type=date,
)

option_member_attributes

Member attribute definitions for FR contract option attributes.

contract_option module-attribute 

contract_option = MemberAttributeDefinition[int](
    name="contract_option",
    display_name="Contract option",
    description="Subscribed option number",
    getter=_get_contract_option,
    raw_type=int,
)

contract_option_end_date module-attribute 

contract_option_end_date = MemberAttributeDefinition[date](
    name="contract_option_end_date",
    display_name="Contract option end date",
    description="Option contract end date",
    getter=_get_contract_option_end_date,
    raw_type=date,
)

contract_option_info module-attribute 

contract_option_info = MemberAttributeDefinition[
    ContractOptionStatus
](
    name="contract_option_info",
    display_name="Contract option info",
    description="Contract option subscription status",
    getter=_get_contract_option_info,
    raw_type=ContractOptionStatus,
    formatter=_format_contract_option_info,
)

contract_option_start_date module-attribute 

contract_option_start_date = MemberAttributeDefinition[
    date
](
    name="contract_option_start_date",
    display_name="Contract option start date",
    description="Option contract start date",
    getter=_get_contract_option_start_date,
    raw_type=date,
)

queries

health_contracts

get_eligible_health_contract 
get_eligible_health_contract()

The eligibility relationship is based on contract population and employment details.

For the eligibility to be considered, the contract and employment must NOT be cancelled. Also, contracts must be signed.

Source code in components/fr/internal/business_logic/contract/queries/health_contracts_backend/query_api.py 
256
257
258
259
260
261
262
def get_eligible_health_contract() -> _GetEligibleHealthContract:
    """The eligibility relationship is based on contract population and employment details.

    For the eligibility to be considered, the contract and employment
    must NOT be cancelled. Also, contracts must be signed.
    """
    return _GetEligibleHealthContract()
get_health_contract_information 
get_health_contract_information(contract_id, on_date)
Source code in components/fr/internal/business_logic/contract/queries/health_contracts.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def get_health_contract_information(
    contract_id: int, on_date: date
) -> HealthContractInfo:
    contract = get_or_raise_missing_resource(Contract, contract_id)
    is_company_contractee = contract.company_contractee is not None
    contract_version = get_ongoing_contract_version(
        contract_id=contract_id, on_date=on_date
    )
    company_name = (
        contract.company_contractee.display_name
        if contract.company_contractee
        else None
    )

    return HealthContractInfo(
        contractee_type=(
            ContracteeType.company
            if is_company_contractee
            else ContracteeType.individual
        ),
        company_name=company_name,
        company_participation_unit="cents",
        is_partner_coverage_mandatory=(
            contract_version.is_partner_coverage_mandatory
            if contract_version
            and contract_version.is_partner_coverage_mandatory is not None
            else None
        ),
        is_children_coverage_mandatory=(
            contract_version.is_children_coverage_mandatory
            if contract_version
            and contract_version.is_children_coverage_mandatory is not None
            else None
        ),
        company_participation_partner=(
            int(contract_version.participation_partner)
            if contract_version and contract_version.participation_partner is not None
            else None
        ),
        company_participation_children=(
            int(contract_version.participation_children)
            if contract_version and contract_version.participation_children is not None
            else None
        ),
        is_direct_billing=(
            contract_version.is_direct_billing
            if contract_version and contract_version.is_direct_billing
            else False
        ),
    )
is_health_contract_plan_with_madelin 
is_health_contract_plan_with_madelin(contract_id, on_date)
Source code in components/fr/internal/business_logic/contract/queries/health_contracts.py 
313
314
315
316
317
318
def is_health_contract_plan_with_madelin(contract_id: int, on_date: date) -> bool:
    primary_contract = get_or_raise_missing_resource(Contract, contract_id)

    primary_plan = primary_contract.plan_on(on_date)

    return primary_plan.has_madelin

income

get_dependent_income_brackets_for_subscriptions 
get_dependent_income_brackets_for_subscriptions(
    subscription_ids, on_date
)

Get dependent income brackets for multiple subscriptions from contract discount rules.

This function extracts income brackets from the discount rules associated with each contract. It filters for discounts that are valid on the given date and converts discount rule income brackets to IncomeBracketEntity objects.

Parameters:

Name Type Description Default
subscription_ids list[int]

List of subscription IDs to get dependent brackets for

 required
on_date date

The date to check the income brackets against

 required

Returns: dict[int, list[IncomeBracketEntity]]: Mapping of subscription_id to dependent brackets

Source code in components/fr/internal/contract/queries/income.py 
 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 get_dependent_income_brackets_for_subscriptions(
    subscription_ids: list[int],
    on_date: date,
) -> dict[int, list[IncomeBracketEntity]]:
    """
    Get dependent income brackets for multiple subscriptions from contract discount rules.

    This function extracts income brackets from the discount rules associated with each contract.
    It filters for discounts that are valid on the given date and converts discount rule
    income brackets to IncomeBracketEntity objects.

    Args:
        subscription_ids: List of subscription IDs to get dependent brackets for
        on_date: The date to check the income brackets against
    Returns:
        dict[int, list[IncomeBracketEntity]]: Mapping of subscription_id to dependent brackets
    """
    if not subscription_ids:
        return {}

    contracts = _get_contracts_by_ids(subscription_ids)

    result = {}
    for contract in contracts:
        # Only return brackets for contracts linked to a company (employee plans)
        # and not individual contracts (retiree)
        if contract.company_contractee is None:
            continue

        brackets = []

        # Find discounts that are valid on the given date
        for discount in contract.discounts:
            if discount.start_month <= on_date <= discount.end_month:
                # Extract income brackets from discount rules
                for discount_rule in discount.discount_rules:
                    if discount_rule.primary_min_income is not None:
                        bracket = IncomeBracketEntity(
                            min=discount_rule.primary_min_income,
                            max=discount_rule.primary_max_income,
                        )
                        brackets.append(bracket)

        # Sort brackets by min income for consistent ordering
        brackets.sort(key=lambda b: b.min)

        if brackets:
            result[contract.id] = brackets

    return result
get_income_brackets_for_subscriptions 
get_income_brackets_for_subscriptions(
    subscription_ids, on_date, for_target=None
)

Returns income brackets for multiple subscriptions on a specific date. This function is optimized to avoid N+1 queries by batching database operations.

Parameters:

Name Type Description Default
subscription_ids list[int]

List of subscription IDs to get income brackets for

 required
on_date date

The date to check the income brackets against

 required
for_target Optional[IncomeBracketTarget]

Optional target for which the income brackets are requested

 None

Returns: dict[int, list[IncomeBracketEntity]]: Mapping of subscription_id to income brackets Only includes subscriptions that have income brackets

Source code in components/fr/internal/contract/queries/income.py 
125
126
127
128
129
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def get_income_brackets_for_subscriptions(
    subscription_ids: list[int],
    on_date: date,
    for_target: Optional[IncomeBracketTarget] = None,
) -> dict[int, list[IncomeBracketEntity]]:
    """
    Returns income brackets for multiple subscriptions on a specific date.
    This function is optimized to avoid N+1 queries by batching database operations.

    Args:
        subscription_ids: List of subscription IDs to get income brackets for
        on_date: The date to check the income brackets against
        for_target: Optional target for which the income brackets are requested
    Returns:
        dict[int, list[IncomeBracketEntity]]: Mapping of subscription_id to income brackets
        Only includes subscriptions that have income brackets
    """
    if not subscription_ids:
        return {}

    option_brackets = get_option_income_brackets_for_subscriptions(
        subscription_ids, on_date
    )

    dependent_brackets = get_dependent_income_brackets_for_subscriptions(
        subscription_ids, on_date
    )

    result = {}
    for subscription_id in subscription_ids:
        brackets_for_option = option_brackets.get(subscription_id, [])
        brackets_for_dependent = dependent_brackets.get(subscription_id, [])

        if for_target == IncomeBracketTarget.option and not brackets_for_option:
            continue
        if for_target == IncomeBracketTarget.dependent and not brackets_for_dependent:
            continue

        merged_brackets = _merge_income_brackets(
            brackets_for_option, brackets_for_dependent
        )

        if merged_brackets:
            result[subscription_id] = merged_brackets

    return result
get_member_type_for_user_on 
get_member_type_for_user_on(user_id, on_date)
Source code in components/fr/internal/contract/queries/income.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def get_member_type_for_user_on(user_id: str, on_date: date) -> Optional[MemberType]:
    subscription_id = get_subscription_id_for_user_on(
        user_id=int(user_id), on_date=on_date
    )
    if subscription_id is None:
        return None

    contract = get_or_raise_missing_resource(Contract, subscription_id)
    plan = contract.plan_on(on_date)

    if (
        plan.is_collective_retiree_plan
    ):  # For our use case, the retirees are attached to a collective plan
        return MemberType.retiree

    if plan.is_company_plan:
        return MemberType.employee

    return None
get_option_income_brackets_for_subscriptions 
get_option_income_brackets_for_subscriptions(
    subscription_ids, on_date
)

Get option income brackets for multiple subscriptions.

Parameters:

Name Type Description Default
subscription_ids list[int]

List of subscription IDs to get option brackets for

 required
on_date date

The date to check the income brackets against

 required

Returns: dict[int, list[IncomeBracketEntity]]: Mapping of subscription_id to option brackets

Source code in components/fr/internal/contract/queries/income.py 
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
def get_option_income_brackets_for_subscriptions(
    subscription_ids: list[int], on_date: date
) -> dict[int, list[IncomeBracketEntity]]:
    """
    Get option income brackets for multiple subscriptions.

    Args:
        subscription_ids: List of subscription IDs to get option brackets for
        on_date: The date to check the income brackets against
    Returns:
        dict[int, list[IncomeBracketEntity]]: Mapping of subscription_id to option brackets
    """
    if not subscription_ids:
        return {}

    contracts = _get_contracts_by_ids(subscription_ids)

    return {
        contract.id: [
            IncomeBracketEntity(
                min=option_price_rule.primary_income_bracket[0],
                max=option_price_rule.primary_income_bracket[1],
            )
            for option_price_rule in contract.plan_on(on_date).option_price_rules
            if option_price_rule.primary_income_bracket is not None
        ]
        for contract in contracts
    }

components.fr.public.customer_health_partner

get_admin_traits

get_admin_traits_to_notify 

get_admin_traits_to_notify(admin_id, company_ids)

Return the list of AdminTraits for admins who should be notified about the well-being assessment report.

Source code in components/fr/public/customer_health_partner/get_admin_traits.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
def get_admin_traits_to_notify(
    admin_id: str | None, company_ids: list[str]
) -> Sequence[AdminTraits]:
    """
    Return the list of AdminTraits for admins who should be notified about the well-being assessment report.
    """
    users_traits_of_users_to_notify: list[AdminTraits] = []
    # we notify the admin who created the assessment
    if admin_id is not None:
        admin_user = get_or_raise_missing_resource(User, admin_id)
        users_traits_of_users_to_notify.append(AdminTraits(admin_user))

    # we notify all wellbeing referents
    for company_id in company_ids:
        for wellbeing_referent_user_trait in [
            AdminTraits(user)
            for user in get_active_wellbeing_referent_for_company(
                company_id=int(company_id)
            )
        ]:
            if wellbeing_referent_user_trait not in users_traits_of_users_to_notify:
                users_traits_of_users_to_notify.append(wellbeing_referent_user_trait)

    return users_traits_of_users_to_notify

components.fr.public.data_retention

queries

get_inactive_profile_ids_since 

get_inactive_profile_ids_since(inactive_number_of_years)

Return global profile IDs that have been inactive for the given number of years.

"Inactive" means all enrollments are either cancelled before the cutoff or ended before the cutoff.

Parameters:

Name Type Description Default
inactive_number_of_years int

Number of years of inactivity threshold

 required

Returns:

Type Description
list[UUID]

List of global profile IDs (UUID) that have been inactive
Source code in components/fr/public/data_retention/queries.py 
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 get_inactive_profile_ids_since(inactive_number_of_years: int) -> list[UUID]:
    """Return global profile IDs that have been inactive for the given number of years.

    "Inactive" means all enrollments are either cancelled before the cutoff
    or ended before the cutoff.

    Args:
        inactive_number_of_years: Number of years of inactivity threshold

    Returns:
        List of global profile IDs (UUID) that have been inactive
    """
    insurance_profile_ids = _get_inactive_insurance_profile_ids(
        inactive_number_of_years
    )

    insurance_profiles_query = (
        select(InsuranceProfile)
        .where(InsuranceProfile.id.in_(insurance_profile_ids))
        .options(joinedload(InsuranceProfile.user))
    )
    insurance_profiles = current_session.scalars(insurance_profiles_query).all()

    return [ip.user.profile_id for ip in insurance_profiles]

get_inactive_user_ids_since 

get_inactive_user_ids_since(inactive_number_of_years)

Return app user IDs that have been inactive for the given number of years.

"Inactive" means all enrollments are either cancelled before the cutoff or ended before the cutoff.

Parameters:

Name Type Description Default
inactive_number_of_years int

Number of years of inactivity threshold

 required

Returns:

Type Description
list[str]

List of FR app user IDs (str) that have been inactive
Source code in components/fr/public/data_retention/queries.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def get_inactive_user_ids_since(inactive_number_of_years: int) -> list[str]:
    """Return app user IDs that have been inactive for the given number of years.

    "Inactive" means all enrollments are either cancelled before the cutoff
    or ended before the cutoff.

    Args:
        inactive_number_of_years: Number of years of inactivity threshold

    Returns:
        List of FR app user IDs (str) that have been inactive
    """
    insurance_profile_ids = _get_inactive_insurance_profile_ids(
        inactive_number_of_years
    )

    query = select(InsuranceProfile.user_id).where(
        InsuranceProfile.id.in_(insurance_profile_ids)
    )

    return [str(uid) for uid in current_session.scalars(query).all()]

components.fr.public.demo_account

helpers

is_demo_mode_enabled 

is_demo_mode_enabled()

:return: a boolean telling if the app runs in demo mode (i.e. either in the demo environment, or in dev environment with demo mode enabled)

Source code in components/fr/public/demo_account/helpers.py 
 5
 6
 7
 8
 9
10
11
12
13
def is_demo_mode_enabled() -> bool:
    """
    :return: a boolean telling if the app runs in demo mode (i.e. either in the demo environment, or in dev environment with demo mode enabled)
    """
    return (
        is_demo_mode()
        or is_development_mode()
        and current_config.get("LOCAL_DEMO_MODE", False)
    )

components.fr.public.dependents

declare_dependent_status

Dependents dataclass 

Dependents(has_partner, has_children, effective_date=None)

Bases: DataClassJsonMixin

effective_date class-attribute instance-attribute 
effective_date = None
has_children instance-attribute 
has_children
has_partner instance-attribute 
has_partner

get_user_declared_dependents_on_date 

get_user_declared_dependents_on_date(user_id, on_date)
Source code in components/fr/internal/dependents/business_logic/queries/dependents.py 
67
68
69
70
71
72
73
74
75
76
def get_user_dependents_on_date(user_id: int, on_date: date) -> Dependents:
    persisted_dependents = get_user_persisted_dependents_on_date(user_id, on_date)

    result = (
        persisted_dependents_to_dependents(persisted_dependents)
        if persisted_dependents is not None
        else Dependents(has_children=None, has_partner=None)
    )

    return result

update_user_dependents 

update_user_dependents(
    user_id, effective_date, dependents, commit=False
)
Source code in components/fr/internal/dependents/business_logic/actions/dependents.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
def update_user_dependents(
    user_id: int,
    effective_date: date,
    dependents: Dependents,
    commit: bool = False,
) -> Dependents:
    current_persisted_dependents = get_user_persisted_dependents_on_date(
        user_id, effective_date
    )

    if current_persisted_dependents is not None:
        if dependents.has_partner is None:
            dependents = replace(
                dependents, has_partner=current_persisted_dependents.has_partner
            )
        if dependents.has_children is None:
            dependents = replace(
                dependents, has_children=current_persisted_dependents.has_children
            )

    if (
        current_persisted_dependents is not None
        and current_persisted_dependents.effective_date == effective_date
    ):
        current_persisted_dependents.has_partner = dependents.has_partner
        current_persisted_dependents.has_children = dependents.has_children

    else:
        persisted_dependents = PersistedDependents(
            user_id=user_id,
            has_partner=dependents.has_partner,
            has_children=dependents.has_children,
            effective_date=effective_date,
        )
        current_session.add(persisted_dependents)

    if commit:
        current_session.commit()
    return dependents

components.fr.public.document_parsing

classify_document_category

classify_insurance_document_category 

classify_insurance_document_category(transcription)

Classify an insurance document based on its transcription

Source code in components/fr/public/document_parsing/classify_document_category.py 
11
12
13
14
15
16
17
def classify_insurance_document_category(
    transcription: Transcription,
) -> MulticlassPrediction[PredictedDocumentCategory]:
    """
    Classify an insurance document based on its transcription
    """
    return _classify_insurance_document_category(transcription)

entities

document_category

DocumentCategoryDisplayInfo dataclass 
DocumentCategoryDisplayInfo(category, lang, label, icon)

Bases: DataClassJsonMixin

This is the info to display a document category (label, icon)

category instance-attribute 
category
icon instance-attribute 
icon
label instance-attribute 
label
lang instance-attribute 
lang

rejection_reason

RejectionReason dataclass 
RejectionReason(reason, defaultMessage)

Bases: DataClassJsonMixin

A rejection reason is made of 2 things: - a reason, it's like a key - a default Message we want to display to the member. It will be used in email/member dashboard/default of a CareEvent

defaultMessage instance-attribute 
defaultMessage
reason instance-attribute 
reason

errors

Public API for document parsing errors in FR component.

DocumentClassificationError

Bases: Exception

DocumentTranscriptionError

Bases: Exception

queries

document_category

get_document_categories 
get_document_categories(lang=Lang.english)

Get the list of document categories with display info for document in France For now we only return categories that are handled by the new parsing tool return: list[DocumentCategoryDisplayInfo]

Source code in components/fr/public/document_parsing/queries/document_category.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
47
48
49
50
51
52
53
54
55
56
57
58
def get_document_categories(
    lang: Lang = Lang.english,
) -> list[DocumentCategoryDisplayInfo]:
    """
    Get the list of document categories with display info for document in France
    For now we only return categories that are handled by the new parsing tool
    return: list[DocumentCategoryDisplayInfo]
    """
    document_categories_with_icon = [
        (ClaimInsuranceDocumentCategory.quote, "IconNotes"),
        (ClaimInsuranceDocumentCategory.ss_attestation, "IconAddressBook"),
        (
            ClaimInsuranceDocumentCategory.birth_or_adoption_certificate,
            "IconBabyCarriage",
        ),
        (ClaimInsuranceDocumentCategory.ss_decompte, "IconAbacus"),
        (ClaimInsuranceDocumentCategory.invoice, "IconFileInvoice"),
        (ClaimInsuranceDocumentCategory.mutuelle_decompte, "IconScale"),
        (ClaimInsuranceDocumentCategory.prescription, "IconFilePencil"),
        (
            ClaimInsuranceDocumentCategory.mutuelle_no_coverage_attestation,
            "IconUmbrella",
        ),
        (ClaimInsuranceDocumentCategory.unsupported, "IconFileUnknown"),
        (
            ClaimInsuranceDocumentCategory.cancel_teletransmission_request,
            "IconCircleXFilled",
        ),
        (ClaimInsuranceDocumentCategory.medical_results, "IconReportMedical"),
        (ClaimInsuranceDocumentCategory.medical_imaging, "IconDeviceCameraPhone"),
        (ClaimInsuranceDocumentCategory.medical_certificate, "IconFileCertificate"),
        (ClaimInsuranceDocumentCategory.non_claims_prescription, "IconFileDescription"),
        (ClaimInsuranceDocumentCategory.other_health, "IconQuestionMark"),
        (ClaimInsuranceDocumentCategory.unknown, "IconQuestionMark"),
    ]
    return [
        DocumentCategoryDisplayInfo(
            category=category,
            lang=lang,
            label=translate(
                language=lang, key_string=f"parsing_tool.document_categories.{category}"
            ),
            icon=icon,
        )
        for category, icon in document_categories_with_icon
    ]
get_document_category_label 
get_document_category_label(
    document_category, lang=Lang.english
)

Retrieve the label for the given document category return: string

Source code in components/fr/public/document_parsing/queries/document_category.py 
61
62
63
64
65
66
67
68
69
70
71
72
def get_document_category_label(
    document_category: ClaimInsuranceDocumentCategory,
    lang: Lang = Lang.english,
) -> str:
    """
    Retrieve the label for the given document category
    return: string
    """
    return translate(
        language=lang,
        key_string=f"parsing_tool.document_categories.{document_category}",
    )

document_transcription

get_best_possible_transcription 
get_best_possible_transcription(
    document,
    min_text_len=0,
    require_geometry=False,
    use_transcription_cache=False,
    transcribers_priority=None,
)

Get the best possible transcription of a document.

Source code in components/fr/public/document_parsing/queries/document_transcription.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def get_best_possible_transcription(
    document: TranscriptibleDocument,
    min_text_len: int = 0,
    require_geometry: bool = False,
    use_transcription_cache: bool = False,
    transcribers_priority: list[TranscriberLabel] | None = None,
) -> Transcription:
    """
    Get the best possible transcription of a document.
    """
    return _get_best_possible_transcription(
        document=document,
        min_text_len=min_text_len,
        require_geometry=require_geometry,
        use_transcription_cache=use_transcription_cache,
        transcribers_priority=transcribers_priority,
    )

helpers

HEALTH_DOCUMENT_UPLOAD_URL module-attribute 
HEALTH_DOCUMENT_UPLOAD_URL = (
    "https://alan.com/app/dashboard?documentUploader=claim"
)
REJECTION_BODY_FOR_USER module-attribute 
REJECTION_BODY_FOR_USER = {'default_message': {unreadable: f'🧐 Nous ne parvenons pas à lire les informations sur ce document.

👉 Nous vous invitons à nous le transférer à nouveau :
  - dans un format standard type (.JPEG, .PNG ou .PDF),
  - En vous assurant que l'image est nette et que la qualité est suffisante.

  Transférer une nouvelle version du document :{HEALTH_DOCUMENT_UPLOAD_URL}
', unusable_partial_document: f'🤔 Le document envoyé n'est pas complet. Des informations sont donc manquantes pour pouvoir estimer votre remboursement.

👉 Nous vous invitons à nous le renvoyer dans son intégralité (toutes les pages, chaque page entièrement visible).

💁 Dans le cas d'un décompte de soins, vous pouvez le télécharger depuis le site de la Sécurité Sociale, rubrique "Mes paiements" en cliquant sur "Télécharger un décompte" (la petite flèche située à droite des frais de santé concernés).

Transférer une nouvelle version du document : f{HEALTH_DOCUMENT_UPLOAD_URL}', duplicate: "🤔 On dirait que ce document est déjà en notre possession.\n\n👉 Si c'est une erreur, vous pouvez nous renvoyer le bon document, nous le traiterons rapidement.", beneficiary_not_covered: "🤔 On dirait que ce document concerne un bénéficiaire qui n'est pas couvert par Alan.\n\n👉 Nous ne pouvons donc pas procéder au remboursement.", insufficent_info_for_parsing: "🤔 Le document ne contient pas toutes les informations pour estimer votre remboursement. Il y manque :\n\n- Nom du bénéficiaire\n- Nom du professionnel de santé\n- Numéro d'inscription au registre ADELI du professionnel de santé\n- Type de soin / consultation\n- Prix de l'acte (pour chacun des actes séparément)\n- Base de remboursement de la Sécurité sociale ou BRSS (pour chacun des actes séparément)\n- Répartition du dépassement d'honoraires (sur chacun des actes)\n- Code de regroupement pour chacun des actes mentionnés\n- Code de regroupement pour chacun des éléments de l'équipement optique afin que nous puissions identifier la complexité de vos verres et appliquer la garantie correspondante\n- Date de chaque soin / consultation- Décompte complet et détaillé correspondant\n- Facture acquittée à la réception du colis\n- Devis initial avec les bases des actes envisagés\n- Type de soins ou facture détaillée correspondante\n- Si l'intervention porte sur les deux yeux ou un seul oeil (si oui, lequel ?)\n- Décompte complet et détaillé des soins remboursés, accompagné de ce document\n- Certains codes de regroupements ou codes actes apparaissant dans le document sont périmés et ne doivent plus être utilisés\n\n😀 Cependant sachez qu'en général, nous n'avons pas besoin de facture ! Si votre télétransmission est activée, et que votre soin est couvert par la Sécurité sociale, nous devrions recevoir les infos directement de leur part. Toutefois, si vous trouvez le délai trop long (> 15 jours), vous gardez bien sûr la possibilité :\n- de vérifier sur votre compte Améli si le remboursement a déjà été fait par la Sécurité sociale : si oui, vous pouvez nous transmettre le décompte à la place,\n- et, à défaut, de contacter le professionnel de santé pour savoir s'il a bien fait la déclaration de son côté", non_EU_prescription: "🤔 On dirait que ce document est une ordonnance obtenue en dehors de l'Union Européenne.\n\n👉 Nous acceptons uniquement les ordonnances venant des pays de l'Union Européenne.\n\nTransférer une nouvelle version du document : https://alan.com/app/dashboard?documentUploader=claim", other_insurer_decompte_100: '🧐 Nous ne pouvons pas effectuer de remboursement complémentaire car les soins ont déjà été intégralement pris en charge par votre complémentaire principale. Merci de nous transmettre uniquement des documents pour lesquels il reste des frais à votre charge.', cares_received_abroad: '😎 Alan couvre les soins des séjours à l’étranger sur la base de remboursement de la Sécurité Sociale française.\n\n👉 Pensez donc à conserver l’ensemble des factures acquittées et des justificatifs de paiement lors des soins effectués hors de France.\nEnsuite, adressez ces factures accompagnées du formulaire S3125 « Soins reçus à l’étranger », à votre caisse d’assurance maladie.\n\nVous trouverez ce document sur le site Ameli : https://www.ameli.fr/sites/default/files/formualires/221/s3125.pdf\n\nNous récupérerons ensuite les informations de la Sécurité Sociale (télétransmission ou envoi du décompte de votre part)', secu_decompte_100: '🧐 Nous ne pouvons pas effectuer de remboursement complémentaire car les soins ont déjà été intégralement pris en charge par la sécurité sociale. Merci de nous transmettre uniquement des documents pour lesquels il reste des frais à votre charge.', ask_paid_invoice: f'🤔 Le document indique une demande de règlement de votre part.

👉 Nous vous invitons à nous transférer la facture acquittée (parfois appelée quittance ou bordereau).

Vous avez également la possibilité de nous transférer un décompte de Sécu (si possible au format .PDF pour un traitement automatique ultra-rapide !).

Transférer une nouvelle version du document : {HEALTH_DOCUMENT_UPLOAD_URL}', missing_mt_or_dmt: '🤔 Le document ne contient pas toutes les informations pour estimer votre remboursement.\n\n👉 Nous vous invitons à renseigner le DMT/MT en cliquant sur le bouton ci-dessous ou à nous renvoyer une version contenant les informations suivantes :\n- Code MT ou "Mode de traitement". Ce mode est un code utilisé par l\'Assurance Maladie pour identifier la nature du soin;\n- Code DMT ou "Discipline Médico-tarifaire". Cette discipline est utilisée dans le contexte hospitalier, et permet de caractériser l\'activité de l\'unité médicale ou du service qui vous a fourni le soin.\n\n☎️ Ces informations vous seront fournies en général par l\'hôpital, nous vous invitons à les contacter directement.', etiopathe: "😔 Nous ne pouvons pas couvrir ces soins.\n\n👉 En effet nous ne remboursons les séances d'étiopathie que lorsqu’elles sont dispensées par un professionnel enregistré à l'Institut Français d’Étiopathie.", kine_invoice_without_mezieres: "🤗 Nous avons bien reçu votre facture de kinésithérapie. Elle n'est pas nécessaire pour vous rembourser ! En effet, nous remboursons les actes de kinésithérapie à réception des informations de la Sécurité Sociale.\n\n👉 Si la télétransmission est active, nous devrions recevoir automatiquement ces informations, et nous pourrons calculer votre remboursement dans la foulée.\nSi la sécurité sociale vous a déjà remboursé et que la télétransmission tarde à arriver ou qu’elle est inactive, transférez-nous votre décompte Améli reprenant ces soins depuis votre application Alan.", hospital_receipt: '🤗 Nous avons bien le reçu de paiement pour vos frais hospitaliers.\n\n👉 Ce seul document ne nous permet pas de traiter votre dossier. Si vos soins ont eu lieu en :\nCLINIQUE PRIVEE :\n- Nous avons besoin du bordereau de facturation détaillé acquitté : celui-ci reprend l\'intégralité de vos frais hospitaliers et est à demander au secrétariat de l\'établissement dans lequel vous avez reçu les soins. Sans ce document, nous ne pourrons malheureusement pas traiter votre dossier.\nHOPITAL PUBLIC :\n- Nous avons besoin de l\'avis des sommes à payer détaillé (recto et verso s\'il y en a un).\n- Cependant, dans le cas où vous voyez déjà des soins hospitaliers en attente de traitement depuis votre compte Alan (frais de séjour, consultation anesthésiste à l\'hôpital, consultation chirurgien à l\'hôpital, etc.), n\'hésitez pas à nous contacter afin que nous puissions manuellement rapprocher les informations entre-elles.\n\nN\'hésitez pas à consulter <a href=\'https://alan.com/app/helpV2/a_Mon-remboursement-na-pas-ete-effectue-pour-mes-frais-dhopital\' target="_blank">cet article de notre centre d\'aide</a> pour y voir plus clair', laboratory_invoice: "🤗 Nous avons bien reçu votre facture d'analyses en laboratoire. Elle n'est pas nécessaire pour vous rembourser ! En effet, nous remboursons les actes de laboratoire à réception des informations de la Sécurité Sociale.\n\n👉 Si la télétransmission est active, nous devrions recevoir automatiquement ces informations, et nous pourrons calculer votre remboursement dans la foulée.\nSi la sécurité sociale vous a déjà remboursé et que la télétransmission tarde à arriver ou qu’elle est inactive, transférez-nous votre décompte Améli reprenant ces soins depuis votre application Alan.", titre_de_recette: "🤔 Le document indique que vous avez payé l'intégralité des soins.\n\n👉 Nous vous invitons à envoyer ce document à votre Caisse de Sécurité sociale afin de vous faire rembourser la part qu'ils vous doivent.\n\nLes informations nécessaires à votre remboursement par Alan seront alors disponibles.", acupuncture: '😔 Nous ne pouvons pas couvrir ces soins.\n\n👉 En effet nous ne remboursons les séances d’acupuncture que lorsqu’elles sont dispensées par un médecin diplômé, une sage-femme ou un dentiste (et seulement dans le cas où votre couverture Alan inclut ce type de soin).\n\nTransférer une nouvelle version du document : https://alan.com/app/dashboard?documentUploader=claim', alternative_medicine_quantity_limit_reached: '🤔 Le document ne contient pas toutes les informations pour estimer votre remboursement.\n\n👉 Nous vous invitons à nous renvoyer une version contenant les informations suivantes :\n\n- Date à laquelle chaque soin a été réalisé\n\nTransférer une nouvelle version du document : https://alan.com/app/dashboard?documentUploader=claim', quittance_100: '🤔 Nous avons bien pris connaissance de votre document, et il semblerait que vous ayez réglé l’intégralité de la somme. Vous devez donc l’envoyer à la Sécurité sociale, elle en a besoin pour vous rembourser.\n\n👉 Une fois que nous aurons reçu les informations de remboursements de la part de la Sécu, nous serons en mesure de vous rembourser à notre tour.', hospi_addressed_alan: "🧐 Nous ne pouvons pas vous rembourser cette facture car celle-ci est adressée à Alan.\n\nCela signifie donc que vous n'avez rien réglé. L'établissement doit envoyer la facture à Almérys (notre partenaire de tiers-payant).\n\nNous vous invitons donc à contacter l'établissement afin de leur faire part de cette information.", estimation_already_sent: f'🤭 On dirait que notre estimation vous à déjà été transmis.
  👉 Si c’est une erreur, vous pouvez nous renvoyer le bon document. On s’en occupera rapidement!
  Transférer une nouvelle version du document : {HEALTH_DOCUMENT_UPLOAD_URL}
        ', almerys_PEC_rejected: "🤔 Ce document indique qu'une prise en charge a été effectuée auprès d'Almérys et que celle-ci a été rejetée.\n\n  👉 En effet, conformément au refus délivré par Almerys, votre forfait est épuisé et nous ne pouvons donc intervenir.\n\n  Si vous avez des questions au sujet de vos forfait, n'hésitez pas à nous contacter directement depuis votre espace Alan.", almerys_PEC_request: f'🤔 Ce document indique qu'une demande de prise en charge a été effectuée auprès d'Almérys. Ce document ne nous est pas destiné et ne nous permet pas de vous rembourser.

  👉 Si vous avez réglé des frais, nous vous invitons à nous adresser la facture acquittée.

  Transférer une nouvelle version du document : {HEALTH_DOCUMENT_UPLOAD_URL}', arret_de_travail: '🤭 Ce document ne semble pas être lié à un remboursement de soins.\n\n  👉 Si vous souhaitez demander le versement d\'indemnités complémentaires, vous pouvez ajouter ce document depuis la page principale de votre compte Alan sur ordinateur, en sélectionnant l\'onglet "Prévoyance" au centre de l\'écran.\n  Si vous ne voyez pas d’onglet “Prévoyance” sur votre espace, c’est peut-être que votre employeur n’a pas souscrit de contrat de Prévoyance avec nous, ou que le dossier d’arrêt de travail n’est pas encore ouvert : dans les deux cas, nous vous inviter à vérifier tout cela avec lui !\n  Si vous êtes en portabilité, nous vous invitons à nous contacter pour ouvrir votre dossier.', feuille_de_soins: '😑 Nous avons bien pris connaissance de votre feuille de soin. Il est important de l’envoyer uniquement à la Sécurité sociale, elle en a besoin pour vous rembourser.\n  👉 Une fois votre soin effectué et le remboursement de la Sécu fait, toutes les informations nécessaires seront disponibles pour qu’Alan vous rembourse. Plus besoin de nous envoyer de feuille de soin.', french_state_payment_attestation: " Nous avons bien reçu votre attestation de paiement. Elle n'est pas nécessaire pour vous rembourser ! Si ce n'est pas déjà fait, envoyez-nous le bordereau de facturation acquitté pour ces soins.\n\n  👉 Si vous avez reçu une demande de notre part pour valider que vous avez avancé les soins, merci de valider cela directement sur la page du soin correspondant dans l'application. Si vous n'avez reçu aucune demande de notre part, on est bons ! 👌", hospi_PEC_request: '🤔 Ce document est une demande de prise en charge pour des frais d’hospitalisation.\n\n  👉 Vous pouvez à tout moment et en toute autonomie effectuer votre demande depuis votre compte.\n  Depuis un ordinateur :\n  - Rendez-vous sur votre espace perso et cliquez en bas à gauche sur “Séjour à l’hôpital”.\n  - Une petite explication sur les conditions d’une prise en charge s’affiche, avec le bouton “Demander une prise en charge”.\n  - Cliquez dessus et remplissez le formulaire !\n  Depuis l’application mobile Alan :\n  - Depuis l’onglet “Demandes” (en bas de l’écran), cliquez sur “Séjour à l’hôpital”.', indemnite_journalieres: '🤭 Ce document ne semble pas être lié à un remboursement de soins.\n\n  👉 S\'il s’agit pour vous de justifier de l\'indemnisation de votre arrêt de travail par la Sécu, dans le but de demander le versement de vos indemnités complémentaires, vous pouvez ajouter ce document depuis la page principale de votre compte Alan sur ordinateur, en sélectionnant l\'onglet "Prévoyance" au centre de l\'écran.\n  Si vous ne voyez pas d’onglet “Prévoyance” sur votre espace depuis un ordinateur, c’est peut-être que votre employeur n’a pas souscrit de contrat de Prévoyance avec nous, ou qu\'il n\'a pas encore ouvert le dossier d’arrêt de travail : dans les deux cas, nous vous inviter à vérifier tout cela avec lui !\n\n  👉 S\'il s’agit de justifier du maintien de votre portabilité, vous pouvez ajouter ce document, toujours depuis la page principale de votre compte Alan sur ordinateur, en cliquant sur "Couverture - Ex-salarié".', payment_card_ticket: "🤗 Nous avons bien reçu votre ticket de carte bleue. Il n'est pas nécessaire pour vous rembourser ! Si ce n'est pas déjà fait et que le soin effectué n'est pas couvert par la Sécurité sociale, envoyez-nous la facture détaillée et acquittée.\n\n  👉 Si vous avez reçu une demande de notre part pour valider que vous avez avancé les soins, merci de valider cela directement sur la page du soin correspondant dans l'application. Si vous n'avez reçu aucune demande de notre part, on est bons ! 👌", RIB: ' On dirait que ce document n’a pas de rapport avec un remboursement de santé.\n\n  👉 S’il s’agit de mettre à jour votre RIB, vous pouvez le faire directement depuis votre compte:\n  - Cliquez sur votre nom tout en haut à droite de l’écran puis sur Mon compte\n  - Cliquez sur Coordonnées bancaires au centre de l’écran\n  - Cliquez ensuite sur Modifier pour modifier votre RIB.', PEC_request: " Ce document est une demande de prise en charge qui doit être transmise à notre partenaire de tiers-payant.\n\n  👉 Merci d'indiquer à votre professionnel de santé de s’adresser directement à Almerys.", generic_not_usable: f'🤭 On dirait que ce document n’a pas de rapport avec un remboursement de santé.

  👉 Si c’est une erreur, vous pouvez nous renvoyer le bon document. On s’en occupera rapidement !

  Transférer une nouvelle version du document : {HEALTH_DOCUMENT_UPLOAD_URL}', attestation_perimee: f'👉 Nous vous invitons à nous renvoyer une attestation en cours de validité (datée de moins de 3 mois).

  Transférer une nouvelle version du document : {HEALTH_DOCUMENT_UPLOAD_URL}', ss_attestation_incomplete: f'🤭 Le document ne contient pas toutes les informations pour que nous puissions le traiter. Il nous manque des informations concernant le parent de référence.

  👉 Nous vous invitons à nous renvoyer une version mentionnant le parent de référence ainsi que les bénéficiaires qui lui sont rattachés.

  Transférer une nouvelle version du document : {HEALTH_DOCUMENT_UPLOAD_URL}'}}
SUB_REJECTION_LABEL module-attribute 
SUB_REJECTION_LABEL = {
    beneficiary_name_missing: "Le nom du bénéficiaire n'est pas spécifié",
    beneficiary_name_not_in_policy: "Le nom du bénéficiaire n'est pas couvert par le contrat",
    beneficiary_name_not_matching: "Le nom du bénéficiaire ne correspond pas à celui enregistré sur le contrat",
    health_professional_name_missing: "Le nom du professionnel de santé n'est pas spécifié",
    health_professional_name_exact_name_unknown: "Le nom exact du professionnel de santé n'est pas spécifié",
    health_professional_adeli_missing: "Le Nº ADELI/RPPS n'est pas spécifié",
    health_professional_adeli_not_readable: "Le Nº ADELI/RPPS n'est pas complet/lisible",
    health_professional_signature_missing: "La signature n'est pas présente",
    health_professional_signature_stamp_missing: "Le tampon n'est pas présent",
    care_type_unknown: "Le type de soin n'est pas spécifié",
    care_act_price_not_separated: "Le prix de chaque soin n'est pas indiqué séparément",
    care_act_missing: "Le code de regroupement pour chacun des actes mentionnés n'est pas spécifié",
    care_act_optic_equipment_missing: "Le code de regroupement pour chacun des éléments de l’équipement optique n’est pas spécifié",
    care_act_expired_code: "Certains codes de regroupements ou codes actes apparaissant dans le document sont périmés et ne doivent plus être utilisés",
    care_act_not_specified: "Le détail des actes envisagés n'est pas spécifié",
    care_act_eyes_missing: "Oeil(s) non spécifié(s)",
    quote_date_missing: "La date n'est pas spécifiée",
    quote_date_invalid: "La date n'est pas valide",
    quote_date_expired: "Le devis a expiré",
    brss_missing: "La BRSS de chaque soin n'est pas spécifiée",
    brss_not_separated: "La BRSS de chaque soin n'est pas indiquée séparément",
    exceeding_fees_not_separated: "Les dépassements d'honoraires de chaque soin ne sont pas indiqués séparément",
    surco_quote_missing: "Le devis initial avec les bases des actes envisagés est manquant",
}

rejection

CATEGORY_REJECTION_REASONS module-attribute 
CATEGORY_REJECTION_REASONS = {
    prescription: [
        unreadable,
        unusable_partial_document,
        duplicate,
        beneficiary_not_covered,
        insufficent_info_for_parsing,
        non_EU_prescription,
    ],
    mutuelle_decompte: [
        unreadable,
        unusable_partial_document,
        duplicate,
        beneficiary_not_covered,
        insufficent_info_for_parsing,
        other_insurer_decompte_100,
    ],
    ss_decompte: [
        unreadable,
        unusable_partial_document,
        duplicate,
        beneficiary_not_covered,
        cares_received_abroad,
        insufficent_info_for_parsing,
        secu_decompte_100,
    ],
    invoice: [
        unreadable,
        unusable_partial_document,
        duplicate,
        ask_paid_invoice,
        beneficiary_not_covered,
        cares_received_abroad,
        insufficent_info_for_parsing,
        missing_mt_or_dmt,
        etiopathe,
        kine_invoice_without_mezieres,
        hospital_receipt,
        laboratory_invoice,
        titre_de_recette,
        acupuncture,
        alternative_medicine_quantity_limit_reached,
        quittance_100,
        hospi_addressed_alan,
    ],
    mutuelle_no_coverage_attestation: [
        unreadable,
        unusable_partial_document,
        duplicate,
        beneficiary_not_covered,
        insufficent_info_for_parsing,
    ],
    birth_or_adoption_certificate: [
        unreadable,
        unusable_partial_document,
        duplicate,
        beneficiary_not_covered,
        insufficent_info_for_parsing,
    ],
    quote: [
        unreadable,
        unusable_partial_document,
        duplicate,
        beneficiary_not_covered,
        cares_received_abroad,
        insufficent_info_for_parsing,
        missing_mt_or_dmt,
        etiopathe,
        acupuncture,
        estimation_already_sent,
    ],
    unsupported: [
        unreadable,
        duplicate,
        beneficiary_not_covered,
        insufficent_info_for_parsing,
        almerys_PEC_rejected,
        almerys_PEC_request,
        arret_de_travail,
        feuille_de_soins,
        french_state_payment_attestation,
        hospi_PEC_request,
        indemnite_journalieres,
        payment_card_ticket,
        RIB,
        PEC_request,
        generic_not_usable,
        laboratory_invoice,
    ],
    non_claims_prescription: DEFAULT_REJECTION_REASONS,
    medical_certificate: DEFAULT_REJECTION_REASONS,
    medical_imaging: DEFAULT_REJECTION_REASONS,
    medical_results: DEFAULT_REJECTION_REASONS,
    cancel_teletransmission_request: DEFAULT_REJECTION_REASONS,
    other_health: [
        unreadable,
        duplicate,
        beneficiary_not_covered,
        insufficent_info_for_parsing,
        almerys_PEC_rejected,
        almerys_PEC_request,
        arret_de_travail,
        feuille_de_soins,
        french_state_payment_attestation,
        hospi_PEC_request,
        indemnite_journalieres,
        payment_card_ticket,
        RIB,
        PEC_request,
        generic_not_usable,
    ],
    ss_attestation: [
        unreadable,
        unusable_partial_document,
        duplicate,
        beneficiary_not_covered,
        insufficent_info_for_parsing,
        attestation_perimee,
        ss_attestation_incomplete,
    ],
}
DEFAULT_REJECTION_REASONS module-attribute 
DEFAULT_REJECTION_REASONS = [
    unreadable,
    unusable_partial_document,
    duplicate,
    beneficiary_not_covered,
    insufficent_info_for_parsing,
    non_EU_prescription,
]
get_rejection_reasons_per_category 
get_rejection_reasons_per_category(document_category)

Get the list of rejection reason for a document category with the default message we want to send to the user return: list[RejectionReason]

Source code in components/fr/public/document_parsing/queries/rejection.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def get_rejection_reasons_per_category(
    document_category: ClaimInsuranceDocumentCategory,
) -> list[RejectionReason]:
    """
    Get the list of rejection reason for a document category with the default message we want to send to the user
    return: list[RejectionReason]
    """
    reason_keys = CATEGORY_REJECTION_REASONS.get(document_category, [])
    return [
        RejectionReason(
            reason=key, defaultMessage=REJECTION_BODY_FOR_USER["default_message"][key]
        )
        for key in reason_keys
    ]
get_sub_rejection_reasons_per_category 
get_sub_rejection_reasons_per_category(document_category)

Get the list of sub rejection reason for a document category For now we only return for quote return: dict[str, list[RejectionReason]] | None

Source code in components/fr/public/document_parsing/queries/rejection.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def get_sub_rejection_reasons_per_category(
    document_category: ClaimInsuranceDocumentCategory,
) -> dict[str, list[RejectionReason]] | None:
    """
    Get the list of sub rejection reason for a document category
    For now we only return for quote
    return: dict[str, list[RejectionReason]] | None
    """
    if document_category == ClaimInsuranceDocumentCategory.quote:
        return {
            DocumentRejectionReason.insufficent_info_for_parsing: [
                RejectionReason(reason=key, defaultMessage=SUB_REJECTION_LABEL[key])
                for key in DocumentSubRejectionReason
            ]
        }
    return None

components.fr.public.dsn

company

Set of functions to query information about companies present in the DSN.

The DSN data model is composed of DSN "companies" and "establishments". A "company" in the context of the module does NOT refer specifically to a DSN "company" but to any entity (usually an "establishment").

get_companies_from_sirens 

get_companies_from_sirens(sirens)

Fetch companies for the given SIRENs.

Source code in components/fr/public/dsn/company.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
def get_companies_from_sirens(sirens: Sequence[str]) -> list[DSNCompanyData]:
    """
    Fetch companies for the given SIRENs.
    """
    from components.fr.internal.dsn.models.dsn_company import DsnCompany
    from components.fr.internal.dsn.models.dsn_establishment import DsnEstablishment

    rows = current_session.execute(
        select(
            DsnCompany.siren,
            DsnEstablishment.nic,
            DsnEstablishment.name,
            DsnEstablishment.postal_code,
            DsnEstablishment.city,
            DsnEstablishment.apet,
        )
        .select_from(DsnEstablishment)
        .join(DsnEstablishment.dsn_company)
        .where(DsnCompany.siren.in_(sirens), DsnEstablishment.nic.is_not(None))
        .order_by(DsnCompany.siren, DsnEstablishment.name, DsnEstablishment.postal_code)
    ).all()

    return [
        DSNCompanyData(
            siren=row.siren,
            nic=row.nic,
            name=row.name,
            postal_code=row.postal_code,
            city=row.city,
            ape=row.apet,
        )
        for row in rows
    ]

get_company_from_siret 

get_company_from_siret(siret)

Fetch the company based on its SIRET.

Source code in components/fr/public/dsn/company.py 
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
def get_company_from_siret(siret: str) -> DSNCompanyData | None:
    """
    Fetch the company based on its SIRET.
    """
    from components.fr.internal.dsn.models.dsn_company import DsnCompany
    from components.fr.internal.dsn.models.dsn_establishment import DsnEstablishment

    if len(siret) != 14:
        raise ValueError(f"SIRET must be 14 characters long, got '{siret}'")
    siren = siret[:9]
    nic = siret[9:]

    result = (
        current_session.query(DsnCompany, DsnEstablishment)  # noqa: ALN085
        .select_from(DsnEstablishment)
        .join(DsnCompany)
        .filter(
            DsnCompany.siren == siren,
            DsnEstablishment.nic == nic,
        )
        # Assumption: the SIRET refers to a unique DSN Establishment
        .one_or_none()
    )

    if not result:
        return None

    dsn_company, dsn_establishment = result

    return DSNCompanyData(
        siren=siren,
        nic=nic,
        name=dsn_establishment.name,
        ape=dsn_establishment.apet,
        postal_code=dsn_establishment.postal_code,
        city=dsn_establishment.city,
    )

contract_download

get_contracts_download_zip_for_companies 

get_contracts_download_zip_for_companies(company_ids)

This function is used to download the contracts (both PDF and XML, in ZIP) for a given companies.

Parameters:

Name Type Description Default
company_ids list[int] | set[int]

List of company IDs to download contracts for

 required

Returns:

Type Description
tuple[dict[str, str] | None, bool]

tuple[dict[str, str] | None, bool]: Zip attachment and boolean indicating if any FDP was found
Source code in components/fr/public/dsn/contract_download.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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get_contracts_download_zip_for_companies(
    company_ids: list[int] | set[int],
) -> tuple[dict[str, str] | None, bool]:
    """
    This function is used to download the contracts (both PDF and XML, in ZIP) for a given companies.

    Args:
        company_ids: List of company IDs to download contracts for

    Returns:
        tuple[dict[str, str] | None, bool]: Zip attachment and boolean indicating if any FDP was found
    """
    attachments: dict[str, IO[Any]] = {}
    has_fdp = False

    companies = current_session.scalars(
        select(Company).filter(Company.id.in_(company_ids))
    ).all()

    for company in companies:
        company_folder = f"{company.display_name}"

        # Fdp covers all health/prev contracts of the company, in case of multiple contracts which contract we use to get the fdp is irrelevant
        health_contracts = [
            contract
            for contract in company.contracts
            if contract.status == ContractStatus.active
        ]
        if not health_contracts:
            continue

        latest_health_fdp = None
        for contract in health_contracts:
            if contract.latest_fiche_de_parametrage:
                latest_health_fdp = contract.latest_fiche_de_parametrage
                break

        prevoyance_contracts = [
            contract
            for contract in company.prevoyance_contracts
            if contract.status == ContractStatus.active
        ]

        latest_prevoyance_fdp = None
        for contract in prevoyance_contracts:  # type: ignore[assignment]
            if contract.latest_fiche_de_parametrage:
                latest_prevoyance_fdp = contract.latest_fiche_de_parametrage
                break

        if latest_health_fdp:
            has_fdp = True
            _add_attachments(attachments, company_folder, latest_health_fdp.id)
        if latest_prevoyance_fdp:
            has_fdp = True
            _add_attachments(attachments, company_folder, latest_prevoyance_fdp.id)

    zip_attachment: dict[str, str] | None = (
        zip_as_attachment(
            attachments,
            "Alan - Instructions et fiches de paramétrage DSN.zip"
            if has_fdp
            else "Alan - Instructions.zip",
        )
        if attachments
        else None
    )

    return zip_attachment, has_fdp

entities

DSNCompanyData dataclass 

DSNCompanyData(siren, nic, name, postal_code, city, ape)

Bases: DataClassJsonMixin

The data about a company, from the DSN establishments data.

ape instance-attribute 
ape

The APE code of the company.

city instance-attribute 
city
name instance-attribute 
name

The entity's "enseigne" from the DSN.

nic instance-attribute 
nic
postal_code instance-attribute 
postal_code
siren instance-attribute 
siren
siret property 
siret

The SIRET of the company.

components.fr.public.employees

employees

get_employees_for_admin_dashboard 

get_employees_for_admin_dashboard(
    company_ids,
    scope_ids,
    cursor,
    limit,
    search,
    employee_types,
    employment_types,
    coverage_statuses,
    professional_categories,
    ccn_ids,
    status_details,
    coverage_status_details=None,
    mandatory_coverages=None,
)

Get employees for the admin dashboard. It won't make any admin rights checks, please make sure the user has the rights to see the given companies & scopes. :param company_ids: The companies to fetch the employees from, it will fetch all employees from the companies, except if scopes for those companies are also given :param scope_ids: The scopes to fetch the employees from, if some scopes belong to given companies ids, only employees from those scopes will be fetched :param cursor: The cursor to handle pagination :param limit: The limit of employees to fetch :param search: The search string to filter employees :param employee_types: The employee types to filter :param employment_types: The employment types to filter :param coverage_statuses: The coverage statuses to filter :param professional_categories: The professional categories to filter :param ccn_ids: The CCN ids to filter :param mandatory_coverages: The mandatory coverages to filter (true = mandatory, false = optional) :return: The paginated employees, filtered by given filters

Source code in components/fr/public/employees/employees.py 
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
68
69
70
71
72
73
74
75
76
77
78
79
def get_employees_for_admin_dashboard(
    company_ids: Iterable[int],
    scope_ids: Iterable[uuid.UUID],
    cursor: int,
    limit: int,
    search: str | None,
    employee_types: set[EmployeeQueryBuilderEmployeeType] | None,
    employment_types: set[EmploymentType] | None,
    coverage_statuses: set[CoverageStatus] | None,
    professional_categories: set[ProfessionalCategory | None] | None,
    ccn_ids: set[int | None] | None,
    status_details: set[StatusDetail] | None,
    coverage_status_details: set[StatusDetail] | None = None,
    mandatory_coverages: set[bool] | None = None,
) -> PaginatedEmployeesForAdminDashboard:
    """
    Get employees for the admin dashboard.
    It won't make any admin rights checks, please make sure the user has the rights to see the given companies & scopes.
    :param company_ids: The companies to fetch the employees from, it will fetch all employees from the companies, except if scopes for those companies are also given
    :param scope_ids: The scopes to fetch the employees from, if some scopes belong to given companies ids, only employees from those scopes will be fetched
    :param cursor: The cursor to handle pagination
    :param limit: The limit of employees to fetch
    :param search: The search string to filter employees
    :param employee_types: The employee types to filter
    :param employment_types: The employment types to filter
    :param coverage_statuses: The coverage statuses to filter
    :param professional_categories: The professional categories to filter
    :param ccn_ids: The CCN ids to filter
    :param mandatory_coverages: The mandatory coverages to filter (true = mandatory, false = optional)
    :return: The paginated employees, filtered by given filters
    """
    nics_by_company_id, entity_codes_by_company_id = get_scope_filtering_values(
        scope_ids=scope_ids
    )
    all_company_ids = (
        set(company_ids)
        | set(nics_by_company_id.keys())
        | set(entity_codes_by_company_id.keys())
    )

    return internal_get_employees_for_admin_dashboard(
        all_company_ids,
        cursor,
        limit,
        nics_by_company_id,
        entity_codes_by_company_id,
        search,
        employee_types,
        employment_types,
        coverage_statuses,
        professional_categories,
        ccn_ids,
        status_details,
        coverage_status_details,
        mandatory_coverages,
    )

terminated_employees

get_terminated_employee_details 

get_terminated_employee_details(user_id, company_id)
Source code in components/fr/public/employees/terminated_employees.py 
19
20
21
22
def get_terminated_employee_details(  # noqa: D103
    user_id: str, company_id: int
) -> TerminatedEmployeeDetails:
    return internal_get_terminated_employee_details(user_id, company_id)

get_terminated_employees 

get_terminated_employees(
    company_ids,
    operational_scope_ids,
    cursor,
    limit,
    search,
    terminated_employee_types,
)
Source code in components/fr/public/employees/terminated_employees.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
def get_terminated_employees(  # noqa: D103
    company_ids: Iterable[int],
    operational_scope_ids: Iterable[UUID],
    cursor: int,
    limit: int,
    search: str | None,
    terminated_employee_types: set[TerminatedEmployeeType] | None,
) -> PaginatedTerminatedEmployeesForAdminDashboard:
    from components.fr.internal.operational_scopes.business_logic.queries import (
        get_scope_filtering_values,
        get_scope_filtering_values_for_user,
    )

    nics_by_company_id, entity_codes_by_company_id = (
        get_scope_filtering_values(operational_scope_ids)
        if operational_scope_ids
        else get_scope_filtering_values_for_user(g.current_user.id, set(company_ids))
    )
    all_company_ids = (
        set(company_ids)
        | set(nics_by_company_id.keys())
        | set(entity_codes_by_company_id.keys())
    )

    return internal_get_terminated_employees(
        all_company_ids,
        cursor,
        limit,
        search,
        terminated_employee_types=terminated_employee_types,
        nics_by_company_id=nics_by_company_id,
        entity_codes_by_company_id=entity_codes_by_company_id,
    )

get_terminated_employees_counts 

get_terminated_employees_counts(
    company_ids, operational_scope_ids, search=None
)
Source code in components/fr/public/employees/terminated_employees.py 
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
def get_terminated_employees_counts(  # noqa: D103
    company_ids: Iterable[int],
    operational_scope_ids: Iterable[UUID],
    search: str | None = None,
) -> TerminatedEmployeesByTypeCounts:
    from components.fr.internal.operational_scopes.business_logic.queries import (
        get_scope_filtering_values,
        get_scope_filtering_values_for_user,
    )

    nics_by_company_id, entity_codes_by_company_id = (
        get_scope_filtering_values(operational_scope_ids)
        if operational_scope_ids
        else get_scope_filtering_values_for_user(g.current_user.id, set(company_ids))
    )
    all_company_ids = (
        set(company_ids)
        | set(nics_by_company_id.keys())
        | set(entity_codes_by_company_id.keys())
    )

    return internal_get_terminated_employees_counts(
        company_ids=all_company_ids,
        nics_by_company_id=nics_by_company_id,
        entity_codes_by_company_id=entity_codes_by_company_id,
        search=search,
    )

components.fr.public.employment

admin_resolvers

dsn_removal_suggestion_resolver

DsnRemovalSuggestionResolver

Bases: AdminErrorResolver

Resolver for DSN removal suggestion blocked movements.

Admins can either confirm the removal or cancel it.

apply_resolution 
apply_resolution(blocked_movement, action, params=None)

Apply the admin's resolution choice.

Parameters:

Name Type Description Default
blocked_movement CoreBlockedMovement | UpstreamBlockedMovement

The blocked movement to resolve

 required
action AdminResolutionAction

cancel (ignore suggestion) or apply (remove the employee)

 required
params dict[str, Any] | None

Not used — no overrides needed for DSN removal suggestions.

 None
Source code in components/fr/public/employment/admin_resolvers/dsn_removal_suggestion_resolver.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
@override
def apply_resolution(
    self,
    blocked_movement: CoreBlockedMovement | UpstreamBlockedMovement,
    action: AdminResolutionAction,
    params: dict[str, Any] | None = None,
) -> None:
    """
    Apply the admin's resolution choice.

    Args:
        blocked_movement: The blocked movement to resolve
        action: cancel (ignore suggestion) or apply (remove the employee)
        params: Not used — no overrides needed for DSN removal suggestions.
    """
    core_bm = mandatory_type(
        expected_type=CoreBlockedMovement,
        value=blocked_movement,
    )

    if core_bm.status == BlockedMovementStatus.resolved:
        current_logger.info(
            "Blocked movement already resolved, skipping",
            layer="apply_resolution",
            clear_attributes={"blocked_movement_id": str(core_bm.id)},
        )
        return

    if action == AdminResolutionAction.cancel:
        manually_cancel_core_blocked_movement(
            core_blocked_movement_id=core_bm.id,
            commit=True,
        )
    elif action == AdminResolutionAction.apply:
        source_rules_override = replace(
            get_source_rules(SourceType.fr_dsn_removal_suggestion),
            terminate=RuleCaseAction.apply,
            terminate_with_end_date_more_than_100_days_in_the_past=RuleCaseAction.apply,
        )
        retry_core_blocked_movement(
            blocked_movement_id=core_bm.id,
            source_rules_override=source_rules_override,
            commit=True,
        )
    else:
        assert_never(action)
can_resolve 
can_resolve(blocked_movement)

Only core blocked movements from DSN removal suggestions with a termination rule case.

Source code in components/fr/public/employment/admin_resolvers/dsn_removal_suggestion_resolver.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@override
def can_resolve(
    self,
    blocked_movement: CoreBlockedMovement | UpstreamBlockedMovement,
) -> bool:
    """Only core blocked movements from DSN removal suggestions with a termination rule case."""
    if isinstance(blocked_movement, UpstreamBlockedMovement):
        return False

    if blocked_movement.source_type != SourceType.fr_dsn_removal_suggestion:
        return False

    if not self.is_termination(blocked_movement):
        return False

    return super().can_resolve(blocked_movement)
category property 
category

Category for grouping in admin reporting alerts.

error_code property 
error_code
get_resolution_context 
get_resolution_context(blocked_movement)

Extract data needed for admin resolution modal.

Returns employee info, event date, and termination reason — same data as displayed on the fixAdminInputRequired page.

Source code in components/fr/public/employment/admin_resolvers/dsn_removal_suggestion_resolver.py 
 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
123
124
125
126
127
128
129
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
@override
def get_resolution_context(
    self,
    blocked_movement: CoreBlockedMovement | UpstreamBlockedMovement,
) -> AdminResolutionContext:
    """
    Extract data needed for admin resolution modal.

    Returns employee info, event date, and termination reason — same data
    as displayed on the fixAdminInputRequired page.
    """
    core_bm = mandatory_type(
        expected_type=CoreBlockedMovement,
        value=blocked_movement,
    )

    employment_declaration = self._extract_employment_declaration(core_bm)
    end_date = employment_declaration.end_date
    if not end_date:
        raise AdminResolverInternalError(
            "Missing end_date in employment declaration for DSN removal suggestion.",
            blocked_movement_id=core_bm.id,
            resolver_class=self.__class__.__name__,
        )

    user = get_resource_or_none(User, int(core_bm.user_id))
    company = get_resource_or_none(Company, int(core_bm.company_id))

    employee_full_name = ""
    employee_email = ""
    if user:
        profile_service = ProfileService.create()
        profile = profile_service.get_or_raise_profile(user.profile_id)
        employee_full_name = f"{profile.first_name} {profile.last_name}".strip()
        employee_email = profile.email or ""

    ssn_or_ntt = ""
    if user and user.insurance_profile:
        ssn_or_ntt = user.insurance_profile.ssn or user.insurance_profile.ntt or ""

    infinitely_valid_extended_informations = next(
        (
            ei
            for ei in employment_declaration.extended_informations
            if ei.validity_period is None
        ),
        None,
    )

    termination_reason = (
        infinitely_valid_extended_informations.values.get("termination_type", "")
        if infinitely_valid_extended_informations
        else ""
    )

    return create_admin_resolution_context(
        blocked_movement_id=core_bm.id,
        error_code=self.error_code,
        source_type=core_bm.source_type,
        employee_full_name=employee_full_name,
        employee_email=employee_email,
        company_display_name=company.display_name if company else "",
        created_at=core_bm.created_at.isoformat(),
        employee_identifier=ssn_or_ntt,
        event_date=end_date.isoformat(),
        motif_fin_contrat=termination_reason,
    )
is_termination 
is_termination(blocked_movement)

Check that the action_not_allowed is indeed a termination.

Source code in components/fr/public/employment/admin_resolvers/dsn_removal_suggestion_resolver.py 
65
66
67
68
69
70
def is_termination(self, blocked_movement: CoreBlockedMovement) -> bool:
    """Check that the action_not_allowed is indeed a termination."""
    not_allowed_rule_cases = blocked_movement.context.get(
        "not_allowed_rule_cases", []
    )
    return any(rc in self._TERMINATION_RULE_CASES for rc in not_allowed_rule_cases)

missing_ssn_ntt_resolver

MissingSsnNttResolutionParams

Bases: TypedDict

Parameters for resolving a missing SSN/NTT blocked movement.

All fields are optional for cancel action. For apply action, both inputType and inputValue are required.

Expected keys
  • inputType: Literal["ssn", "ntt"] - Type of identifier to add
  • inputValue: str - The SSN or NTT value
inputType instance-attribute 
inputType
inputValue instance-attribute 
inputValue
MissingSsnNttResolver

Bases: AdminErrorResolver

Resolver for missing required SSN or NTT blocked movements.

Missing required SSN or NTT occur when an employee declaration has no SSN or NTT. Admins can resolve these by: - Canceling the declaration - Applying with either an SSN or NTT override (mutually exclusive)

apply_resolution 
apply_resolution(blocked_movement, action, params=None)

Apply the admin's resolution choice to the missing required SSN or NTT blocked movement.

Parameters:

Name Type Description Default
blocked_movement CoreBlockedMovement | UpstreamBlockedMovement

The blocked movement to resolve

 required
action AdminResolutionAction

The action chosen by the admin (cancel or apply)

 required
params dict[str, Any] | None

Optional parameters containing: - inputType: "ssn" or "ntt" - inputValue: the validated SSN or NTT string

 None

Raises:

Type Description
ResolutionActionError

If the action is invalid or the resolution cannot be applied
Source code in components/fr/public/employment/admin_resolvers/missing_ssn_ntt_resolver.py 
147
148
149
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
178
@override
def apply_resolution(
    self,
    blocked_movement: CoreBlockedMovement | UpstreamBlockedMovement,
    action: AdminResolutionAction,
    params: dict[str, Any] | None = None,
) -> None:
    """Apply the admin's resolution choice to the missing required SSN or NTT blocked movement.

    Args:
        blocked_movement: The blocked movement to resolve
        action: The action chosen by the admin (cancel or apply)
        params: Optional parameters containing:
            - inputType: "ssn" or "ntt"
            - inputValue: the validated SSN or NTT string

    Raises:
        ResolutionActionError: If the action is invalid or the resolution cannot be applied
    """
    self.validate_params(action, params)

    upstream_blocked_movement = mandatory_type(
        expected_type=UpstreamBlockedMovement,
        value=blocked_movement,
    )

    if action == AdminResolutionAction.cancel:
        _cancel_resolution(upstream_blocked_movement)
    elif action == AdminResolutionAction.apply:
        _apply_resolution(upstream_blocked_movement, params)
    else:
        assert_never(action)
can_resolve 
can_resolve(blocked_movement)

Check if this resolver can handle the given blocked movement.

Only upstream blocked movements can be resolved.

Parameters:

Name Type Description Default
blocked_movement CoreBlockedMovement | UpstreamBlockedMovement

The blocked movement to check

 required

Returns:

Type Description
bool

True if this is an upstream blocked movement, False otherwise
Source code in components/fr/public/employment/admin_resolvers/missing_ssn_ntt_resolver.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@override
def can_resolve(
    self,
    blocked_movement: CoreBlockedMovement | UpstreamBlockedMovement,
) -> bool:
    """Check if this resolver can handle the given blocked movement.

    Only upstream blocked movements can be resolved.

    Args:
        blocked_movement: The blocked movement to check

    Returns:
        True if this is an upstream blocked movement, False otherwise
    """
    if isinstance(blocked_movement, CoreBlockedMovement):
        return False

    return super().can_resolve(blocked_movement)
category property 
category

Category for grouping in admin reporting alerts.

error_code property 
error_code

The error code this resolver handles.

get_resolution_context 
get_resolution_context(blocked_movement)

Return resolution data for a missing SSN/NTT blocked movement.

Extracts employee information and defines display fields for the admin.

Parameters:

Name Type Description Default
blocked_movement CoreBlockedMovement | UpstreamBlockedMovement

The blocked movement to get context for

 required

Returns:

Type Description
AdminResolutionContext

AdminResolutionContext with employee and company information

Raises:

Type Description
ResolutionActionError

If blocked movement is missing required data
Source code in components/fr/public/employment/admin_resolvers/missing_ssn_ntt_resolver.py 
 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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
@override
def get_resolution_context(
    self,
    blocked_movement: CoreBlockedMovement | UpstreamBlockedMovement,
) -> AdminResolutionContext:
    """Return resolution data for a missing SSN/NTT blocked movement.

    Extracts employee information and defines display fields for the admin.

    Args:
        blocked_movement: The blocked movement to get context for

    Returns:
        AdminResolutionContext with employee and company information

    Raises:
        ResolutionActionError: If blocked movement is missing required data
    """
    upstream_blocked_movement = mandatory_type(
        expected_type=UpstreamBlockedMovement,
        value=blocked_movement,
    )

    if upstream_blocked_movement.company_id is None:
        raise ResolutionActionError(
            f"Blocked movement {upstream_blocked_movement.id} has no company_id"
        )

    company_info = get_resource_or_none(
        Company, int(upstream_blocked_movement.company_id)
    )

    upstream_data_record = mandatory(
        upstream_blocked_movement.upstream_data["record"],
        message="Upstream data record is required",
    )

    first_name = upstream_data_record.get("first_name", "")
    last_name = upstream_data_record.get("last_name", "")
    employee_full_name = f"{first_name} {last_name}".strip()

    return create_admin_resolution_context(
        blocked_movement_id=upstream_blocked_movement.id,
        error_code=self.error_code,
        source_type=upstream_blocked_movement.source_type,
        employee_full_name=employee_full_name,
        employee_email=upstream_data_record.get("email", ""),
        company_display_name=company_info.display_name if company_info else "",
        created_at=str(upstream_blocked_movement.created_at),
        employee_external_identifier=upstream_data_record.get("external_id"),
        effective_start_date=str(upstream_data_record.get("start_date", "")),
        employee_identifier=upstream_data_record.get("ssn", ""),
    )

no_eligible_contract_found_resolver

AvailableContractInfo

Bases: TypedDict

Info about an available contract for resolution context.

ccn_code instance-attribute 
ccn_code
contract_id instance-attribute 
contract_id
professional_category instance-attribute 
professional_category
start_date instance-attribute 
start_date
NoEligibleContractFoundResolver

Bases: AdminErrorResolver

Resolver for no_eligible_contract_found blocked movements.

This error occurs when attempting to affiliate an employee to a contract that doesn't exist or doesn't match the employee's data (date, category, CCN).

apply_resolution 
apply_resolution(blocked_movement, action, params=None)

Apply the admin's resolution choice.

Parameters:

Name Type Description Default
blocked_movement CoreBlockedMovement | UpstreamBlockedMovement

The blocked movement to resolve

 required
action AdminResolutionAction

cancel or apply

 required
params dict[str, Any] | None

Optional overrides for date, category, or CCN. Expected keys: - start_date_override: ISO date string for new start date - professional_category_override: Professional category string - ccn_code_override: CCN code string

 None
Source code in components/fr/public/employment/admin_resolvers/no_eligible_contract_found_resolver.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
@override
def apply_resolution(
    self,
    blocked_movement: CoreBlockedMovement | UpstreamBlockedMovement,
    action: AdminResolutionAction,
    params: dict[str, Any] | None = None,
) -> None:
    """
    Apply the admin's resolution choice.

    Args:
        blocked_movement: The blocked movement to resolve
        action: cancel or apply
        params: Optional overrides for date, category, or CCN.
                Expected keys:
                - start_date_override: ISO date string for new start date
                - professional_category_override: Professional category string
                - ccn_code_override: CCN code string
    """
    core_blocked_movement = mandatory_type(
        expected_type=CoreBlockedMovement,
        value=blocked_movement,
    )

    if action == AdminResolutionAction.cancel:
        manually_cancel_core_blocked_movement(
            core_blocked_movement_id=core_blocked_movement.id,
            commit=True,
        )
    elif action == AdminResolutionAction.apply:
        validated_params = self._extract_and_sanitize_params(params)
        if validated_params is None:
            raise AdminResolverInternalError(
                "expected validated params for apply action, but got none. This is likely because no params were provided in the request.",
                blocked_movement_id=core_blocked_movement.id,
                resolver_class=self.__class__.__name__,
            )

        employment_declaration = self._extract_employment_declaration(
            core_blocked_movement
        )

        start_date_override = validated_params.get("start_date")
        ccn_code_override = validated_params.get("ccn_code")
        professional_category_override = validated_params.get(
            "professional_category"
        )

        if start_date_override:
            employment_declaration = override_employment_declaration_start_date(
                employment_declaration, start_date_override
            )

        if ccn_code_override or professional_category_override:
            employment_declaration = self._override_extended_information(
                employment_declaration,
                professional_category_override=professional_category_override,
                ccn_code_override=ccn_code_override,
            )

        population_info = self._extract_population_info_from_declaration(
            employment_declaration, employment_declaration.start_date
        )

        final_ccn_code = ccn_code_override or population_info.ccn_code or ""
        final_professional_category = (
            professional_category_override or population_info.professional_category
        )

        if not self._contract_exists(
            company_id=int(core_blocked_movement.company_id),
            on_date=employment_declaration.start_date,
            ccn_code=final_ccn_code,
            professional_category=final_professional_category,
        ):
            raise ResolutionAttemptError(
                f"Still no contract found for blocked movement {core_blocked_movement.id} with given overrides",
                error_code=EmploymentInputErrorCode.retry_failed_given_admin_overrides.value,
            )

        retry_core_blocked_movement(
            blocked_movement_id=core_blocked_movement.id,
            employment_declaration_override=employment_declaration,
            commit=True,
        )
    else:
        assert_never(action)
can_resolve 
can_resolve(blocked_movement)

Only core blocked movements can be resolved.

Source code in components/fr/public/employment/admin_resolvers/no_eligible_contract_found_resolver.py 
123
124
125
126
127
128
129
130
131
132
@override
def can_resolve(
    self,
    blocked_movement: CoreBlockedMovement | UpstreamBlockedMovement,
) -> bool:
    """Only core blocked movements can be resolved."""
    if isinstance(blocked_movement, UpstreamBlockedMovement):
        return False

    return super().can_resolve(blocked_movement)
category property 
category

Category for grouping in admin reporting alerts.

error_code property 
error_code

The error code this resolver handles.

get_resolution_context 
get_resolution_context(blocked_movement)

Extract and format data needed for admin resolution.

Returns:

Type Description
AdminResolutionContext
  • Employee information (name, email, identifier)
AdminResolutionContext
  • Attempted values (date, category, CCN)
AdminResolutionContext
  • Company's available contracts
Source code in components/fr/public/employment/admin_resolvers/no_eligible_contract_found_resolver.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
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
178
179
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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
@override
def get_resolution_context(
    self,
    blocked_movement: CoreBlockedMovement | UpstreamBlockedMovement,
) -> AdminResolutionContext:
    """
    Extract and format data needed for admin resolution.

    Returns:
        - Employee information (name, email, identifier)
        - Attempted values (date, category, CCN)
        - Company's available contracts
    """
    core_blocked_movement: CoreBlockedMovement = mandatory_type(
        expected_type=CoreBlockedMovement,
        value=blocked_movement,
    )

    employment_declaration = self._extract_employment_declaration(
        core_blocked_movement
    )

    employment_start_date, is_valid_employment_start_date = (
        cast_to_date_with_validation(employment_declaration.start_date)
    )

    if not is_valid_employment_start_date or employment_start_date is None:
        raise AdminResolverInternalError(
            "Missing or invalid start_date in employment declaration, this should never happen.",
            blocked_movement_id=core_blocked_movement.id,
            employment_start_date=employment_declaration.start_date,
            is_valid_employment_start_date=is_valid_employment_start_date,
            resolver_class=self.__class__.__name__,
        )

    user = get_resource_or_none(User, int(core_blocked_movement.user_id))
    employee_full_name: str = user.full_name if user else ""

    infinitely_valid_extended_informations = first_or_none(
        employment_declaration.extended_informations,
        predicate=lambda update: update.validity_period is None,
    )

    employee_email = (
        infinitely_valid_extended_informations.values.get("invite_email", "")
        if infinitely_valid_extended_informations
        else ""
    )
    employee_identifier = (
        (
            infinitely_valid_extended_informations.values.get("ssn")
            or infinitely_valid_extended_informations.values.get("ntt")
            or ""
        )
        if infinitely_valid_extended_informations
        else ""
    )

    population_info = self._extract_population_info_from_declaration(
        employment_declaration, employment_start_date
    )
    ccn_code = population_info.ccn_code or ""
    professional_category = (
        population_info.professional_category.value
        if population_info.professional_category
        else ""
    )

    available_contracts = self._get_company_contracts(
        int(core_blocked_movement.company_id), employment_start_date
    )

    if not available_contracts:
        raise AdminResolverInternalError(
            "No available contracts found for blocked movement, this should never happen.",
            blocked_movement_id=core_blocked_movement.id,
            company_id=core_blocked_movement.company_id,
            employment_start_date=employment_start_date.isoformat(),
            resolver_class=self.__class__.__name__,
        )

    company = get_resource_or_none(Company, int(core_blocked_movement.company_id))

    context_data = {
        "available_contracts": available_contracts,
        "company_display_name": company.display_name if company else "",
        "company_id": core_blocked_movement.company_id,
        "ccn_code": ccn_code,
        "created_at": str(core_blocked_movement.created_at),
        "employee_email": employee_email,
        "employee_full_name": employee_full_name,
        "employee_identifier": employee_identifier,
        "employment_start_date": employment_start_date.isoformat(),
        "professional_category": professional_category,
    }

    return AdminResolutionContext(
        blocked_movement_id=core_blocked_movement.id,
        error_code=self.error_code,
        source_type=core_blocked_movement.source_type,
        context=context_data,
    )
ResolutionContextData

Bases: TypedDict

Data provided to admin for resolution decision.

available_contracts instance-attribute 
available_contracts
ccn_code instance-attribute 
ccn_code
company_display_name instance-attribute 
company_display_name
company_id instance-attribute 
company_id
created_at instance-attribute 
created_at
employee_email instance-attribute 
employee_email
employee_full_name instance-attribute 
employee_full_name
employee_identifier instance-attribute 
employee_identifier
employment_start_date instance-attribute 
employment_start_date
professional_category instance-attribute 
professional_category
SanitizedResolutionParams

Bases: TypedDict

Validated and extracted resolution parameters.

ccn_code instance-attribute 
ccn_code
professional_category instance-attribute 
professional_category
start_date instance-attribute 
start_date

fr_country_gateway

FrCountryGateway

Bases: CountryGateway[FrExtendedValues]

Implementation of the Employment Component's CountryGateway for France.

app_name class-attribute instance-attribute 
app_name = ALAN_FR
are_companies_in_same_account 
are_companies_in_same_account(company_id_1, company_id_2)
Source code in components/fr/public/employment/fr_country_gateway.py 
 96
 97
 98
 99
100
101
102
103
104
@override
def are_companies_in_same_account(
    self, company_id_1: str, company_id_2: str
) -> bool:
    from components.fr.internal.models.company import Company

    company_1 = get_or_raise_missing_resource(Company, int(company_id_1))
    company_2 = get_or_raise_missing_resource(Company, int(company_id_2))
    return company_1.account_id == company_2.account_id
extract_external_employee_id_from_data 
extract_external_employee_id_from_data(
    data, field_name, company_id
)
Source code in components/fr/public/employment/fr_country_gateway.py 
166
167
168
169
170
171
172
173
174
175
176
177
@override
def extract_external_employee_id_from_data(
    self,
    data: dict[str, Any],
    field_name: str,
    company_id: int,
) -> "ExtractorResult[str | None]":
    from components.fr.internal.fr_employment_data_sources.business_logic.rules.extractors import (
        extract_external_id_from_data,
    )

    return extract_external_id_from_data(data, field_name, company_id)
get_account_name 
get_account_name(account_id)
Source code in components/fr/public/employment/fr_country_gateway.py 
106
107
108
109
110
@override
def get_account_name(self, account_id: UUID) -> str:
    from components.fr.internal.models.account import Account

    return get_or_raise_missing_resource(Account, account_id).name
get_admin_error_resolver_classes 
get_admin_error_resolver_classes()
Source code in components/fr/public/employment/fr_country_gateway.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
@override
def get_admin_error_resolver_classes(self) -> dict[str, type[Any]]:
    from components.employment.public.business_logic.rules.employment_input_error_info import (
        EmploymentInputErrorCode,
    )
    from components.employment.public.exceptions import ActionNotAllowed
    from components.fr.public.employment.admin_resolvers.dsn_removal_suggestion_resolver import (
        DsnRemovalSuggestionResolver,
    )
    from components.fr.public.employment.admin_resolvers.missing_ssn_ntt_resolver import (
        MissingSsnNttResolver,
    )
    from components.fr.public.employment.admin_resolvers.no_eligible_contract_found_resolver import (
        NoEligibleContractFoundResolver,
    )
    from components.fr_health_insurance_affiliation.public.exceptions import (
        NoEligibleContractFoundError,
    )

    return {
        ActionNotAllowed.error_code: DsnRemovalSuggestionResolver,
        EmploymentInputErrorCode.missing_required_ssn_ntt.value: MissingSsnNttResolver,
        NoEligibleContractFoundError.error_code: NoEligibleContractFoundResolver,
    }
get_blocked_invitations_validation_broadcast_id 
get_blocked_invitations_validation_broadcast_id()
Source code in components/fr/public/employment/fr_country_gateway.py 
237
238
239
@override
def get_blocked_invitations_validation_broadcast_id(self) -> str | None:
    return "221"
get_company_information 
get_company_information(company_id)
Source code in components/fr/public/employment/fr_country_gateway.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@override
def get_company_information(
    self,
    company_id: str,
) -> CompanyInformation | None:
    from components.fr.internal.models.company import Company

    company = get_resource_or_none(Company, int(company_id))
    if not company:
        return None
    return CompanyInformation(
        display_name=company.display_name,
        account_id=company.account_id,
        is_opt_out_for_offshoring=company.is_opt_out_for_offshoring,
    )
get_consumers_to_notify_for_legacy_backfill 
get_consumers_to_notify_for_legacy_backfill()

Returns the consumers to notify for the legacy backfill source type.

Source code in components/fr/public/employment/fr_country_gateway.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
@override
def get_consumers_to_notify_for_legacy_backfill(
    self,
) -> set[EmploymentConsumer[FrExtendedValues]]:
    """
    Returns the consumers to notify for the legacy backfill source type.
    """
    from components.occupational_health.public.employment.employment_consumer import (
        occupational_health_employment_change_consumer,
    )

    return {
        occupational_health_employment_change_consumer,
    }
get_employee_email_from_extended_values 
get_employee_email_from_extended_values(extended_values)

Retrieves the employee email from FR extended values.

Source code in components/fr/public/employment/fr_country_gateway.py 
140
141
142
143
144
145
@override
def get_employee_email_from_extended_values(
    self, extended_values: FrExtendedValues
) -> str | None:
    """Retrieves the employee email from FR extended values."""
    return extended_values.get("invite_email")
get_employee_identifier_for_country 
get_employee_identifier_for_country(extended_values)
Source code in components/fr/public/employment/fr_country_gateway.py 
134
135
136
137
138
@override
def get_employee_identifier_for_country(
    self, extended_values: FrExtendedValues
) -> str | None:
    return extended_values.get("ssn") or extended_values.get("ntt")
get_employment_consumers 
get_employment_consumers()

Gets all employment consumers contributed by this country.

Notes: 1. ALL Employment Consumers will be called regardless of the country of origin. 2. The function that will be called must have all local code as LOCAL imports - otherwise, this breaks Canada (where loading non-CA models is forbidden)

Source code in components/fr/public/employment/fr_country_gateway.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
@override
def get_employment_consumers(self) -> set[EmploymentConsumer[FrExtendedValues]]:
    """
    Gets all employment consumers contributed by this country.

    Notes:
    1. ALL Employment Consumers will be called regardless of the country of origin.
    2. The function that will be called must have all local code as LOCAL imports - otherwise, this breaks Canada
    (where loading non-CA models is forbidden)
    """
    from components.fr.public.operational_scopes.employment_consumer import (
        operational_scopes_employment_change_consumer,
    )
    from components.fr_health_insurance_affiliation.public.employment_consumer import (
        fr_health_affiliation_employment_change_consumer,
    )
    from components.occupational_health.public.employment.employment_consumer import (
        occupational_health_employment_change_consumer,
    )

    return {
        operational_scopes_employment_change_consumer,
        fr_health_affiliation_employment_change_consumer,
        occupational_health_employment_change_consumer,
    }
get_local_under_investigation_error_codes 
get_local_under_investigation_error_codes()
Source code in components/fr/public/employment/fr_country_gateway.py 
301
302
303
304
305
306
307
308
309
310
311
@override
def get_local_under_investigation_error_codes(self) -> frozenset[str]:
    from components.fr.internal.fr_employment_data_sources.enums.user_matching_error_code import (
        UserMatchingErrorCode,
    )

    return frozenset(
        {
            UserMatchingErrorCode.only_matched_users_with_inconsistent_information.value,
        }
    )
get_retry_function 
get_retry_function()

(Advanced) Get the function used for retrying Core Blocked Movements.

You should generally not need to implement this.

Source code in components/fr/public/employment/fr_country_gateway.py 
153
154
155
156
157
158
159
160
161
162
163
164
@override
def get_retry_function(self) -> RetryFunction[FrExtendedValues]:
    """
    (Advanced) Get the function used for retrying Core Blocked Movements.

    You should generally not need to implement this.
    """
    from components.fr.internal.fr_employment_data_sources.business_logic.global_affiliation_transition.ingest_employment_declaration import (
        ingest_employment_declaration_with_fr_legacy_update,
    )

    return ingest_employment_declaration_with_fr_legacy_update
get_source_detail_for_blocked_movement 
get_source_detail_for_blocked_movement(
    employment_source_data_id,
)
Source code in components/fr/public/employment/fr_country_gateway.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
@override
def get_source_detail_for_blocked_movement(
    self, employment_source_data_id: UUID
) -> str | None:
    from components.employment.public.business_logic.queries.employment_source_data import (
        get_employment_source_data_from_id,
    )

    employment_source_data = get_employment_source_data_from_id(
        employment_source_data_id
    )

    if employment_source_data.source_type == SourceType.fr_external_api:
        if "payfit" in employment_source_data.metadata.get("endpoint", ""):
            return "payfit"
    return None
get_subscriptions_for_company 
get_subscriptions_for_company(company_id)
Source code in components/fr/public/employment/fr_country_gateway.py 
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
@override
def get_subscriptions_for_company(
    self, company_id: str
) -> "list[SubscriptionVersionSummary]":
    from components.contracting.public.subscription.subscription import (
        SubscriptionScope,
        get_subscription_versions_for_owner,
    )
    from components.fr.internal.business_logic.company.queries.company import (
        get_account_id_for_fr_company,
    )

    subscriptions = super().get_subscriptions_for_company(company_id)

    account_id = get_account_id_for_fr_company(company_id=int(company_id))
    return subscriptions + get_subscription_versions_for_owner(
        owner_ref=account_id,
        subscription_scope=SubscriptionScope.occupational_health,
    )
get_upstream_retry_handler 
get_upstream_retry_handler(source_type)
Source code in components/fr/public/employment/fr_country_gateway.py 
147
148
149
150
151
@override
def get_upstream_retry_handler(
    self, source_type: SourceType
) -> UpstreamBlockedMovementRetryFunction[FrExtendedValues] | None:
    return _fr_upstream_retry_handlers.get(source_type)
get_user_admined_company_ids 
get_user_admined_company_ids(user_id)
Source code in components/fr/public/employment/fr_country_gateway.py 
112
113
114
115
116
117
118
@override
def get_user_admined_company_ids(self, user_id: str) -> list[str]:
    from components.fr.internal.business_logic.company.queries.company import (
        get_user_admined_company_ids,
    )

    return [str(id) for id in get_user_admined_company_ids(int(user_id))]
get_user_email 
get_user_email(user_id)
Source code in components/fr/public/employment/fr_country_gateway.py 
127
128
129
130
131
132
@override
def get_user_email(self, user_id: str) -> str | None:
    from components.fr.internal.models.user import User

    user = get_resource_or_none(User, int(user_id))
    return user.email if user else None
get_user_full_name 
get_user_full_name(user_id)
Source code in components/fr/public/employment/fr_country_gateway.py 
120
121
122
123
124
125
@override
def get_user_full_name(self, user_id: str) -> str | None:
    from components.fr.internal.models.user import User

    user = get_resource_or_none(User, int(user_id))
    return user.full_name if user else None
last_stale_invite_notification_email_sent_to_admin_on 
last_stale_invite_notification_email_sent_to_admin_on(
    company_id,
)
Source code in components/fr/public/employment/fr_country_gateway.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
@override
def last_stale_invite_notification_email_sent_to_admin_on(
    self, company_id: str
) -> datetime | None:
    from components.fr.internal.models.company import Company
    from components.fr.internal.models.email_log import EmailLog

    company = get_or_raise_missing_resource(Company, int(company_id))

    new_blocked_invitation_broadcast = "cio-broadcast-221"

    admins = [
        company_admin.user_id
        for company_admin in company.company_admins  # company_admins include account and operation scope admins
    ]

    stale_invite_notification_email = (
        current_session.scalars(
            select(EmailLog)
            .filter(
                EmailLog.mailer == new_blocked_invitation_broadcast,
                EmailLog.sent.is_(
                    True
                ),  # mypy thinks is_ doesn't exist but it does
                EmailLog.user_id.in_(admins),
            )
            .order_by(EmailLog.sent_at.desc())
        )
        .unique()
        .first()
    )
    if not stale_invite_notification_email:
        return None
    return stale_invite_notification_email.sent_at

fr_extended_values

EmployeeTerminationType

Bases: AlanBaseEnum

ani_member_declared_new_job class-attribute instance-attribute 
ani_member_declared_new_job = 'ani_member_declared_new_job'
deceased class-attribute instance-attribute 
deceased = 'deceased'
departure_end_apprentice class-attribute instance-attribute 
departure_end_apprentice = 'departure_end_apprentice'
departure_end_cdd class-attribute instance-attribute 
departure_end_cdd = 'departure_end_cdd'
departure_negociated_termination class-attribute instance-attribute 
departure_negociated_termination = (
    "departure_negociated_termination"
)
departure_resignation class-attribute instance-attribute 
departure_resignation = 'departure_resignation'
departure_resignation_with_ani class-attribute instance-attribute 
departure_resignation_with_ani = (
    "departure_resignation_with_ani"
)
departure_termination class-attribute instance-attribute 
departure_termination = 'departure_termination'
departure_termination_fault class-attribute instance-attribute 
departure_termination_fault = 'departure_termination_fault'
dispensed class-attribute instance-attribute 
dispensed = 'dispensed'
fraud_confirmed class-attribute instance-attribute 
fraud_confirmed = 'fraud_confirmed'
invited_in_ani class-attribute instance-attribute 
invited_in_ani = 'invited_in_ani'
invited_in_retirement class-attribute instance-attribute 
invited_in_retirement = 'invited_in_retirement'
migrated class-attribute instance-attribute 
migrated = 'migrated'
other class-attribute instance-attribute 
other = 'other'
retirement class-attribute instance-attribute 
retirement = 'retirement'
retirement_with_ani class-attribute instance-attribute 
retirement_with_ani = 'retirement_with_ani'
unpaid_invoices class-attribute instance-attribute 
unpaid_invoices = 'unpaid_invoices'
unpaid_leave class-attribute instance-attribute 
unpaid_leave = 'unpaid_leave'
unpaid_leave_exit class-attribute instance-attribute 
unpaid_leave_exit = 'unpaid_leave_exit'

ExemptionType

Bases: AlanBaseEnum

acs_or_cmu class-attribute instance-attribute 
acs_or_cmu = 'acs_or_cmu'
already_covered_indiv class-attribute instance-attribute 
already_covered_indiv = 'already_covered_indiv'
already_covered_military class-attribute instance-attribute 
already_covered_military = 'already_covered_military'
already_covered_other_employer class-attribute instance-attribute 
already_covered_other_employer = (
    "already_covered_other_employer"
)
already_covered_partner class-attribute instance-attribute 
already_covered_partner = 'already_covered_partner'
already_covered_regime_local class-attribute instance-attribute 
already_covered_regime_local = (
    "already_covered_regime_local"
)
already_covered_state_agent class-attribute instance-attribute 
already_covered_state_agent = 'already_covered_state_agent'
cdd_already_covered class-attribute instance-attribute 
cdd_already_covered = 'cdd_already_covered'
cdd_less_than_a_year class-attribute instance-attribute 
cdd_less_than_a_year = 'cdd_less_than_a_year'
cdd_less_than_three_months class-attribute instance-attribute 
cdd_less_than_three_months = 'cdd_less_than_three_months'
continue_to_refuse_coverage class-attribute instance-attribute 
continue_to_refuse_coverage = 'continue_to_refuse_coverage'
health_payment_beneficiary class-attribute instance-attribute 
health_payment_beneficiary = 'health_payment_beneficiary'
low_salary class-attribute instance-attribute 
low_salary = 'low_salary'
madelin class-attribute instance-attribute 
madelin = 'madelin'
refuse_coverage class-attribute instance-attribute 
refuse_coverage = 'refuse_coverage'

FrEmploymentDeclaration module-attribute 

FrEmploymentDeclaration = EmploymentDeclaration[
    FrExtendedValues
]

FrExtendedValues

Bases: ExtendedValuesDict

Local values for France employments in the Employment Component

ccn_code instance-attribute 
ccn_code
entity_code instance-attribute 
entity_code
invite_address instance-attribute 
invite_address
invite_as_exempted instance-attribute 
invite_as_exempted
invite_birth_date instance-attribute 
invite_birth_date
invite_email instance-attribute 
invite_email
invite_exemption_type instance-attribute 
invite_exemption_type
invite_health_coverage_insurer_name instance-attribute 
invite_health_coverage_insurer_name
invite_iban instance-attribute 
invite_iban
invite_in_unpaid_leave instance-attribute 
invite_in_unpaid_leave
invite_unpaid_leave_ends_on instance-attribute 
invite_unpaid_leave_ends_on
invite_unpaid_leave_mandatory_coverage instance-attribute 
invite_unpaid_leave_mandatory_coverage
is_alsace_moselle instance-attribute 
is_alsace_moselle
nic instance-attribute 
nic
ntt instance-attribute 
ntt
professional_category instance-attribute 
professional_category
ssn instance-attribute 
ssn
termination_type instance-attribute 
termination_type

ProfessionalCategory

Bases: AlanBaseEnum

This enum is used to model both the professional category of persons and abstract objects, such as contracts: - persons can be "cadres", "non_cadres", or NULL which models we don't know their professional category; - abstract entities can be "cadres", "non_cadres", or NULL which models we don't have to apply any constraint on the object regarding the professional category. Either way, we want to get rid of the "all" value. It was more targeted at the second use case but it's more coherent to model objects the same way for both kind of objects.

all class-attribute instance-attribute 
all = 'all'
cadres class-attribute instance-attribute 
cadres = 'cadres'
display_name staticmethod 
display_name(population)
Source code in components/fr/internal/models/enums/professional_category.py 
35
36
37
38
39
40
41
42
43
44
@staticmethod
def display_name(population: Optional["ProfessionalCategory"]) -> str | None:
    return (
        {
            ProfessionalCategory.cadres: "cadres",
            ProfessionalCategory.non_cadres: "non cadres",
        }.get(population)
        if population
        else None
    )
non_cadres class-attribute instance-attribute 
non_cadres = 'non_cadres'
suffix staticmethod 
suffix(population)
Source code in components/fr/internal/models/enums/professional_category.py 
24
25
26
27
28
29
30
31
32
33
@staticmethod
def suffix(population: Optional["ProfessionalCategory"]) -> str:
    return (
        {
            ProfessionalCategory.cadres: PROFESSIONAL_CATEGORY_CADRE_SUFFIX,
            ProfessionalCategory.non_cadres: PROFESSIONAL_CATEGORY_NON_CADRE_SUFFIX,
        }.get(population, "")
        if population
        else ""
    )

queries

get_user_active_employment_in_company_on_or_after 

get_user_active_employment_in_company_on_or_after(
    user_id, company_id, active_date
)
Source code in components/fr/internal/business_logic/company/queries/employment.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@tracer.wrap()
def get_user_active_employment_in_company_on_or_after(
    user_id: int, company_id: int, active_date: date
) -> Employment | None:
    user = get_or_raise_missing_resource(User, user_id)
    active_date = active_date or utctoday()

    return next(
        (
            e
            for e in user.employments
            if e.company_id == company_id
            and e.is_ever_active_between(active_date, None)
        ),
        None,
    )

components.fr.public.enrollment

components.fr.public.entities

components.fr.public.enums

components.fr.public.events

subscription

subscribe_to_events 

subscribe_to_events()

All event subscriptions for France should be done here.

Source code in components/fr/public/events/subscription.py 
 1
 2
 3
 4
 5
 6
 7
 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
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
def subscribe_to_events() -> None:
    """
    All event subscriptions for France should be done here.
    """
    from components.ca.public.events.subscription import subscribe_to_ca_global_events
    from components.es.public.events.subscription import subscribe_to_es_global_events
    from components.fr.internal.claim_management.hp_payment.public.events.subscription import (
        subscribe_to_hp_payment_events,
    )
    from components.fr.internal.claim_management.public.reimbursement_payment.api import (
        update_reimbursement_payment_status,
    )
    from components.fr.internal.events.subscribers import (
        notify_tracking_when_address_changed,
        unblock_non_verified_accounts_by_verification_request_id,
    )
    from components.global_profile.public.events import (
        ProfileAddressChanged,
    )
    from components.id_verification.public.events import (
        IdVerificationValidatedForOnboarding,
    )
    from components.payment_gateway.public.messages import (
        PaymentRequestCreated,
        PaymentRequestUpdated,
    )
    from components.payment_gateway.public.workspaces import (
        FR_JPMORGAN_CLAIMS_WORKSPACE_KEY,
    )
    from shared.messaging.broker import get_message_broker
    from shared.queuing.config import CLAIM_ENGINE_QUEUE, LOW_PRIORITY_QUEUE

    message_broker = get_message_broker()

    subscribe_to_es_global_events()
    subscribe_to_ca_global_events()
    subscribe_to_hp_payment_events()

    # Subscriptions to global profile events

    message_broker.subscribe_async(
        ProfileAddressChanged,
        notify_tracking_when_address_changed,
        queue_name=LOW_PRIORITY_QUEUE,
    )

    # Subscriptions to ID verification events
    message_broker.subscribe_async(
        IdVerificationValidatedForOnboarding,
        unblock_non_verified_accounts_by_verification_request_id,
        queue_name=LOW_PRIORITY_QUEUE,
    )

    # Subscribe to payment gateway events
    message_broker.subscribe_async(
        PaymentRequestCreated,
        update_reimbursement_payment_status,
        # Only for JPMorgan FR workspace
        predicate=lambda m: m.workspace_key == FR_JPMORGAN_CLAIMS_WORKSPACE_KEY,
        queue_name=CLAIM_ENGINE_QUEUE,
    )
    message_broker.subscribe_async(
        PaymentRequestUpdated,
        update_reimbursement_payment_status,
        # Only for JPMorgan FR workspace
        predicate=lambda m: m.workspace_key == FR_JPMORGAN_CLAIMS_WORKSPACE_KEY,
        queue_name=CLAIM_ENGINE_QUEUE,
    )

components.fr.public.feature

is_feature_active 

is_feature_active(feature_name)

Check if a feature is active :param feature_name: the name of the feature :return: true if the feature with this name is found and active

Source code in components/fr/public/feature.py 
 4
 5
 6
 7
 8
 9
10
11
12
def is_feature_active(
    feature_name: str,
) -> bool:
    """
    Check if a feature is active
    :param feature_name: the name of the feature
    :return: true if the feature with this name is found and active
    """
    return Feature.is_enabled(name=feature_name)

components.fr.public.fraud_detection

actions

fraud_relevant_user_change

FraudRelevantUserChangeSource

Bases: AlanBaseEnum

EnrollmentAsBeneficiary class-attribute instance-attribute 
EnrollmentAsBeneficiary = 'enrollment_as_beneficiary'
UserSignup class-attribute instance-attribute 
UserSignup = 'user_signup'
track_changed_email_for_fraud 
track_changed_email_for_fraud(
    actor_id,
    user_id,
    new_email,
    old_email,
    source=None,
    commit=False,
)
Source code in components/fr/internal/fraud_detection/business_logic/actions/fraud_relevant_user_change.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def track_changed_email_for_fraud(
    actor_id: int,
    user_id: int,
    new_email: str | None,
    old_email: str | None,
    source: FraudRelevantUserChangeSource | None = None,
    commit: bool = False,
) -> None:
    def _email_record(email: str | None) -> FraudRelevantUserChangeValuesContent:
        return FraudRelevantUserChangeValuesContent(data=EmailRecord(email=email))

    change = FraudRelevantUserChange(
        user_id=user_id,
        actor_id=actor_id,
        type=FraudRelevantUserChangeType.EmailUpdated,
        source=source,
        new_values=_email_record(new_email),
        old_values=_email_record(old_email),
    )
    current_session.add(change)

    if commit:
        current_session.commit()

enums

FraudRelevantUserChangeType

Bases: AlanBaseEnum

EmailUpdated class-attribute instance-attribute 
EmailUpdated = 'email_updated'
SettlementIbanUpdated class-attribute instance-attribute 
SettlementIbanUpdated = 'settlement_iban_updated'

queries

find_most_recent_fraud_relevant_user_changes 

find_most_recent_fraud_relevant_user_changes(
    user_ids,
    change_type=None,
    limit=None,
    since=None,
    until=None,
)
Source code in components/fr/internal/fraud_detection/business_logic/queries/fraud_relevant_user_change.py 
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
def find_most_recent_fraud_relevant_user_changes(
    user_ids: Iterable[int],
    change_type: FraudRelevantUserChangeType | None = None,
    limit: int | None = None,
    since: datetime | None = None,
    until: datetime | None = None,
) -> list[FraudRelevantUserChangeEntity]:
    query = current_session.query(FraudRelevantUserChange).filter(  # noqa: ALN085
        FraudRelevantUserChange.user_id.in_(user_ids)
    )
    if change_type:
        query = query.filter(FraudRelevantUserChange.type == change_type)
    if since:
        query = query.filter(FraudRelevantUserChange.created_at >= since)
    if until:
        query = query.filter(FraudRelevantUserChange.created_at <= until)
    if limit:
        query = query.limit(limit)
    query = query.order_by(FraudRelevantUserChange.created_at.asc())

    return [
        FraudRelevantUserChangeEntity(
            user_id=change.user_id,
            actor_id=change.actor_id,
            type=change.type,
            source=change.source,
            new_values=change.new_values,
            old_values=change.old_values,
            occurred_at=change.created_at,
        )
        for change in query.all()
    ]

has_open_fraud_investigation 

has_open_fraud_investigation(user_id)
Source code in components/fr/internal/fraud_detection/business_logic/queries/has_open_fraud_investigation.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def has_open_fraud_investigation(user_id: int) -> bool:
    return bool(
        current_session.scalar(
            select(
                exists()
                .where(FraudInvestigation.closed_at.is_(None))
                .where(FraudInvestigation.id == CareActSuspicion.fraud_investigation_id)
                .where(CareActSuspicion.care_act_id == CareAct.id)
                .where(CareAct.id == CareActConsolidatedInfo.care_act_id)
                .where(
                    CareActConsolidatedInfo.insurance_profile_id == InsuranceProfile.id
                )
                .where(InsuranceProfile.user_id == user_id)
            )
        )
    )

components.fr.public.global_customer_dashboard

admin

get_admined_entities_for_entity_selector_fr 

get_admined_entities_for_entity_selector_fr(user_id)
Source code in components/fr/public/global_customer_dashboard/admin.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def get_admined_entities_for_entity_selector_fr(  # noqa: D103
    user_id: str,
) -> list[AdminedEntityForEntitySelector]:
    from components.fr.internal.models.account import Account
    from components.fr.internal.models.company import Company
    from components.fr.internal.operational_scopes.models.operational_scope import (
        OperationalScope,
    )

    return get_admined_entities_for_entity_selector_global(
        user_id=user_id,
        operational_scope_model=OperationalScope,
        company_model=Company,
        account_model=Account,
        admined_entities_query_api=admined_entities_query_api_fr(),
    )

get_pending_onboardings_for_admined_entity_selector_fr 

get_pending_onboardings_for_admined_entity_selector_fr(
    user_id,
)
Source code in components/fr/public/global_customer_dashboard/admin.py 
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
def get_pending_onboardings_for_admined_entity_selector_fr(  # noqa: D103
    user_id: str,
) -> list[PendingOnboarding]:
    from components.fr.internal.business_logic.company.queries.company import (
        get_companies_onboarding_status,
    )
    from components.fr.internal.models.company import Company
    from components.fr.internal.models.enums.company_onboarding_status import (
        CompanyOnboardingStatus,
    )

    def companies_having_pending_onboardings(company_ids: list[str]) -> dict[str, bool]:
        company_id_to_onboarding_status = get_companies_onboarding_status(
            [int(company_id) for company_id in company_ids]
        )

        return {
            company_id: company_id_to_onboarding_status[int(company_id)]
            not in {
                CompanyOnboardingStatus.completed,
                CompanyOnboardingStatus.missing_signature_callback,
            }
            for company_id in company_ids
        }

    return get_pending_onboardings_from_admined_entities(
        user_id=user_id,
        company_model=Company,
        companies_having_pending_onboardings=companies_having_pending_onboardings,
        admined_entities_query_api=admined_entities_query_api_fr(),
    )

product_setting

get_account_settings_for_company 

get_account_settings_for_company(company_id)

Returns the list of activated AccountSetting for the given company :param company_id: The ID of the company we want AccountSetting for :return: The list of activated AccountSetting for this company

Source code in components/fr/public/global_customer_dashboard/product_setting.py 
 7
 8
 9
10
11
12
13
def get_account_settings_for_company(company_id: int) -> list[ProductSetting]:
    """
    Returns the list of activated AccountSetting for the given company
    :param company_id: The ID of the company we want AccountSetting for
    :return: The list of activated AccountSetting for this company
    """
    return get_account_settings_for_company_internal(company_id=company_id)

components.fr.public.health_pricing

pricing_type

PricingType

Bases: AlanBaseEnum

Enum for pricing types that are used in the health pricing component.

adult_children class-attribute instance-attribute 
adult_children = 'adult_children'
children class-attribute instance-attribute 
children = 'children'
family class-attribute instance-attribute 
family = 'family'
partner class-attribute instance-attribute 
partner = 'partner'
primary class-attribute instance-attribute 
primary = 'primary'

components.fr.public.intercom

actions

intercom_user_id

get_or_create_intercom_user_id 
get_or_create_intercom_user_id(is_admin, user_id)

Get or create an Intercom user ID for the given user.

Parameters:

Name Type Description Default
is_admin bool

Whether to use the pro email (admin) or personal email.

 required
user_id int

The user ID to get or create the Intercom user for.

 required

Returns:

Type Description
str | None

The Intercom user ID, or None if creation failed.
Source code in components/fr/public/intercom/actions/intercom_user_id.py 
 5
 6
 7
 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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def get_or_create_intercom_user_id(is_admin: bool, user_id: int) -> str | None:
    """Get or create an Intercom user ID for the given user.

    Args:
        is_admin: Whether to use the pro email (admin) or personal email.
        user_id: The user ID to get or create the Intercom user for.

    Returns:
        The Intercom user ID, or None if creation failed.
    """
    from components.fr.internal.business_logic.intercom.queries.users import (
        get_intercom_id_with_pro_perso,
    )
    from components.fr.internal.models.user import User
    from components.global_profile.public.api import ProfileService
    from shared.helpers.get_or_else import get_or_raise_missing_resource
    from shared.helpers.logging.logger import current_logger
    from shared.helpers.typing import mandatory
    from shared.services.intercom.intercom_care_client import get_care_intercom_client

    intercom_ids = get_intercom_id_with_pro_perso(user_id=user_id)
    user = get_or_raise_missing_resource(User, user_id)

    profile_service = ProfileService.create()
    profile = profile_service.user_compat.get_or_raise_profile_by_user_id(user_id)
    # TODO(@crew-global-architecture): Profile entity missing pro_email, using user.pro_email for now
    intercom_user_id = (
        intercom_ids.pro
        if is_admin and user.pro_email and user.pro_email != profile.email
        else intercom_ids.perso
    )

    care_intercom_client = get_care_intercom_client(AppName.ALAN_FR)

    if not intercom_user_id:
        current_logger.warning(
            "get_or_create_intercom_user_id: Intercom User not found, creating Intercom user",
            user_id=user_id,
        )
        # TODO(@crew-global-architecture): Profile entity missing pro_email, manual fallback needed
        email = (
            mandatory(user.pro_email or profile.email)
            if is_admin
            else mandatory(profile.email)
        )
        intercom_user_id = care_intercom_client.create_intercom_user(
            email=mandatory(email),
            user_id=user_id,
            full_name=normalize_names(profile.first_name, profile.last_name),
        )
    return intercom_user_id

components.fr.public.kyc

actions

install_reimbursement_blocker_for_siren 

install_reimbursement_blocker_for_siren(
    siren, reason, block_from=None, block_to=None, save=True
)
Source code in components/fr/internal/kyc/business_logic/actions/reimbursement_blocker.py 
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
def install_reimbursement_blocker_for_siren(
    siren: str,
    reason: BlockReimbursementsReason,
    block_from: date | None = None,
    block_to: date | None = None,
    save: bool = True,
) -> None:
    if not is_siren_valid(siren):
        current_logger.warning(
            f"Trying to block reimbursement for invalid siren {siren}, skipping"
        )
        return

    if has_siren_an_active_blocker_for_reason_and_span(
        siren=siren, reason=reason, block_from=block_from, block_to=block_to
    ):
        current_logger.warning(
            f"Siren {siren} is already blocked for reason {reason} on date span {block_from} to {block_to}, skipping"
        )
        return

    blocker = ReimbursementBlocker(
        scope=ReimbursementBlockerScope.siren,
        siren=siren,
        reason=reason,
        started_at=utcnow(),
        block_from=block_from,
        block_to=block_to,
    )
    current_session.add(blocker)

    # To be used depending on the transaction context:
    # - before Ondorse calls, we want to commit whatever happens next
    # - in self-serve context, we don’t want to trigger the external transaction
    if save:
        current_session.commit()

enums

BlockReimbursementsReason

Bases: AlanBaseEnum

ani_entry class-attribute instance-attribute 
ani_entry = 'ani_entry'
ani_missing_justifications class-attribute instance-attribute 
ani_missing_justifications = 'ani_missing_justifications'
ani_switch_overlapping class-attribute instance-attribute 
ani_switch_overlapping = 'ani_switch_overlapping'
company_risk_rejected class-attribute instance-attribute 
company_risk_rejected = 'company_risk_rejected'
fraud_suspicion class-attribute instance-attribute 
fraud_suspicion = 'fraud_suspicion'
id_not_verified class-attribute instance-attribute 
id_not_verified = 'id_not_verified'
manual class-attribute instance-attribute 
manual = 'manual'
pending_company_kyc class-attribute instance-attribute 
pending_company_kyc = 'pending_company_kyc'

helpers

is_siren_already_cleared 

is_siren_already_cleared(siren)

Check if the last KYC Check on the SIREN was an approbation Return False in all other cases

Source code in components/fr/internal/kyc/business_logic/helpers/kyc_checks.py 
135
136
137
138
139
140
141
142
143
144
145
def is_siren_already_cleared(siren: str) -> bool:
    """
    Check if the last KYC Check on the SIREN was an approbation
    Return False in all other cases
    """
    latest_siren_kyc_status = get_latest_status_for_siren(siren)

    if not latest_siren_kyc_status:
        return False

    return latest_siren_kyc_status == KycCheckStatusReview.approved

is_siren_valid 

is_siren_valid(siren)
Source code in components/fr/internal/kyc/business_logic/helpers/siren.py 
6
7
def is_siren_valid(siren: str) -> bool:
    return not _is_zero_made(siren) and _luhn_check(siren, SIREN_VALID_LENGTH)  # type: ignore[no-untyped-call]

queries

company_exists_for_siren 

company_exists_for_siren(siren)

Check if a Company exists with the given SIREN.

Source code in components/fr/internal/kyc/business_logic/queries/siren.py 
 7
 8
 9
10
11
12
13
14
15
def company_exists_for_siren(siren: str) -> bool:
    """
    Check if a Company exists with the given SIREN.
    """
    return bool(
        current_session.scalar(
            select(exists(select(Company).where(Company.siren == siren)))
        )
    )

is_siren_currently_blocked 

is_siren_currently_blocked(siren)
Source code in components/fr/internal/kyc/business_logic/queries/reimbursement_blocker.py 
18
19
def is_siren_currently_blocked(siren: str) -> bool:
    return is_siren_blocked_at_date(siren=siren, at_date=date.today())

components.fr.public.member_attributes

getter_decorators

Decorators that resolve a global Profile into FR-specific models for member attribute getters.

Each decorator pops profile from kwargs, resolves it to the corresponding FR model, injects the result under a model-specific kwarg name, and forwards the remaining kwargs to the wrapped getter.

T module-attribute 

T = TypeVar('T')

uses_fr_contract 

uses_fr_contract(getter)

Profile -> FR Contract. Injects contract=.

Source code in components/fr/public/member_attributes/getter_decorators.py 
77
78
79
80
81
82
83
84
85
86
87
def uses_fr_contract(getter: Callable[..., T]) -> Callable[..., T | None]:
    """Profile -> FR Contract. Injects ``contract=``."""

    @functools.wraps(getter)
    def wrapper(*, profile: Profile, **kwargs: Any) -> T | None:
        resolved = resolve_contract(profile)
        if resolved is None:
            return None
        return getter(contract=resolved, **kwargs)

    return wrapper

uses_fr_enrollment 

uses_fr_enrollment(getter)

Pass-through for FR Enrollment. Injects enrollment=.

Source code in components/fr/public/member_attributes/getter_decorators.py 
103
104
105
106
107
108
109
110
111
112
def uses_fr_enrollment(getter: Callable[..., T]) -> Callable[..., T | None]:
    """Pass-through for FR Enrollment. Injects ``enrollment=``."""

    @functools.wraps(getter)
    def wrapper(*, enrollment: Any = None, **kwargs: Any) -> T | None:
        if enrollment is None:
            return None
        return getter(enrollment=enrollment, **kwargs)

    return wrapper

uses_fr_insurance_profile 

uses_fr_insurance_profile(getter)

Profile -> FR InsuranceProfile. Injects insurance_profile=.

Source code in components/fr/public/member_attributes/getter_decorators.py 
38
39
40
41
42
43
44
45
46
47
48
def uses_fr_insurance_profile(getter: Callable[..., T]) -> Callable[..., T | None]:
    """Profile -> FR InsuranceProfile. Injects ``insurance_profile=``."""

    @functools.wraps(getter)
    def wrapper(*, profile: Profile, **kwargs: Any) -> T | None:
        resolved = resolve_insurance_profile(profile)
        if resolved is None:
            return None
        return getter(insurance_profile=resolved, **kwargs)

    return wrapper

uses_fr_option_settings 

uses_fr_option_settings(getter)

Profile -> FR PolicyOptionSettings. Injects option_settings=.

Source code in components/fr/public/member_attributes/getter_decorators.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def uses_fr_option_settings(getter: Callable[..., T]) -> Callable[..., T | None]:
    """Profile -> FR PolicyOptionSettings. Injects ``option_settings=``."""

    @functools.wraps(getter)
    def wrapper(*, profile: Profile, **kwargs: Any) -> T | None:
        resolved = resolve_option_settings(profile)
        if resolved is None:
            return None
        return getter(option_settings=resolved, **kwargs)

    return wrapper

uses_fr_policy 

uses_fr_policy(getter)

Profile -> FR current Policy. Injects policy=.

Source code in components/fr/public/member_attributes/getter_decorators.py 
51
52
53
54
55
56
57
58
59
60
61
def uses_fr_policy(getter: Callable[..., T]) -> Callable[..., T | None]:
    """Profile -> FR current Policy. Injects ``policy=``."""

    @functools.wraps(getter)
    def wrapper(*, profile: Profile, **kwargs: Any) -> T | None:
        resolved = resolve_current_policy(profile)
        if resolved is None:
            return None
        return getter(policy=resolved, **kwargs)

    return wrapper

uses_fr_primary_user 

uses_fr_primary_user(getter)

Profile -> FR primary User. Injects user=.

Source code in components/fr/public/member_attributes/getter_decorators.py 
64
65
66
67
68
69
70
71
72
73
74
def uses_fr_primary_user(getter: Callable[..., T]) -> Callable[..., T | None]:
    """Profile -> FR primary User. Injects ``user=``."""

    @functools.wraps(getter)
    def wrapper(*, profile: Profile, **kwargs: Any) -> T | None:
        resolved = resolve_primary_user(profile)
        if resolved is None:
            return None
        return getter(user=resolved, **kwargs)

    return wrapper

uses_fr_user 

uses_fr_user(getter)

Profile -> FR User. Injects user=.

Source code in components/fr/public/member_attributes/getter_decorators.py 
25
26
27
28
29
30
31
32
33
34
35
def uses_fr_user(getter: Callable[..., T]) -> Callable[..., T | None]:
    """Profile -> FR User. Injects ``user=``."""

    @functools.wraps(getter)
    def wrapper(*, profile: Profile, **kwargs: Any) -> T | None:
        resolved = resolve_user(profile)
        if resolved is None:
            return None
        return getter(user=resolved, **kwargs)

    return wrapper

resolvers

Resolve a global Profile into FR-specific models.

These are internal implementation details of the getter decorators and should not be imported directly by other components.

resolve_contract 

resolve_contract(profile)

Profile -> FR Contract (via current policy).

Source code in components/fr/public/member_attributes/resolvers.py 
50
51
52
53
def resolve_contract(profile: Profile) -> "Contract | None":
    """Profile -> FR Contract (via current policy)."""
    policy = resolve_current_policy(profile)
    return policy.contract if policy else None

resolve_current_policy 

resolve_current_policy(profile)

Profile -> FR current Policy.

Source code in components/fr/public/member_attributes/resolvers.py 
38
39
40
41
def resolve_current_policy(profile: Profile) -> "Policy | None":
    """Profile -> FR current Policy."""
    ip = resolve_insurance_profile(profile)
    return ip.current_policy if ip else None

resolve_insurance_profile 

resolve_insurance_profile(profile)

Profile -> FR InsuranceProfile.

Source code in components/fr/public/member_attributes/resolvers.py 
32
33
34
35
def resolve_insurance_profile(profile: Profile) -> "InsuranceProfile | None":
    """Profile -> FR InsuranceProfile."""
    user = resolve_user(profile)
    return user.insurance_profile if user else None

resolve_option_settings 

resolve_option_settings(profile)

Profile -> FR PolicyOptionSettings (via current policy).

Source code in components/fr/public/member_attributes/resolvers.py 
56
57
58
59
60
61
62
63
64
65
66
def resolve_option_settings(profile: Profile) -> "PolicyOptionSettings | None":
    """Profile -> FR PolicyOptionSettings (via current policy)."""
    from components.fr.internal.business_logic.policy.queries.policy_option_settings import (  # noqa: ALN043
        get_policy_option_settings,
    )

    policy = resolve_current_policy(profile)
    if not policy:
        return None

    return get_policy_option_settings(policy.id)

resolve_primary_user 

resolve_primary_user(profile)

Profile -> FR primary User (via policy.primary_profile).

Source code in components/fr/public/member_attributes/resolvers.py 
44
45
46
47
def resolve_primary_user(profile: Profile) -> "User | None":
    """Profile -> FR primary User (via policy.primary_profile)."""
    policy = resolve_current_policy(profile)
    return policy.primary_profile.user if policy and policy.primary_profile else None

resolve_user 

resolve_user(profile)

Profile -> FR User.

Source code in components/fr/public/member_attributes/resolvers.py 
23
24
25
26
27
28
29
def resolve_user(profile: Profile) -> "User | None":
    """Profile -> FR User."""
    from components.fr.internal.user.business_logic.queries.profile import (  # noqa: ALN043
        get_user_from_profile_id,
    )

    return get_user_from_profile_id(profile.id)

components.fr.public.member_lifecycle

ani

member_attributes

ani_employment_duration_justification_requirement module-attribute 
ani_employment_duration_justification_requirement = MemberAttributeDefinition[
    AniEmploymentDurationJustificationRequirement
](
    name="ani_employment_duration_justification_requirement",
    display_name="ANI employment duration justification requirement",
    description="ANI employment duration justification requirement status",
    getter=_get_ani_employment_duration_justification_requirement,
    raw_type=AniEmploymentDurationJustificationRequirement,
    formatter=_format_ani_employment_duration_justification_requirement,
)
ani_entry_date module-attribute 
ani_entry_date = MemberAttributeDefinition[date](
    name="ani_entry_date",
    display_name="ANI entry date",
    description="ANI portability entry date",
    getter=_get_ani_entry_date,
    raw_type=date,
)
ani_justification_rejection_reason module-attribute 
ani_justification_rejection_reason = MemberAttributeDefinition[
    JustificationRejectionStatus
](
    name="ani_justification_rejection_reason",
    display_name="ANI justification rejection reason",
    description="Rejection reason from latest ANI justification",
    getter=_get_ani_justification_rejection_reason,
    raw_type=JustificationRejectionStatus,
    formatter=_format_ani_justification_rejection_reason,
)
ani_justification_requirement module-attribute 
ani_justification_requirement = MemberAttributeDefinition[
    AniJustificationRequirement
](
    name="ani_justification_requirement",
    display_name="ANI justification requirement",
    description="Current ANI justification requirement status",
    getter=_get_ani_justification_requirement,
    raw_type=AniJustificationRequirement,
    formatter=_format_ani_justification_requirement,
)
ani_justification_status_history module-attribute 
ani_justification_status_history = MemberAttributeDefinition(
    name="ani_justification_status_history",
    display_name="ANI justification status history",
    description="History of ANI justification statuses by month",
    getter=_get_ani_justification_status_history,
    raw_type=dict,
    formatter=_format_ani_justification_status_history,
)
ani_payslip_status_history module-attribute 
ani_payslip_status_history = MemberAttributeDefinition(
    name="ani_payslip_status_history",
    display_name="ANI payslip status history",
    description="History of ANI payslip statuses",
    getter=_get_ani_payslip_status_history,
    raw_type=list,
    formatter=_format_ani_payslip_status_history,
)
max_ani_end_date module-attribute 
max_ani_end_date = MemberAttributeDefinition[date](
    name="max_ani_end_date",
    display_name="Max ANI end date",
    description="Maximum ANI portability end date",
    getter=_get_max_ani_end_date,
    raw_type=date,
)

exemption

member_attributes

NOT_APPLICABLE_STR module-attribute 
NOT_APPLICABLE_STR = 'N/A'
exemption_justification_rejection_reason module-attribute 
exemption_justification_rejection_reason = MemberAttributeDefinition[
    ExemptionJustificationRejectionReason
](
    name="exemption_justification_rejection_reason",
    display_name="Exemption justification rejection reason",
    description="Justification rejection reason of the user's current exemption",
    getter=_get_exemption_justification_rejection_reason,
    raw_type=ExemptionJustificationRejectionReason,
    formatter=_format_exemption_justification_rejection_reason,
)
exemption_justification_status module-attribute 
exemption_justification_status = MemberAttributeDefinition[
    JustificationStatus
](
    name="exemption_justification_status",
    display_name="Exemption justification status",
    description="Justification status of the user's current exemption",
    getter=_get_exemption_justification_status,
    raw_type=JustificationStatus,
)
exemption_status module-attribute 
exemption_status = MemberAttributeDefinition[
    ExemptionOnboardingStatus
](
    name="exemption_status",
    display_name="Exemption status",
    description="Onboarding status of the user's current exemption",
    getter=_get_exemption_status,
    raw_type=ExemptionOnboardingStatus,
)
recent_exemptions module-attribute 
recent_exemptions = MemberAttributeDefinition[
    list[ExemptionMemberAttribute]
](
    name="recent_exemptions",
    display_name="Recent exemptions",
    description="Recent non-cancelled exemptions for the user",
    getter=_get_recent_exemptions,
    raw_type=list,
    formatter=_format_recent_exemptions,
)

components.fr.public.operational_scopes

employment_consumer

operational_scopes_employment_change_consumer 

operational_scopes_employment_change_consumer(
    employment_change, event_bus_orchestrator
)
Source code in components/fr/public/operational_scopes/employment_consumer.py 
13
14
15
16
17
18
19
20
21
22
23
24
def operational_scopes_employment_change_consumer(  # noqa: D103
    employment_change: EmploymentChange["FrExtendedValues"],
    event_bus_orchestrator: EventBusOrchestrator,
) -> None:
    if employment_change.country_code != CountryCode.fr:
        return

    from components.fr.internal.operational_scopes.business_logic.actions import (
        process_employment_change,
    )

    process_employment_change(employment_change, event_bus_orchestrator)

factories

OperationalScopeFactory

Bases: AlanBaseFactory['OperationalScope']

Meta
model class-attribute instance-attribute 
model = OperationalScope
Params
is_group class-attribute instance-attribute 
is_group = Trait(
    type=group, value=lazy_attribute(lambda o: name)
)
company class-attribute instance-attribute 
company = SubFactory(CompanyFullFactory)
name class-attribute instance-attribute 
name = LazyFunction(word)
type class-attribute instance-attribute 
type = siret
value class-attribute instance-attribute 
value = sequence(lambda n: f'{n}')

components.fr.public.policy

member_attributes

PolicyStatus

Bases: AlanBaseEnum

Policy lifecycle status for member attribute.

active_policy class-attribute instance-attribute 
active_policy = 'active_policy'
ani class-attribute instance-attribute 
ani = 'ani'
ended_policy class-attribute instance-attribute 
ended_policy = 'ended_policy'
invited_ani class-attribute instance-attribute 
invited_ani = 'invited_ani'
no_policy class-attribute instance-attribute 
no_policy = 'no_policy'
unknown class-attribute instance-attribute 
unknown = 'unknown'
upcoming_policy class-attribute instance-attribute 
upcoming_policy = 'upcoming_policy'

all_birth_or_adoption_certificates_def module-attribute 

all_birth_or_adoption_certificates_def = MemberAttributeDefinition[
    list[BirthOrAdoptionCertificateInfo]
](
    name="all_birth_or_adoption_certificates",
    display_name="All birth or adoption certificates",
    description="List of birth or adoption certificates for the policy",
    getter=_get_all_birth_or_adoption_certificates,
    raw_type=list[BirthOrAdoptionCertificateInfo],
    formatter=_format_birth_and_adoption_certificates,
)

date_block_reimbursements module-attribute 

date_block_reimbursements = MemberAttributeDefinition[date](
    name="date_block_reimbursements",
    display_name="Date block reimbursements",
    description="Date from which reimbursements are blocked",
    getter=_get_date_block_reimbursements,
    raw_type=date,
)

employee_costs module-attribute 

employee_costs = MemberAttributeDefinition[
    BeneficiaryPrice
](
    name="employee_costs",
    display_name="Employee costs",
    description="Employee pricing breakdown by beneficiary type",
    getter=_get_employee_costs,
    raw_type=BeneficiaryPrice,
    formatter=_format_employee_costs,
)

employee_costs_option module-attribute 

employee_costs_option = MemberAttributeDefinition[
    list[BeneficiaryPrice]
](
    name="employee_costs_option",
    display_name="Employee costs option",
    description="Option pricing breakdown by beneficiary type",
    getter=_get_employee_costs_option,
    raw_type=list[BeneficiaryPrice],
    formatter=_format_employee_costs_option,
)

most_recent_quote_document module-attribute 

most_recent_quote_document = MemberAttributeDefinition[
    QuoteDocumentInfo
](
    name="most_recent_quote_document",
    display_name="Most recent quote document",
    description="Most recent quote document for the policy",
    getter=_get_most_recent_quote_document,
    raw_type=QuoteDocumentInfo,
)

policy_id module-attribute 

policy_id = MemberAttributeDefinition[int](
    name="policy_id",
    display_name="Policy ID",
    description="Current policy ID",
    getter=_get_policy_id,
    raw_type=int,
)

policy_status module-attribute 

policy_status = MemberAttributeDefinition[PolicyStatus](
    name="policy_status",
    display_name="Policy status",
    description="Current policy lifecycle status (active, ended, ANI, etc.)",
    getter=_get_policy_status,
    raw_type=PolicyStatus,
    formatter=_format_policy_status,
)

policy_termination_type module-attribute 

policy_termination_type = MemberAttributeDefinition[
    EmployeeTerminationType
](
    name="policy_termination_type",
    display_name="Policy termination type",
    description="Type of policy termination",
    getter=_get_policy_termination_type,
    raw_type=EmployeeTerminationType,
    formatter=_format_policy_termination_type,
)

recent_quote_documents module-attribute 

recent_quote_documents = MemberAttributeDefinition[
    list[QuoteDocumentInfo]
](
    name="recent_quote_documents",
    display_name="Recent quote documents",
    description="Recent quote documents for the policy",
    getter=_get_recent_quote_documents,
    raw_type=list[QuoteDocumentInfo],
)

reimbursement_block_reason module-attribute 

reimbursement_block_reason = MemberAttributeDefinition[
    BlockReimbursementsReason
](
    name="reimbursement_block_reason",
    display_name="Reimbursement block reason",
    description="Reason reimbursements are blocked",
    getter=_get_reimbursement_block_reason,
    raw_type=BlockReimbursementsReason,
    formatter=_format_reimbursement_block_reason,
)

policy

get_n_base_filtered_children_on 

get_n_base_filtered_children_on(
    policy_id, date, adult_children
)
Source code in components/fr/public/policy/policy.py 
17
18
19
20
21
22
23
@deprecated(
    "Avoid using beneficiary count methods, use rather the unified pricing abstraction to get costs."
)
def get_n_base_filtered_children_on(  # noqa: D103
    policy_id: int, date: datetime.date, adult_children: bool
) -> int:
    return internal_get_n_base_filtered_children_on(policy_id, date, adult_children)

get_n_base_paid_filtered_children_on 

get_n_base_paid_filtered_children_on(
    policy_id, on_date, adult_children
)
Source code in components/fr/public/policy/policy.py 
26
27
28
29
30
31
32
@deprecated(
    "Avoid using beneficiary count methods, use rather the unified pricing abstraction to get costs."
)
def get_n_base_paid_filtered_children_on(  # noqa: D103
    policy_id: int, on_date: datetime.date, adult_children: bool
) -> int:
    return internal_get_n_base_paid_children_on(policy_id, on_date, adult_children)

is_policy_change_significant 

is_policy_change_significant(
    old_policy_id,
    new_policy_id,
    new_start_date,
    old_policy_precomputed_coverage=None,
)

Returns True if the change between two policies is significant, and member notification and action is required.

Returns False if the change can be considered "transparent" and does not require member action.

This function should be used primarily for employee transfer cases.

Source code in components/fr/public/policy/policy.py 
110
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
143
144
145
146
147
148
149
def is_policy_change_significant(
    old_policy_id: int,
    new_policy_id: int,
    new_start_date: datetime.date,
    old_policy_precomputed_coverage: _PrecomputedPolicyCoverage | None = None,
) -> bool:
    """
    Returns True if the change between two policies is significant, and member notification and action is required.

    Returns False if the change can be considered "transparent" and does not require member action.

    This function should be used primarily for employee transfer cases.
    """
    last_day_date = new_start_date - datetime.timedelta(days=1)

    if is_policy_coverage_different(
        old_policy_id=old_policy_id,
        new_policy_id=new_policy_id,
        last_day_date=last_day_date,
        new_start_date=new_start_date,
        old_policy_precomputed_coverage=old_policy_precomputed_coverage,
    ):
        return True

    (no_option_available_anymore, new_option, new_price_for_same_option) = (
        change_in_option_coverage(old_policy_id, new_policy_id, new_start_date)
    )
    if no_option_available_anymore or new_option or new_price_for_same_option:
        return True

    old_policy = get_or_raise_missing_resource(Policy, old_policy_id)
    was_ani = old_policy.is_ani_on(new_start_date - datetime.timedelta(days=1))
    if was_ani:
        return True

    was_individual = old_policy.contract.contractee_type == ContracteeType.individual
    if was_individual:
        return True

    return False

is_policy_coverage_different 

is_policy_coverage_different(
    old_policy_id,
    new_policy_id,
    last_day_date,
    new_start_date,
    old_policy_precomputed_coverage=None,
)

Returns True if the coverage differs between the two policies (either in pricing or in actual coverage).

Source code in components/fr/public/policy/policy.py 
 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
def is_policy_coverage_different(
    old_policy_id: int,
    new_policy_id: int,
    last_day_date: datetime.date,
    new_start_date: datetime.date,
    old_policy_precomputed_coverage: _PrecomputedPolicyCoverage | None = None,
) -> bool:
    """
    Returns True if the coverage differs between the two policies (either in pricing or in actual coverage).
    """
    from components.fr.internal.business_logic.health_pricing.policy import (
        get_policy_cost_split_by_primary_and_dependants,
        is_different_salary_rate_pricing_between_policies,
    )

    old_policy = get_or_raise_missing_resource(Policy, old_policy_id)
    new_policy = get_or_raise_missing_resource(Policy, new_policy_id)

    old_coverage_key = (
        old_policy_precomputed_coverage.coverage_key
        if old_policy_precomputed_coverage is not None
        else old_policy.contract.health_coverage_key(last_day_date)
    )
    new_coverage_key = new_policy.contract.health_coverage_key(new_start_date)

    old_policy_price_split = (
        old_policy_precomputed_coverage.policy_price_split
        if old_policy_precomputed_coverage is not None
        else get_policy_cost_split_by_primary_and_dependants(
            old_policy.id, last_day_date
        )
    )
    new_policy_price_split = get_policy_cost_split_by_primary_and_dependants(
        new_policy.id, new_start_date
    )

    old_coverage_price = old_policy_price_split.total_cost
    new_coverage_price = new_policy_price_split.total_cost

    is_coverage_different = (
        old_coverage_key != new_coverage_key
        or old_coverage_price != new_coverage_price
        or is_different_salary_rate_pricing_between_policies(
            old_policy=old_policy,
            new_policy=new_policy,
            last_day_date=last_day_date,
            new_start_date=new_start_date,
        )
    )
    return is_coverage_different

precompute_policy_coverage_for_policy_comparison 

precompute_policy_coverage_for_policy_comparison(
    policy_id, new_start_date
)

For use with is_policy_change_significant in case you need to cancel the policy afterwards (otherwise, after cancellation, the cost becomes 0)

Source code in components/fr/public/policy/policy.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def precompute_policy_coverage_for_policy_comparison(
    policy_id: int, new_start_date: datetime.date
) -> _PrecomputedPolicyCoverage:
    """
    For use with is_policy_change_significant in case you need to cancel the policy afterwards (otherwise, after cancellation, the cost becomes 0)
    """
    from components.fr.internal.business_logic.health_pricing.policy import (
        get_policy_cost_split_by_primary_and_dependants,
    )

    policy = get_or_raise_missing_resource(Policy, policy_id)
    return _PrecomputedPolicyCoverage(
        policy.contract.health_coverage_key(new_start_date),
        get_policy_cost_split_by_primary_and_dependants(policy.id, new_start_date),
    )

components.fr.public.prevoyance

entities

PrevoyanceGuaranteeSpecBeneficiaryEligibilityCriteria dataclass 

PrevoyanceGuaranteeSpecBeneficiaryEligibilityCriteria(
    beneficiary_relationship_type,
    max_age_at_disability_start_date=None,
    min_covered_age=None,
    max_covered_age=None,
    disabled_max_covered_age=None,
    student_max_covered_age=None,
    unemployed_without_benefits_max_covered_age=None,
    apprentice_max_covered_age=None,
    is_dependent=None,
    is_under_supervision_or_in_psychiatric_hospital=None,
    paid_for_the_funeral=None,
)

Bases: DataClassJsonMixin

This dataclass describes the beneficiary criteria to be eligible to a prevoyance guarantee spec.

Only specific beneficiary could benefit from the coverage of certain guarantees (eg rente education only for children at charge, member in case of work stoppage)

apprentice_max_covered_age class-attribute instance-attribute 
apprentice_max_covered_age = None
beneficiary_relationship_type instance-attribute 
beneficiary_relationship_type
disabled_max_covered_age class-attribute instance-attribute 
disabled_max_covered_age = None
is_dependent class-attribute instance-attribute 
is_dependent = None
is_under_supervision_or_in_psychiatric_hospital class-attribute instance-attribute 
is_under_supervision_or_in_psychiatric_hospital = None
max_age_at_disability_start_date class-attribute instance-attribute 
max_age_at_disability_start_date = None
max_covered_age class-attribute instance-attribute 
max_covered_age = None
min_covered_age class-attribute instance-attribute 
min_covered_age = None
paid_for_the_funeral class-attribute instance-attribute 
paid_for_the_funeral = None
student_max_covered_age class-attribute instance-attribute 
student_max_covered_age = None
unemployed_without_benefits_max_covered_age class-attribute instance-attribute 
unemployed_without_benefits_max_covered_age = None

PrevoyanceGuaranteeSpecEligibilityCriteria dataclass 

PrevoyanceGuaranteeSpecEligibilityCriteria(
    event_criteria, member_criteria, beneficiary_criteria
)

Bases: DataClassJsonMixin

These criteria determine in which circumstances guarantees are eligible and could be covered.

When in such circumstances, we would only create / suggest claims with the eligible guarantees (eg for a death: death lump sum, rente education, funerals, etc).

Criteria are of different nature: - Event: guarantees only apply to specific events, in specific situation (eg. accidental death) - Member: guarantees eligibility could depend on the member tenure, on the member dependents, on their disability condition etc - Beneficiary: some guarantees only apply to specific beneficiaries (eg rente education only for children at charge)

Exploration can be found here https://docs.google.com/spreadsheets/d/1yyLaVasQnwSVhASZ3c41d7hFkp56RxiRmSwlMBHj2Mg/edit?usp=sharing ⧉ https://docs.google.com/document/d/1qIsYQgD6Iy9dLN61iKAbAD0S-LLHRCcmvE_owSyvFbk/edit?usp=sharing ⧉

beneficiary_criteria instance-attribute 
beneficiary_criteria
event_criteria instance-attribute 
event_criteria
member_criteria instance-attribute 
member_criteria

PrevoyanceGuaranteeSpecEventEligibilityCriteria dataclass 

PrevoyanceGuaranteeSpecEventEligibilityCriteria(
    event_type,
    event_user_relationship_type,
    death_event_user_min_age=None,
    death_event_user_max_age=None,
    event_is_not_covered_by_secu=None,
    event_is_covered_in_parental_leave=None,
    event_is_accidental=None,
    event_is_accidental_or_after_professional_disease=None,
    event_required_assistance_from_third_party=None,
    event_has_led_to_hospitalisation=None,
    event_is_long_term_condition=None,
    work_stoppage_valid_reasons=None,
    work_stoppage_min_duration_days=None,
    death_is_before_member_death=None,
    death_is_simultaneous_partner_death=None,
    simultaneous_partner_death_max_delay_months=None,
    disability_is_atmp=None,
    disability_rate_min=None,
    disability_rate_max=None,
    disability_categories=None,
)

Bases: DataClassJsonMixin

This dataclass describes the prevoyance event criteria to be eligible to a prevoyance guarantee spec

Guarantee would only cover specific events based on these criteria (eg. type of event, if death is accidental or on the disability rate)

death_event_user_max_age class-attribute instance-attribute 
death_event_user_max_age = None
death_event_user_min_age class-attribute instance-attribute 
death_event_user_min_age = None
death_is_before_member_death class-attribute instance-attribute 
death_is_before_member_death = None
death_is_simultaneous_partner_death class-attribute instance-attribute 
death_is_simultaneous_partner_death = None
disability_categories class-attribute instance-attribute 
disability_categories = None
disability_is_atmp class-attribute instance-attribute 
disability_is_atmp = None
disability_rate_max class-attribute instance-attribute 
disability_rate_max = None
disability_rate_min class-attribute instance-attribute 
disability_rate_min = None
event_has_led_to_hospitalisation class-attribute instance-attribute 
event_has_led_to_hospitalisation = None
event_is_accidental class-attribute instance-attribute 
event_is_accidental = None
event_is_accidental_or_after_professional_disease class-attribute instance-attribute 
event_is_accidental_or_after_professional_disease = None
event_is_covered_in_parental_leave class-attribute instance-attribute 
event_is_covered_in_parental_leave = None
event_is_long_term_condition class-attribute instance-attribute 
event_is_long_term_condition = None
event_is_not_covered_by_secu class-attribute instance-attribute 
event_is_not_covered_by_secu = None
event_required_assistance_from_third_party class-attribute instance-attribute 
event_required_assistance_from_third_party = None
event_type instance-attribute 
event_type
event_user_relationship_type instance-attribute 
event_user_relationship_type
simultaneous_partner_death_max_delay_months class-attribute instance-attribute 
simultaneous_partner_death_max_delay_months = None
work_stoppage_min_duration_days class-attribute instance-attribute 
work_stoppage_min_duration_days = None
work_stoppage_valid_reasons class-attribute instance-attribute 
work_stoppage_valid_reasons = None

PrevoyanceGuaranteeSpecMemberEligibilityCriteria dataclass 

PrevoyanceGuaranteeSpecMemberEligibilityCriteria(
    min_tenure_months=None,
    max_tenure_months=None,
    max_tenure_or_less_than_200_hours_per_trimester=None,
    min_hours_worked_in_trimester=None,
    max_hours_worked_in_trimester=None,
    min_covered_presence_days=None,
    has_dependent_children=None,
    has_dependent=None,
    is_married_or_pacsed=None,
)

Bases: DataClassJsonMixin

This dataclass describes the member criteria to be eligible to a prevoyance guarantee spec

It could be based on the member tenure in the company at the event date or on their dependents at charge

has_dependent class-attribute instance-attribute 
has_dependent = None
has_dependent_children class-attribute instance-attribute 
has_dependent_children = None
is_married_or_pacsed class-attribute instance-attribute 
is_married_or_pacsed = None
max_hours_worked_in_trimester class-attribute instance-attribute 
max_hours_worked_in_trimester = None
max_tenure_months class-attribute instance-attribute 
max_tenure_months = None
max_tenure_or_less_than_200_hours_per_trimester class-attribute instance-attribute 
max_tenure_or_less_than_200_hours_per_trimester = None
min_covered_presence_days class-attribute instance-attribute 
min_covered_presence_days = None
min_hours_worked_in_trimester class-attribute instance-attribute 
min_hours_worked_in_trimester = None
min_tenure_months class-attribute instance-attribute 
min_tenure_months = None

queries

PrevoyanceCCNContstraintEntity dataclass 

PrevoyanceCCNContstraintEntity(
    id, ccn_id, non_cadre_coverage_required
)

Bases: DataClassJsonMixin

Represents a constraint for a prevoyance contract CCN.

Attributes:

Name Type Description
id int

The unique identifier of the constraint.
ccn_id int

The ID of the CCN.
non_cadre_coverage_required bool

Indicates if non-cadre coverage for all contracts on the ccn.
ccn_id instance-attribute 
ccn_id
id instance-attribute 
id
non_cadre_coverage_required instance-attribute 
non_cadre_coverage_required

PrevoyanceParticipationCCNContstraintEntity dataclass 

PrevoyanceParticipationCCNContstraintEntity(
    id, ccn_id, professional_category, minimum_participation
)

Bases: DataClassJsonMixin

Represents a participation constraint for a prevoyance contract CCN.

Attributes:

Name Type Description
id int

The unique identifier of the constraint.
ccn_id int

The ID of the CCN.
professional_category Optional[ProfessionalCategory]

The professional category.
minimum_participation float

The minimum participation required for the professional category on the ccn.
ccn_id instance-attribute 
ccn_id
id instance-attribute 
id
minimum_participation instance-attribute 
minimum_participation
professional_category instance-attribute 
professional_category

get_prevoyance_ccn_constraints 

get_prevoyance_ccn_constraints(ccn_ids)

Return the list of PrevoyanceCCNCoinstraints for a list of ccn_ids.

Source code in components/fr/public/prevoyance/queries.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def get_prevoyance_ccn_constraints(
    ccn_ids: list[int],
) -> list[PrevoyanceCCNContstraintEntity] | None:
    """
    Return the list of PrevoyanceCCNCoinstraints for a list of ccn_ids.
    """
    prevoyance_ccn_constraints = (
        current_session.query(PrevoyanceCCNConstraint)  # noqa: ALN085
        .filter(PrevoyanceCCNConstraint.ccn_id.in_(ccn_ids))
        .all()
    )
    if len(prevoyance_ccn_constraints) == 0:
        return None

    return [
        PrevoyanceCCNContstraintEntity(
            id=prevoyance_ccn_constraint.id,  # type: ignore[arg-type]
            ccn_id=prevoyance_ccn_constraint.ccn_id,
            non_cadre_coverage_required=prevoyance_ccn_constraint.non_cadre_coverage_required,
        )
        for prevoyance_ccn_constraint in prevoyance_ccn_constraints
    ]

get_prevoyance_participation_ccn_constraints 

get_prevoyance_participation_ccn_constraints(
    ccn_id, professional_category=None
)

Return the list of CCN constraints that apply to the company's prevoyance contract.

Source code in components/fr/public/prevoyance/queries.py 
 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
123
124
125
126
127
128
129
def get_prevoyance_participation_ccn_constraints(
    ccn_id: int,
    professional_category: ProfessionalCategory | None = None,
) -> PrevoyanceParticipationCCNContstraintEntity | None:
    """
    Return the list of CCN constraints that apply to the company's prevoyance contract.
    """
    ccn = current_session.get(CCN, ccn_id)
    if not ccn:
        return None

    prevoyance_participation_ccn_constraints = (
        current_session.query(PrevoyanceParticipationCCNConstraint)  # noqa: ALN085
        .filter(
            PrevoyanceParticipationCCNConstraint.ccn_id == ccn.id,
            # We always include all or none entries so that we can apply them to `cadres/non-cadres` as well.
            or_(
                PrevoyanceParticipationCCNConstraint.professional_category
                == ProfessionalCategory.all,
                PrevoyanceParticipationCCNConstraint.professional_category == None,
                PrevoyanceParticipationCCNConstraint.professional_category
                == professional_category,
            ),
        )
        .all()
    )

    if len(prevoyance_participation_ccn_constraints) == 0:
        return None

    # With the DB constraints we can only ever have one entry for a given ccn_id and professional_category, but we check just in case and for typing.
    if len(prevoyance_participation_ccn_constraints) > 1:
        current_logger.error(
            f"Multiple participation constraints found for ccn_id={ccn_id}, professional_category={professional_category}, returning None and not handling ccn compliance in this case."
        )

        return None

    prevoyance_participation_ccn_constraint = prevoyance_participation_ccn_constraints[
        0
    ]

    return PrevoyanceParticipationCCNContstraintEntity(
        id=prevoyance_participation_ccn_constraint.id,  # type: ignore[arg-type]
        ccn_id=prevoyance_participation_ccn_constraint.ccn_id,
        professional_category=prevoyance_participation_ccn_constraint.professional_category,
        minimum_participation=prevoyance_participation_ccn_constraint.minimum_participation,  # type: ignore[arg-type]
    )

get_work_stoppages_for_users 

get_work_stoppages_for_users(
    user_ids, ever_active_during_period=None
)

Get the list of work stoppages for a list of users.

Source code in components/fr/public/prevoyance/queries.py 
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def get_work_stoppages_for_users(
    user_ids: Iterable[int],
    ever_active_during_period: tuple[date, date] | None = None,
) -> dict[int, list[WorkStoppage]]:
    """
    Get the list of work stoppages for a list of users.
    """
    from components.fr.internal.models.employment import Employment
    from components.fr.internal.prevoyance_claim_management.models.internalized_work_stoppage import (
        InternalizedWorkStoppage,
    )
    from components.fr.internal.prevoyance_claim_management.models.internalized_work_stoppage_info import (
        InternalizedWorkStoppageInfo,
    )

    query = (
        current_session.query(  # noqa: ALN085
            InternalizedWorkStoppage.id,
            InternalizedWorkStoppageInfo.start_date,
            InternalizedWorkStoppageInfo.end_date,
            InternalizedWorkStoppageInfo.reason,
            Employment.company_id,
            Employment.user_id,
        )
        .select_from(InternalizedWorkStoppage)
        .join(InternalizedWorkStoppage.employment)
        .join(InternalizedWorkStoppage.internalized_work_stoppage_infos)  # type: ignore[arg-type]
        .filter(
            Employment.user_id.in_(user_ids),
            InternalizedWorkStoppageInfo.is_cancelled.is_(False),
            InternalizedWorkStoppageInfo.is_obsolete.is_(False),
        )
    )

    if ever_active_during_period is not None:
        period_start, period_end = ever_active_during_period
        query = query.filter(
            InternalizedWorkStoppageInfo.start_date <= period_end,
            InternalizedWorkStoppageInfo.end_date >= period_start,
        )

    work_stoppage_by_user_id_mapping: dict[int, list[WorkStoppage]] = defaultdict(list)

    for row in query.all():
        work_stoppage_by_user_id_mapping[row.user_id].append(
            WorkStoppage(
                user_id=row.user_id,
                company_id=row.company_id,
                reason=row.reason,
                start_date=row.start_date,
                end_date=row.end_date,
            )
        )

    return dict(work_stoppage_by_user_id_mapping)

components.fr.public.queries

command_logs

get_command_logs 

get_command_logs(start_at, end_at)
Source code in components/fr/public/queries/command_logs.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def get_command_logs(start_at: datetime, end_at: datetime) -> list[CommandLogEntity]:  # noqa: D103
    logs = current_session.scalars(
        select(CommandLog).filter(
            CommandLog.created_at >= start_at, CommandLog.created_at < end_at
        )
    )

    return [
        CommandLogEntity(
            id=log.id,
            command=log.command,
            run_at=log.run_at,
            completed_at=log.completed_at,
            success=log.success,
            model_slug="commandlog",
        )
        for log in logs
    ]

policy_direct_billing_contract

get_policy_direct_billing_contract_on 

get_policy_direct_billing_contract_on(policy_id, date)
Source code in components/fr/internal/business_logic/policy_direct_billing_contract/queries/policy_direct_billing_contract.py 
25
26
27
28
29
def get_policy_direct_billing_contract_on(
    policy_id: int, date: date
) -> PolicyDirectBillingContract | None:
    policy = get_or_raise_missing_resource(Policy, policy_id)
    return policy.active_direct_billing_contract_on(date)

unpaid_leave

get_unpaid_leave_settings_from_policy 

get_unpaid_leave_settings_from_policy(policy_id)
Source code in components/fr/internal/business_logic/unpaid_leave/queries/unpaid_leave_settings.py 
21
22
23
24
25
26
27
28
29
30
def get_unpaid_leave_settings_from_policy(policy_id: int) -> UnpaidLeaveSettings:
    employment = get_policy_employment(policy_id)

    company_name = get_company_name_from_enrollment(employment.id) if employment else ""
    unpaid_leave = get_current_unpaid_leave_from_policy_or_none(policy_id)

    if unpaid_leave is None:
        return UnpaidLeaveSettings(is_unpaid_leave=False, company_name=company_name)

    return UnpaidLeaveSettings.from_unpaid_leave(unpaid_leave, company_name)

get_unpaid_leave_settings_from_user_health_contract 

get_unpaid_leave_settings_from_user_health_contract(
    user_id, health_contract_id, on_date
)
Source code in components/fr/internal/business_logic/unpaid_leave/queries/unpaid_leave_settings.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def get_unpaid_leave_settings_from_user_health_contract(
    user_id: int,
    health_contract_id: int,
    on_date: date,
) -> UnpaidLeaveSettings | None:
    company = get_company_from_contract(contract_id=str(health_contract_id))

    employment = get_user_active_employment_in_company_on_or_after(
        user_id=user_id,
        company_id=int(company.id),
        active_date=on_date,
    )
    if employment:
        return get_unpaid_leave_settings_from_employment(employment_id=employment.id)
    return None

components.fr.public.scim_api

adapter

FrScimAdapter 

FrScimAdapter()

Bases: GenericScimAdapter

SCIM adapter for fr_api.

Source code in components/fr/public/scim_api/adapter.py 
22
23
24
def __init__(self) -> None:
    super().__init__()
    self.profile_service = ProfileService.create(app_name=AppName.ALAN_FR)
create_app_user 
create_app_user(first_name, last_name, email)

Create a user with the given first and last name. and returns the user ID.

Source code in components/fr/public/scim_api/adapter.py 
71
72
73
74
75
76
77
78
79
@override
def create_app_user(
    self, first_name: str, last_name: str, email: str
) -> int | uuid.UUID:
    """
    Create a user with the given first and last name. and returns the user ID.
    """
    user = create_profile_with_user(first_name=first_name, last_name=last_name)
    return user.id
get_scim_users_data 
get_scim_users_data(alan_employees)

Returns the first and last name of users from a list of AlanEmployee objects.

Source code in components/fr/public/scim_api/adapter.py 
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
@override
def get_scim_users_data(
    self,
    alan_employees: list[AlanEmployee],  # type: ignore[override]
) -> dict[int | uuid.UUID, AlanEmployeeIdentity]:
    """
    Returns the first and last name of users from a list of AlanEmployee objects.
    """
    user_profiles = self.profile_service.get_profiles(
        profile_ids={
            alan_employee.user.profile_id for alan_employee in alan_employees
        }
    )
    user_profiles_dict = {
        user_profile.id: user_profile for user_profile in user_profiles
    }

    return {
        alan_employee.user_id: AlanEmployeeIdentity(
            first_name=user_profiles_dict[alan_employee.user.profile_id].first_name,
            last_name=user_profiles_dict[alan_employee.user.profile_id].last_name,
        )
        for alan_employee in alan_employees
        if alan_employee.user.profile_id in user_profiles_dict
    }
get_user_data 
get_user_data(user_id)

Returns user's first and last name by user_id.

Source code in components/fr/public/scim_api/adapter.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@override
def get_user_data(self, user_id: int | uuid.UUID) -> AlanEmployeeIdentity:
    """
    Returns user's first and last name by user_id.
    """
    if not isinstance(user_id, int):
        raise TypeError("User ID must be a int")

    user = current_session.query(User).where(User.id == user_id).one_or_none()  # noqa: ALN085
    if user is None:
        raise BaseErrorCode.missing_resource(f"User id {user_id} not found")
    user_profile = self.profile_service.get_or_raise_profile(
        profile_id=user.profile_id
    )

    return AlanEmployeeIdentity(
        first_name=user_profile.first_name, last_name=user_profile.last_name
    )
profile_service instance-attribute 
profile_service = create(app_name=ALAN_FR)

test

test_adapter

adapter 
adapter()

Fixture for the EsGenericScimAdapter instance.

Source code in components/fr/public/scim_api/test/test_adapter.py 
19
20
21
22
@pytest.fixture
def adapter() -> FrScimAdapter:
    """Fixture for the EsGenericScimAdapter instance."""
    return FrScimAdapter()
profile_service 
profile_service()

Fixture for the profile service.

Source code in components/fr/public/scim_api/test/test_adapter.py 
25
26
27
28
@pytest.fixture
def profile_service() -> ProfileService:
    """Fixture for the profile service."""
    return ProfileService.create(app_name=AppName.ALAN_FR)
test_create_app_user 
test_create_app_user(adapter, profile_service)

Test create_app_user creates a new user correctly.

Source code in components/fr/public/scim_api/test/test_adapter.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@pytest.mark.usefixtures("db")
def test_create_app_user(adapter, profile_service):
    """Test create_app_user creates a new user correctly."""
    user_id = adapter.create_app_user(
        first_name="John", last_name="Doe", email="john.doe@alan.eu"
    )

    created_user = current_session.query(User).filter(User.id == user_id).one_or_none()  # noqa: ALN085

    assert created_user is not None
    created_profile = profile_service.get_profile(profile_id=created_user.profile_id)
    assert created_profile is not None
    assert created_profile.first_name == "John"
    assert created_profile.last_name == "Doe"
test_get_scim_users_data 
test_get_scim_users_data(adapter)

Test get_scim_users_data returns correct mapping of user data.

Source code in components/fr/public/scim_api/test/test_adapter.py 
75
76
77
78
79
80
81
82
83
84
85
86
@pytest.mark.usefixtures("db")
def test_get_scim_users_data(adapter):
    """Test get_scim_users_data returns correct mapping of user data."""
    # Create test data
    user1, user2, employee1, employee2 = _provision_test_data()

    result = adapter.get_scim_users_data([employee1, employee2])
    assert len(result) == 2
    assert result[user1.id].first_name == "John"
    assert result[user1.id].last_name == "Doe"
    assert result[user2.id].first_name == "Jane"
    assert result[user2.id].last_name == "Smith"
test_get_user_data 
test_get_user_data(adapter)

Test get_user_data returns correct user identity.

Source code in components/fr/public/scim_api/test/test_adapter.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@pytest.mark.usefixtures("db")
def test_get_user_data(adapter):
    """Test get_user_data returns correct user identity."""
    # Create test data
    user1, _, _, _ = _provision_test_data()

    # Test with int
    result = adapter.get_user_data(user_id=user1.id)
    assert result.first_name == "John"
    assert result.last_name == "Doe"

    # Test with non-int
    with pytest.raises(TypeError):
        adapter.get_user_data(user_id=uuid.uuid4())

    # Test with non-existent user
    with pytest.raises(BaseErrorCode):
        adapter.get_user_data(user_id=999999)

components.fr.public.services

push_notifications

get_push_notification_logs_for_user 

get_push_notification_logs_for_user(
    app_name,
    app_user_id,
    notification_names,
    created_at__gte=None,
)

Return a list of all the push notification logs ever created for the given user and notification names.

Source code in components/fr/public/services/push_notifications.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def get_push_notification_logs_for_user(
    app_name: AppName,
    app_user_id: str,
    notification_names: list[BasePushNotificationName],
    created_at__gte: datetime | None = None,
) -> list[InMemoryPushNotificationLog]:
    """
    Return a list of all the push notification logs ever created for the given user and notification names.
    """
    return [
        InMemoryPushNotificationLog.from_model(notification_log=item)
        for item in current_session.query(PushNotificationLog).filter(  # noqa: ALN085
            PushNotificationLog.app_id == app_name,
            PushNotificationLog.app_user_id == app_user_id,
            PushNotificationLog.name.in_(notification_names),
            PushNotificationLog.created_at >= created_at__gte  # type: ignore[arg-type]
            if created_at__gte
            else True,
        )
    ]

push_notification_sender_async 

push_notification_sender_async(sender)
Source code in components/fr/internal/push_notifications/push_notification_sender.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def push_notification_sender_async(
    sender: Callable[P, PushNotificationParams | None],
) -> Callable[P, None]:
    @wraps(sender)
    def decorated_function(*args, **kwargs) -> None:  # type: ignore[no-untyped-def]
        pn_params: PushNotificationParams | None = sender(*args, **kwargs)

        if pn_params is None:
            return

        push_notification_logic.send_push_notification_async(
            notification_params=pn_params,
            commit=pn_params.commit,
        )

    return decorated_function

push_notification_sender_sync 

push_notification_sender_sync(sender)
Source code in components/fr/internal/push_notifications/push_notification_sender.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def push_notification_sender_sync(
    sender: Callable[P, PushNotificationParams | None],
) -> Callable[P, None]:
    @wraps(sender)
    def decorated_function(*args, **kwargs) -> None:  # type: ignore[no-untyped-def]
        pn_params: PushNotificationParams | None = sender(*args, **kwargs)

        if pn_params is None:
            return

        push_notification_logic.send_push_notification_sync(
            notification_params=pn_params,
            delete_token=delete_token,
            commit=pn_params.commit,
        )

    return decorated_function

components.fr.public.test_data_generator

get_test_data_generation_config 

get_test_data_generation_config()
Source code in components/fr/internal/admin_tools/fixtures/test_data_generation_config.py 
12
13
14
15
16
17
18
def get_test_data_generation_config() -> TestDataGeneratorConfig:
    return TestDataGeneratorConfig(
        patched_factories=patched_factories,
        handlers=get_fixture_handlers(),
        async_queue=MAIN_QUEUE,
        result_view_builder=result_view_builder,
    )

components.fr.public.tp_card

member_attributes

Member attribute definitions for FR TP card attributes.

can_request_physical_tp_card module-attribute 

can_request_physical_tp_card = MemberAttributeDefinition[
    bool
](
    name="can_request_physical_tp_card",
    display_name="Can request physical TP card",
    description="Whether the member can request a physical TP card",
    getter=_get_can_request_physical_tp_card,
    raw_type=bool,
)

date_physical_tp_card_shipped module-attribute 

date_physical_tp_card_shipped = MemberAttributeDefinition[
    date
](
    name="date_physical_tp_card_shipped",
    display_name="Date physical TP card shipped",
    description="Date the physical TP card was shipped",
    getter=_get_date_physical_tp_card_shipped,
    raw_type=date,
)

did_opt_out_physical_tp_card module-attribute 

did_opt_out_physical_tp_card = MemberAttributeDefinition[
    bool
](
    name="did_opt_out_physical_tp_card",
    display_name="Did opt out of physical TP card",
    description="Whether the member opted out of physical TP card",
    getter=_get_did_opt_out_physical_tp_card,
    raw_type=bool,
)

tp_card_contract_scheme module-attribute 

tp_card_contract_scheme = MemberAttributeDefinition[
    TpCardScheme
](
    name="tp_card_contract_scheme",
    display_name="TP card contract scheme",
    description="Contract-level TP card scheme for current enrollment",
    getter=_get_tp_card_contract_scheme,
    raw_type=TpCardScheme,
)

tp_card_scheme module-attribute 

tp_card_scheme = MemberAttributeDefinition[TpCardScheme](
    name="tp_card_scheme",
    display_name="TP card scheme",
    description="TP card scheme for current enrollment",
    getter=_get_tp_card_scheme,
    raw_type=TpCardScheme,
)

tp_card_scheme_reason module-attribute 

tp_card_scheme_reason = MemberAttributeDefinition[
    InsuranceProfileTpCardSchemeReason
](
    name="tp_card_scheme_reason",
    display_name="TP card scheme reason",
    description="Reason for the TP card scheme",
    getter=_get_tp_card_scheme_reason,
    raw_type=InsuranceProfileTpCardSchemeReason,
)

tp_card_status module-attribute 

tp_card_status = MemberAttributeDefinition[TpCardStatus](
    name="tp_card_status",
    display_name="TP card status",
    description="Status of the latest TP card",
    getter=_get_tp_card_status,
    raw_type=TpCardStatus,
)

tp_rights_activated module-attribute 

tp_rights_activated = MemberAttributeDefinition[bool](
    name="tp_rights_activated",
    display_name="TP rights activated",
    description="Whether TP rights are activated",
    getter=_get_tp_rights_activated,
    raw_type=bool,
)

tp_rights_referent module-attribute 

tp_rights_referent = MemberAttributeDefinition[str](
    name="tp_rights_referent",
    display_name="TP rights referent",
    description="Full name of the TP rights referent",
    getter=_get_tp_rights_referent,
    raw_type=str,
)

tp_rights_status module-attribute 

tp_rights_status = MemberAttributeDefinition[
    TpRightsStatus
](
    name="tp_rights_status",
    display_name="TP rights status",
    description="TP rights status",
    getter=_get_tp_rights_status,
    raw_type=TpRightsStatus,
    formatter=_format_tp_rights_status,
)

components.fr.public.user

actions

user_iban

change_user_iban 
change_user_iban(
    actor_id,
    user_id,
    iban_type,
    iban,
    comment=None,
    allow_same_policy=False,
    set_current_requires_sepa_signed=False,
    save=True,
    rollback_at_end=False,
    force_new_iban_creation=False,
)

Change the user's IBAN: - If changing the billing IBAN, change it directly and make it as current only if sepa is signed or is not required - If changing the settlement IBAN and the user is changing his own IBAN, change it directly - Otherwise create a pending SettlementIBANChange which will have to be reviewed by an operator before being applied

Source code in components/fr/internal/business_logic/user/actions/user_iban.py 
 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
def change_user_iban(
    actor_id: int | None,
    user_id: int,
    iban_type: IBANType,
    iban: str,
    comment: str | None = None,
    allow_same_policy: bool = False,
    set_current_requires_sepa_signed: bool = False,
    save: bool = True,
    rollback_at_end: bool | None = False,
    force_new_iban_creation: bool = False,
) -> IBAN | None:
    """Change the user's IBAN:
    - If changing the billing IBAN, change it directly and make it as current only if sepa is signed or is not required
    - If changing the settlement IBAN and the user is changing his own IBAN, change it directly
    - Otherwise create a pending SettlementIBANChange which will have to be reviewed by an operator before being applied
    """

    actor = (
        get_or_raise_missing_resource(User, actor_id) if actor_id is not None else None
    )
    user = get_or_raise_missing_resource(User, user_id)

    new_iban = None

    if iban_type == IBANType.billing:
        new_iban = _handle_change_user_billing_iban(
            user,
            iban,
            set_current_requires_sepa_signed,
            save,
            # Calling this function with save = False and rollback_at_end None will rollback twice: now, and in the
            # call below. This is probably not a wanted behaviour.
            rollback_at_end=rollback_at_end,
            force_new_iban_creation=force_new_iban_creation,
        )
    elif iban_type == IBANType.settlement:
        same_policy = actor_id == user_id
        if (
            not same_policy
            and actor
            and actor.insurance_profile
            and user.insurance_profile
        ):
            same_policy = (
                actor.insurance_profile.current_policy
                == user.insurance_profile.current_policy
            )
        on_same_policy = allow_same_policy and same_policy

        new_iban = _handle_change_user_settlement_iban(
            mandatory(actor),
            user,
            iban,
            on_same_policy,
            comment,
            save,
            rollback_at_end=rollback_at_end,
        )

    return new_iban

user_lifecycle

get_employee_onboarding_status 
get_employee_onboarding_status(user)
Source code in components/fr/internal/business_logic/user/actions/user_lifecycle.py 
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
def get_employee_onboarding_status(user: User) -> EmployeeOnboardingStatus | None:
    if not user:
        raise ErrorCode.missing_resource()

    has_been_employed = len(user.non_cancelled_employments) > 0

    if not has_been_employed:
        return None

    exemption = user.current_exemption

    active_employment = get_active_or_last_employment_via_relationship(user)

    # When an exemption as been terminated by admin, it has a termination_type whereas if it has been migrated through Marmot, it hasn't.
    # We don't want to redirect an ended non-migrated exemption to onboarding.
    exemption_has_been_terminated_by_admin = (
        exemption
        and exemption.termination_type
        and exemption.termination_type != EmployeeTerminationType.other
        # If the user has a new employment, it means that they should onboard as an employee
        and (
            # It shouldn't happen that the user has an exemption without an employment
            # but the employment can be cancelled if the exemption was ended without being active
            not active_employment
            or (
                active_employment.end_date
                and active_employment.end_date == exemption.end_date
            )
        )
    )

    if exemption and (not exemption.is_ended or exemption_has_been_terminated_by_admin):
        return None

    return get_insured_employee_onboarding_status(user)

create_profile_with_user

create_or_assign_profile_with_authenticatable_user 

create_or_assign_profile_with_authenticatable_user(
    profile_service,
    authentication_service,
    email,
    prehashed_password,
    password=None,
    mfa_required=None,
    first_name=None,
    last_name=None,
    birth_date=None,
    language=None,
    phone_number=None,
    gender=None,
    empty_user=None,
    email_verified=True,
)

Helper to create an authenticatable user (with credentials) and helps manage cross-country identities In this method we create a profile without identity elements, then get (or create) the corresponding user and try to set the credentials of this user. Set credentials handles the cases of conflict with existing identities. Finally, we set the profile email address if all steps before didn't raise

Source code in components/fr/public/user/create_profile_with_user.py 
 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
123
124
125
126
127
128
129
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
@inject_profile_service
@inject_authentication_service
def create_or_assign_profile_with_authenticatable_user(
    profile_service: ProfileService,
    authentication_service: AuthenticationService,
    email: str,
    prehashed_password: str,
    password: str | None = None,
    mfa_required: bool | None = None,
    first_name: str | None = None,
    last_name: str | None = None,
    birth_date: date | None = None,
    language: Lang | None = None,
    phone_number: str | None = None,
    gender: UserGender | None = None,
    empty_user: User | None = None,
    email_verified: bool = True,
) -> User:
    """
    Helper to create an authenticatable user (with credentials) and helps manage cross-country identities
    In this method we create a profile without identity elements, then get (or create) the corresponding user and try
    to set the credentials of this user. Set credentials handles the cases of conflict with existing identities.
    Finally, we set the profile email address if all steps before didn't raise
    """
    selected_language = language or Lang.french
    profile_id = None
    keycloak_id: uuid.UUID | None = None

    existing_user_in_same_country_with_email: User | None = (
        current_session.query(User).filter_by(email=email).one_or_none()  # noqa: ALN085
    )
    if existing_user_in_same_country_with_email is not None:
        current_logger.warning(
            f"can't set credentials: user with email {email} already exists"
        )
        raise BaseErrorCode.user_with_email_already_exists(
            message="User with this email already exists"
        )

    # Check if a profile with the same email already exists (aka a user from another country)
    existing_profile_with_email = profile_service.get_profile_by_email(email)

    if existing_profile_with_email:
        if existing_identity_with_email := authentication_service.get_identity_by_email(
            mandatory(existing_profile_with_email.email)
        ):
            if authentication_service.check_identity_password(
                identity_id=existing_identity_with_email.id,
                prehashed_password=prehashed_password,
            ):
                profile_id = existing_profile_with_email.id
                keycloak_id = existing_identity_with_email.id
                if empty_user is not None:
                    profile_service.merge_profile_into_another(
                        source_profile_id=empty_user.profile_id,
                        target_profile_id=profile_id,
                    )
                    empty_user.profile_id = profile_id
                    current_session.flush()
                _set_profile_data(
                    profile_service=profile_service,
                    profile_id=existing_profile_with_email.id,
                    first_name=first_name,
                    last_name=last_name,
                    birth_date=birth_date,
                    gender=gender,
                    selected_language=selected_language,
                    email=email,
                )

            else:
                raise BaseErrorCode.login_error()
        else:
            # this shouldn't happen but as email between keycloak and our backend is not always consistent it may
            raise ValueError(
                f"Profile with email {email} already exist bu no identity has this email address"
            )

    if empty_user is not None:
        profile_id = empty_user.profile_id
        _set_profile_data(
            profile_service=profile_service,
            profile_id=profile_id,
            first_name=first_name,
            last_name=last_name,
            birth_date=birth_date,
            gender=gender,
            selected_language=selected_language,
            email=email,
        )

    if not profile_id:
        profile_id = empty_user.profile_id if empty_user is not None else uuid.uuid4()
        fr_user_id = IdGenerator.generate_id(User)
        # if there is an email this will also create the identity
        profile_id = profile_service.create_profile(
            profile_id=profile_id,
            email=email,
            first_name=first_name,
            last_name=last_name,
            birth_date=birth_date,
            preferred_language=LanguageMapper.to_entity(selected_language),
            # Insurance onboarding doesn't require a gender, but external teleconsultation does
            gender=GenderMapper.to_entity(gender) if gender else None,
            global_user_id=str(fr_user_id),
        )
        profile_service.change_phone_number(profile_id, phone_number=phone_number)

    user: User | None = (
        current_session.query(User).filter_by(profile_id=profile_id).one_or_none()  # noqa: ALN085
    )

    # Multi-country user (profile already profile in another country)
    if not user:
        user = User(
            first_name=first_name,
            last_name=last_name,
            profile_id=profile_id,
            birth_date=birth_date,
            lang=selected_language,
            phone=phone_number,
            gender=gender,
        )

        current_session.add(user)
        current_session.flush()

    profile_service.ensure_user_profile_link(
        profile_id=profile_id,
        user_id=str(user.id),
        app_name=AppName.ALAN_FR,
    )

    if keycloak_id is not None:
        # we found a matching identity,
        # TODO: @thibaut.caillierez: remove this branch once profile is directly linked to the identity without the user
        user.keycloak_id = keycloak_id
        user.email = email
    else:
        identity = mandatory(
            authentication_service.get_keycloak_identity_by_profile_id(profile_id)
        )
        if password is not None:
            if not is_password_strong_enough(password):
                raise BaseErrorCode.password_not_strong_enough()
            # Overwrite prehashed password with fresh prehashed password
            prehashed_password = PasswordMixin.hash_password(password)
        authentication_service.set_identity_credentials(
            identity_id=identity.id,
            prehashed_password=prehashed_password,
            email=email,
            is_email_verified=email_verified,
            mfa_required=mfa_required,
        )
    return user

create_profile_with_user 

create_profile_with_user(
    *,
    profile_service,
    email=None,
    first_name=None,
    last_name=None,
    birth_date=None,
    language=None,
    phone_number=None,
    gender=None
)

Create a new user (and profile) with the given parameters.

Source code in components/fr/public/user/create_profile_with_user.py 
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
@inject_profile_service
def create_profile_with_user(
    *,
    profile_service: ProfileService,
    email: str | None = None,
    first_name: str | None = None,
    last_name: str | None = None,
    birth_date: date | None = None,
    language: Lang | None = None,
    phone_number: str | None = None,
    gender: UserGender | None = None,
) -> User:
    """
    Create a new user (and profile) with the given parameters.
    """
    selected_language = language or Lang.french

    if email is not None:
        existing_profile_with_email = profile_service.get_profile_by_email(email)

        if existing_profile_with_email:
            raise ValueError(f"Profile with email {email} already exists")

    fr_user_id = IdGenerator.generate_id(User)

    profile_id = profile_service.create_profile(
        email=email,
        first_name=first_name,
        last_name=last_name,
        birth_date=birth_date,
        preferred_language=LanguageMapper.to_entity(selected_language),
        gender=GenderMapper.to_entity(gender),
        global_user_id=str(fr_user_id),
    )
    profile_service.change_phone_number(profile_id, phone_number=phone_number)

    user: User | None = current_session.scalars(
        select(User).where(User.profile_id == profile_id)
    ).one_or_none()

    if not user:
        raise RuntimeError(
            f"User with profile_id {profile_id} has not been created: ProfileService might be misconfigured"
        )

    return user

insurance_profile

get_ssn_and_ntt_for_user 

get_ssn_and_ntt_for_user(user_id)

Get the SSN and NTT for an existing user.

Source code in components/fr/public/user/insurance_profile.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def get_ssn_and_ntt_for_user(
    user_id: int,
) -> tuple[str | None, str | None]:
    """
    Get the SSN and NTT for an existing user.
    """
    insurance_profile = (
        current_session.query(InsuranceProfile)  # noqa: ALN085
        .filter(
            InsuranceProfile.user_id == user_id,
        )
        .one_or_none()
    )

    if not insurance_profile:
        return None, None

    return insurance_profile.ssn, insurance_profile.ntt

get_ssn_and_ntt_for_users 

get_ssn_and_ntt_for_users(user_ids)

Get the SSN and NTT for users.

Source code in components/fr/public/user/insurance_profile.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def get_ssn_and_ntt_for_users(
    user_ids: Iterable[int],
) -> dict[int, tuple[str | None, str | None]]:
    """
    Get the SSN and NTT for users.
    """
    insurance_profiles: ScalarResult[InsuranceProfile] = current_session.execute(
        select(InsuranceProfile).filter(InsuranceProfile.user_id.in_(user_ids))
    ).scalars()

    return {
        insurance_profile.user_id: (insurance_profile.ssn, insurance_profile.ntt)
        for insurance_profile in insurance_profiles
    }

set_ssn_ntt_on_user 

set_ssn_ntt_on_user(user_id, ssn=None, ntt=None)

Set the SSN or NTT for an existing user.

Do NOT commit the session.

Source code in components/fr/public/user/insurance_profile.py 
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
def set_ssn_ntt_on_user(
    user_id: int,
    ssn: str | None = None,
    ntt: str | None = None,
) -> None:
    """
    Set the SSN or NTT for an existing user.

    Do NOT commit the session.
    """
    if not ssn and not ntt:
        raise ValueError("Please provide at least one of SSN or NTT.")

    insurance_profile = (
        current_session.query(InsuranceProfile)  # noqa: ALN085
        .filter(
            InsuranceProfile.user_id == user_id,
        )
        .one_or_none()
    )

    if not insurance_profile:
        insurance_profile = InsuranceProfile(user_id=user_id)
        current_session.add(insurance_profile)

    if ssn and insurance_profile.ssn != ssn:
        current_logger.info(f"Updating SSN for user {user_id}.")
        insurance_profile.ssn = ssn

    if ntt and insurance_profile.ntt != ntt:
        current_logger.info(f"Updating NTT for user {user_id}.")
        insurance_profile.ntt = ntt

    current_session.flush()

member_attributes

has_civil_servant_exemption_justification module-attribute 

has_civil_servant_exemption_justification = MemberAttributeDefinition[
    bool
](
    name="has_civil_servant_exemption_justification",
    display_name="Justificatif de dispense agents publics",
    description="Whether the user's account requires civil servant exemption justification",
    getter=_get_has_civil_servant_exemption_justification,
    raw_type=bool,
)

has_prevoyance_work_stoppages module-attribute 

has_prevoyance_work_stoppages = MemberAttributeDefinition[
    bool
](
    name="has_prevoyance_work_stoppages",
    display_name="Has prevoyance work stoppages",
    description="Whether the user has prevoyance work stoppage claims",
    getter=_get_has_prevoyance_work_stoppages,
    raw_type=bool,
)

has_unpaid_leave_mandatory_coverage module-attribute 

has_unpaid_leave_mandatory_coverage = MemberAttributeDefinition[
    bool
](
    name="has_unpaid_leave_mandatory_coverage",
    display_name="Couverture obligatoire congé sans solde",
    description="Whether the user's account has unpaid leave mandatory coverage",
    getter=_get_has_unpaid_leave_mandatory_coverage,
    raw_type=bool,
)

include_option_billed_to_employee_in_payroll_data module-attribute 

include_option_billed_to_employee_in_payroll_data = MemberAttributeDefinition[
    bool
](
    name="include_option_billed_to_employee_in_payroll_data",
    display_name="Inclure option à la charge du salarié dans la paie",
    description="Whether option billed to employee is included in payroll data",
    getter=_get_include_option_billed_to_employee_in_payroll_data,
    raw_type=bool,
)

is_civil_servant module-attribute 

is_civil_servant = MemberAttributeDefinition[bool](
    name="is_civil_servant",
    display_name="Is civil servant",
    description="Whether the user is a civil servant",
    getter=_get_is_civil_servant,
    raw_type=bool,
)

is_disabled_ani_billing module-attribute 

is_disabled_ani_billing = MemberAttributeDefinition[bool](
    name="is_disabled_ani_billing",
    display_name="Facturation ANI désactivée",
    description="Whether ANI billing is disabled for the user's account",
    getter=_get_is_disabled_ani_billing,
    raw_type=bool,
)

is_under_ecology_contract module-attribute 

is_under_ecology_contract = MemberAttributeDefinition[bool](
    name="is_under_ecology_contract",
    display_name="Contrat écologie",
    description="Whether the user is under an ecology contract",
    getter=_get_is_under_ecology_contract,
    raw_type=bool,
)

termination_letter_sent_at module-attribute 

termination_letter_sent_at = MemberAttributeDefinition[
    datetime
](
    name="termination_letter_sent_at",
    display_name="Termination letter sent at",
    description="Timestamp when termination letter was sent",
    getter=_get_termination_letter_sent_at,
    raw_type=datetime,
)

termination_letter_status module-attribute 

termination_letter_status = MemberAttributeDefinition[str](
    name="termination_letter_status",
    display_name="Termination letter status",
    description="Status of the legacy contract termination letter",
    getter=_get_termination_letter_status,
    raw_type=str,
)

user_states module-attribute 

user_states = MemberAttributeDefinition[
    list[UserStateToDisplay]
](
    name="user_states",
    display_name="User states",
    description="Categorized user states for display",
    getter=_get_user_states,
    raw_type=list[UserStateToDisplay],
    formatter=_format_user_states_to_display,
)

primary_member_attributes

primary_user_company_prevoyance_contracts module-attribute 

primary_user_company_prevoyance_contracts = MemberAttributeDefinition[
    list[PrevoyanceContractInfo]
](
    name="primary_user_company_prevoyance_contracts",
    display_name="Primary user company prevoyance contracts",
    description="Active/upcoming prevoyance contracts for the primary user's company",
    getter=_get_primary_user_company_prevoyance_contracts,
    raw_type=list[PrevoyanceContractInfo],
    formatter=_format_prevoyance_contracts,
)

primary_user_professional_category module-attribute 

primary_user_professional_category = MemberAttributeDefinition[
    ProfessionalCategory
](
    name="primary_user_professional_category",
    display_name="Primary user professional category",
    description="Professional category of the primary user's active employment",
    getter=_get_primary_user_professional_category,
    raw_type=ProfessionalCategory,
    formatter=_format_professional_category,
)

primary_user_states module-attribute 

primary_user_states = MemberAttributeDefinition[
    list[UserStateToDisplay]
](
    name="primary_user_states",
    display_name="Primary user states",
    description="Categorized user states for the primary user (account holder)",
    getter=_get_primary_user_states,
    raw_type=list[UserStateToDisplay],
    formatter=_format_user_states_to_display,
)

queries

EmploymentSearchResult dataclass 

EmploymentSearchResult(
    id, company_id, start_date, end_date, invite_email
)

Employment data returned by user search queries.

company_id instance-attribute 
company_id
end_date instance-attribute 
end_date
id instance-attribute 
id
invite_email instance-attribute 
invite_email
start_date instance-attribute 
start_date

UserSearchResult dataclass 

UserSearchResult(
    id,
    first_name,
    last_name,
    email,
    ssn,
    ntt,
    invite_email,
    employments,
)

User data returned by user search queries.

email instance-attribute 
email
employments instance-attribute 
employments
first_name instance-attribute 
first_name
id instance-attribute 
id
invite_email instance-attribute 
invite_email
last_name instance-attribute 
last_name
ntt instance-attribute 
ntt
ssn instance-attribute 
ssn

get_user 

get_user(user_id)

Get user data

Parameters:

Name Type Description Default
user_id int

user id

 required

Returns:

Type Description
User

French user data
Source code in components/fr/public/user/queries.py 
27
28
29
30
31
32
33
34
35
36
def get_user(user_id: int) -> User:
    """Get user data

    Args:
        user_id: user id

    Returns:
        French user data
    """
    return get_or_raise_missing_resource(User, user_id)

get_user_for_clinic 

get_user_for_clinic(user_id)

Get user data for clinic with linked data

Parameters:

Name Type Description Default
user_id int

user id

 required

Returns:

Type Description
User

French user data
Source code in components/fr/public/user/queries.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get_user_for_clinic(user_id: int) -> User:
    """Get user data for clinic with linked data

    Args:
        user_id: user id

    Returns:
        French user data
    """
    return get_or_raise_missing_resource(
        User,
        user_id,
        options=[
            selectinload(User.address),
            selectinload(User.companies),
            selectinload(User.insurance_profile),
        ],
    )

get_user_from_email 

get_user_from_email(email_address)
Source code in components/fr/public/user/queries.py 
15
16
17
18
19
20
def get_user_from_email(email_address: str) -> User | None:  # noqa: D103
    from components.fr.internal.business_logic.user.queries.user import (
        get_user_from_email as internal_get_user_from_email,
    )

    return internal_get_user_from_email(email_address=email_address)

get_user_pro_email_with_perso_fallback 

get_user_pro_email_with_perso_fallback(user_id)

Get the pro email of the user, falling back to personal email if pro email is not set.

Parameters:

Name Type Description Default
user_id int

The user ID

 required

Returns:

Type Description
str | None

The pro email if available, otherwise the personal email.
Source code in components/fr/public/user/queries.py 
85
86
87
88
89
90
91
92
93
94
95
def get_user_pro_email_with_perso_fallback(user_id: int) -> str | None:
    """Get the pro email of the user, falling back to personal email if pro email is not set.

    Args:
        user_id: The user ID

    Returns:
        The pro email if available, otherwise the personal email.
    """
    user = get_user(user_id)
    return user.pro_email_with_perso_fallback

get_users 

get_users(user_ids)

Get multiple FR users by their IDs in a single database query

Parameters:

Name Type Description Default
user_ids list[int]

List of user IDs to fetch

 required

Returns:

Name Type Description
dict[int, User]

Dictionary mapping user IDs to User objects with insurance_profile eagerly loaded.
Note dict[int, User]

If some IDs don't exist, they won't be in the result.
Source code in components/fr/public/user/queries.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def get_users(user_ids: list[int]) -> dict[int, User]:
    """Get multiple FR users by their IDs in a single database query

    Args:
        user_ids: List of user IDs to fetch

    Returns:
        Dictionary mapping user IDs to User objects with insurance_profile eagerly loaded.
        Note: If some IDs don't exist, they won't be in the result.
    """
    if not user_ids:
        return {}

    users: list[User] = list(
        current_session.scalars(
            select(User)
            .where(User.id.in_(user_ids))
            .options(joinedload(User.insurance_profile))
        )
        .unique()
        .all()
    )

    return {user.id: user for user in users}

is_user 

is_user(user_id)
Source code in components/fr/public/user/queries.py 
23
24
def is_user(user_id: int) -> bool:  # noqa: D103
    return get_resource_or_none(User, user_id) is not None

search_users_by_invite_email 

search_users_by_invite_email(invite_email)

Search users by their employment invite email.

Parameters:

Name Type Description Default
invite_email str

The invite email to search for.

 required

Returns:

Type Description
list[UserSearchResult]

List of matching users with their employments.
Source code in components/fr/public/user/queries.py 
123
124
125
126
127
128
129
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def search_users_by_invite_email(invite_email: str) -> list[UserSearchResult]:
    """Search users by their employment invite email.

    Args:
        invite_email: The invite email to search for.

    Returns:
        List of matching users with their employments.
    """
    from components.fr.internal.models.employment import Employment

    employments = (
        current_session.scalars(
            select(Employment)
            .where(Employment.invite_email == invite_email)
            .where(Employment.is_cancelled.is_(False))
            .options(
                joinedload(Employment.user).joinedload(User.insurance_profile),
                joinedload(Employment.user).joinedload(User.employments),
            )
        )
        .unique()
        .all()
    )

    results: list[UserSearchResult] = []
    seen_user_ids: set[int] = set()

    for emp in employments:
        user = emp.user
        if user.id in seen_user_ids:
            continue
        seen_user_ids.add(user.id)

        insurance_profile = user.insurance_profile
        results.append(
            UserSearchResult(
                id=user.id,
                first_name=user.first_name,
                last_name=user.last_name,
                email=user.email,
                ssn=insurance_profile.ssn if insurance_profile else None,
                ntt=insurance_profile.ntt if insurance_profile else None,
                invite_email=user.last_employment_invite_email,
                employments=[
                    EmploymentSearchResult(
                        id=str(e.id),
                        company_id=str(e.company_id),
                        start_date=e.start_date,
                        end_date=e.end_date,
                        invite_email=e.invite_email,
                    )
                    for e in sorted(user.employments, key=lambda x: x.start_date)
                ],
            )
        )

    return results

components.fr.public.validators

ape_validator

ApeValidationError

Bases: Exception

wrong_format staticmethod 
wrong_format(code)
Source code in components/fr/public/validators/ape_validator.py 
2
3
4
5
6
@staticmethod
def wrong_format(code):  # type: ignore[no-untyped-def]  # noqa: D102
    return ApeValidationError(
        f"The ape code '{code}' must be composed of 4 digits + 1 letter"
    )

ApeValidator

  • Allows to check if a APE code (or NAF) is valid
are_ape_codes_equal staticmethod 
are_ape_codes_equal(ape_code1, ape_code2)

Cleans and validates an APE code (or NAF) and compare them.

NB 1: If either or both values are invalid, returns False. NB 2: If both values are None, returns True.

Source code in components/fr/public/validators/ape_validator.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@staticmethod
def are_ape_codes_equal(ape_code1: str | None, ape_code2: str | None) -> bool:
    """
    Cleans and validates an APE code (or NAF) and compare them.

    NB 1: If either or both values are invalid, returns False.
    NB 2: If both values are None, returns True.
    """
    if ape_code1 is None and ape_code2 is None:
        return True
    try:
        ape_code1 = ApeValidator.validate_ape_code(ape_code1)
        ape_code2 = ApeValidator.validate_ape_code(ape_code2)
    except ApeValidationError:
        return False
    return ape_code1 == ape_code2
validate_ape_code staticmethod 
validate_ape_code(raw_code)

Cleans and validates an APE code (or NAF) and returns it, else raises an error.

NB: None value is considered as valid.

Source code in components/fr/public/validators/ape_validator.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@staticmethod
def validate_ape_code(raw_code: str | None) -> str | None:
    """
    Cleans and validates an APE code (or NAF) and returns it, else raises an error.

    NB: None value is considered as valid.
    """
    if raw_code is None:
        return None

    raw_code = raw_code.replace(".", "").strip()

    if (
        len(raw_code) != 5
        or not raw_code[0:4].isnumeric()
        or not raw_code[-1].isalpha()
    ):
        raise ApeValidationError.wrong_format(raw_code)  # type: ignore[no-untyped-call]

    return raw_code