Skip to content

Api reference

components.occupational_health.public.dependencies

OCCUPATIONAL_HEALTH_COMPONENT_NAME module-attribute 

OCCUPATIONAL_HEALTH_COMPONENT_NAME = 'occupational_health'

OccupationalHealthDependency

Bases: ABC

Dependency interface for occupational_health component to access country-specific functionality.

This interface abstracts away direct dependencies on country-specific components, allowing the occupational_health component to work with different country implementations.

create_profile_with_user abstractmethod 

create_profile_with_user(profile_data)

Create user profile with user data.

Source code in components/occupational_health/public/dependencies.py 
125
126
127
128
@abstractmethod
def create_profile_with_user(self, profile_data: ProfileData) -> tuple[str, UUID]:
    """Create user profile with user data."""
    raise NotImplementedError()

get_account_admins_profile_ids abstractmethod 

get_account_admins_profile_ids(account_id)

Get account admin profile IDs.

Source code in components/occupational_health/public/dependencies.py 
137
138
139
140
@abstractmethod
def get_account_admins_profile_ids(self, account_id: UUID) -> set[UUID]:
    """Get account admin profile IDs."""
    raise NotImplementedError()

get_account_id abstractmethod 

get_account_id(company_id)

Get account ID for a company.

Source code in components/occupational_health/public/dependencies.py 
42
43
44
45
@abstractmethod
def get_account_id(self, company_id: str) -> UUID:
    """Get account ID for a company."""
    raise NotImplementedError()

get_account_id_and_siret abstractmethod 

get_account_id_and_siret(company_id)

Get account ID and SIRET for a company.

Source code in components/occupational_health/public/dependencies.py 
52
53
54
55
@abstractmethod
def get_account_id_and_siret(self, company_id: str) -> tuple[UUID, str | None]:
    """Get account ID and SIRET for a company."""
    raise NotImplementedError()

get_account_id_customers_2025 abstractmethod 

get_account_id_customers_2025()

Get account IDs of customers for the 2025 year.

Source code in components/occupational_health/public/dependencies.py 
147
148
149
150
@abstractmethod
def get_account_id_customers_2025(self) -> set[UUID]:
    """Get account IDs of customers for the 2025 year."""
    raise NotImplementedError()

get_account_name abstractmethod 

get_account_name(account_id)

Get account name.

Source code in components/occupational_health/public/dependencies.py 
81
82
83
84
@abstractmethod
def get_account_name(self, account_id: UUID) -> str:
    """Get account name."""
    raise NotImplementedError()

get_company_admins_profile_ids abstractmethod 

get_company_admins_profile_ids(company_id)

Get company admin profile IDs.

Source code in components/occupational_health/public/dependencies.py 
142
143
144
145
@abstractmethod
def get_company_admins_profile_ids(self, company_id: str) -> set[UUID]:
    """Get company admin profile IDs."""
    raise NotImplementedError()

get_company_display_name_and_account_id abstractmethod 

get_company_display_name_and_account_id(company_id)

Get company display name and associated account ID.

Source code in components/occupational_health/public/dependencies.py 
74
75
76
77
78
79
@abstractmethod
def get_company_display_name_and_account_id(
    self, company_id: str
) -> tuple[str, UUID]:
    """Get company display name and associated account ID."""
    raise NotImplementedError()

get_company_from_siret abstractmethod 

get_company_from_siret(siret)

Get company data from SIRET via DSN.

Source code in components/occupational_health/public/dependencies.py 
 97
 98
 99
100
@abstractmethod
def get_company_from_siret(self, siret: str) -> "DSNCompanyData | None":
    """Get company data from SIRET via DSN."""
    raise NotImplementedError()

get_company_ids_in_account abstractmethod 

get_company_ids_in_account(account_id)

Get all company IDs in an account.

Source code in components/occupational_health/public/dependencies.py 
47
48
49
50
@abstractmethod
def get_company_ids_in_account(self, account_id: UUID) -> list[str]:
    """Get all company IDs in an account."""
    raise NotImplementedError()

get_company_names_by_ids abstractmethod 

get_company_names_by_ids(company_ids)

Get company names by IDs.

Source code in components/occupational_health/public/dependencies.py 
69
70
71
72
@abstractmethod
def get_company_names_by_ids(self, company_ids: set[str]) -> dict[str, str]:
    """Get company names by IDs."""
    raise NotImplementedError()

get_company_siren abstractmethod 

get_company_siren(company_id)

Get company SIREN.

Source code in components/occupational_health/public/dependencies.py 
64
65
66
67
@abstractmethod
def get_company_siren(self, company_id: str) -> str:
    """Get company SIREN."""
    raise NotImplementedError()

get_company_siret abstractmethod 

get_company_siret(company_id, force_nic=None)

Get company SIRET, optionally forcing a specific NIC.

Source code in components/occupational_health/public/dependencies.py 
57
58
59
60
61
62
@abstractmethod
def get_company_siret(
    self, company_id: str, force_nic: str | None = None
) -> str | None:
    """Get company SIRET, optionally forcing a specific NIC."""
    raise NotImplementedError()

get_global_profile_id abstractmethod 

get_global_profile_id(user_id)

Get global profile ID for a user.

Source code in components/occupational_health/public/dependencies.py 
87
88
89
90
@abstractmethod
def get_global_profile_id(self, user_id: str) -> UUID | None:
    """Get global profile ID for a user."""
    raise NotImplementedError()

get_user_id_by_global_profile_mapping abstractmethod 

get_user_id_by_global_profile_mapping(global_profile_ids)

Get global profile ID -> user ID mapping from global health profile IDs.

Source code in components/occupational_health/public/dependencies.py 
102
103
104
105
106
107
@abstractmethod
def get_user_id_by_global_profile_mapping(
    self, global_profile_ids: Iterable[UUID]
) -> dict[UUID, str]:
    """Get global profile ID -> user ID mapping from global health profile IDs."""
    raise NotImplementedError()

get_user_id_from_global_profile_id abstractmethod 

get_user_id_from_global_profile_id(global_profile_id)

Get user ID from global profile ID.

Source code in components/occupational_health/public/dependencies.py 
92
93
94
95
@abstractmethod
def get_user_id_from_global_profile_id(self, global_profile_id: UUID) -> str:
    """Get user ID from global profile ID."""
    raise NotImplementedError()

get_user_id_mapping_from_user_ids abstractmethod 

get_user_id_mapping_from_user_ids(user_ids)

Get global profile ID -> user ID mapping from user IDs.

Source code in components/occupational_health/public/dependencies.py 
109
110
111
112
113
114
@abstractmethod
def get_user_id_mapping_from_user_ids(
    self, user_ids: Iterable[str]
) -> dict[UUID, str]:
    """Get global profile ID -> user ID mapping from user IDs."""
    raise NotImplementedError()

get_work_stoppages_for_users abstractmethod 

get_work_stoppages_for_users(
    user_ids, ever_active_during_period=None
)

Get work stoppages for users.

Source code in components/occupational_health/public/dependencies.py 
116
117
118
119
120
121
122
123
@abstractmethod
def get_work_stoppages_for_users(
    self,
    user_ids: Iterable[str],
    ever_active_during_period: tuple[date, date] | None = None,
) -> dict[str, list["WorkStoppage"]]:
    """Get work stoppages for users."""
    raise NotImplementedError()

set_ssn_ntt_on_user abstractmethod 

set_ssn_ntt_on_user(user_id, ssn, ntt)

Set SSN and NTT on user.

Source code in components/occupational_health/public/dependencies.py 
130
131
132
133
134
135
@abstractmethod
def set_ssn_ntt_on_user(
    self, user_id: str, ssn: str | None, ntt: str | None
) -> None:
    """Set SSN and NTT on user."""
    raise NotImplementedError()

ProfileData dataclass 

ProfileData(
    email=None,
    first_name=None,
    last_name=None,
    birth_date=None,
    language=None,
    phone_number=None,
    gender=None,
)

The profile data used to create a user and profile.

birth_date class-attribute instance-attribute 

birth_date = None

email class-attribute instance-attribute 

email = None

first_name class-attribute instance-attribute 

first_name = None

gender class-attribute instance-attribute 

gender = None

language class-attribute instance-attribute 

language = None

last_name class-attribute instance-attribute 

last_name = None

phone_number class-attribute instance-attribute 

phone_number = None

get_app_dependency 

get_app_dependency()

Retrieves the occupational_health dependency

Source code in components/occupational_health/public/dependencies.py 
162
163
164
165
166
167
168
169
170
def get_app_dependency() -> OccupationalHealthDependency:
    """Retrieves the occupational_health dependency"""
    from flask import current_app

    app = cast("CustomFlask", current_app)
    return cast(
        "OccupationalHealthDependency",
        app.get_component_dependency(OCCUPATIONAL_HEALTH_COMPONENT_NAME),
    )

set_app_dependency 

set_app_dependency(dependency)

Sets the occupational_health dependency to the app

Source code in components/occupational_health/public/dependencies.py 
153
154
155
156
157
158
159
def set_app_dependency(dependency: OccupationalHealthDependency) -> None:
    """Sets the occupational_health dependency to the app"""
    from flask import current_app

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

components.occupational_health.public.employment

employment_consumer

Note: Do not import local country code here, do it in the internal component after checking the country code.

occupational_health_employment_change_consumer 

occupational_health_employment_change_consumer(
    employment_change, event_bus_orchestrator
)
Source code in components/occupational_health/public/employment/employment_consumer.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
def occupational_health_employment_change_consumer(  # noqa: D103
    employment_change: "EmploymentChange[FrExtendedValues]",
    event_bus_orchestrator: "EventBusOrchestrator",
) -> None:
    try:
        from components.employment.public.enums import CountryCode

        if employment_change.country_code != CountryCode.fr:
            return

        from components.occupational_health.internal.business_logic.actions.employment.on_employment_change import (
            on_employment_change,
        )

        on_employment_change(
            employment_change=employment_change,
            event_bus_orchestrator=event_bus_orchestrator,
        )
    except Exception:
        # Catching this here to avoid the exception bubbling up and breaking the whole employment component
        current_logger.exception(
            "Failed to handle employment changes",
            extra={
                "employment_change": employment_change,
            },
        )

requires_siret_information

requires_siret_information 

requires_siret_information(account_id)

Return whether the given account ID requires SIRET information to be provided when tracking employment movements.

Source code in components/occupational_health/public/employment/requires_siret_information.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
def requires_siret_information(account_id: UUID) -> bool:
    """
    Return whether the given account ID requires SIRET information to be provided when tracking employment movements.
    """
    if not has_active_occupational_health_contract(account_id):
        # No need to collect the SIRET if there is no active contract
        return False

    rules = current_session.query(AffiliationStrategyRule).filter(  # noqa: ALN085
        AffiliationStrategyRule.account_id == account_id,
    )

    rules_count = rules.count()

    if rules_count == 0:
        # No rule = will be manual anyway (should not happen often, as it indicates a misconfiguration)
        return False

    if rules_count == 1:
        rule: AffiliationStrategyRule = rules.one()

        is_always_affiliate = (
            rule.action == RuleAction.AFFILIATE and rule.is_wildcard_siret
        )
        if is_always_affiliate:
            return False

        is_manual = (
            rule.action == RuleAction.REQUIRE_MANUAL_REVIEW and rule.is_wildcard_siret
        )
        if is_manual:
            return False

        # Misconfiguration: only one rule, SIRET-based. We'll probably need the SIRET down the line.
        return True

    # Multiple rules = SIRET is needed to make a decision
    return True

components.occupational_health.public.entities

billing

BillingStrategy

Bases: AlanBaseEnum

If the customer is getting an invoice per SIREN or per SIRET.

SIREN_BASED class-attribute instance-attribute 
SIREN_BASED = 'siren_based'
SIRET_BASED class-attribute instance-attribute 
SIRET_BASED = 'siret_based'

CustomerAddress dataclass 

CustomerAddress(street, postal_code, city)

Postal address of a customer.

city instance-attribute 
city
postal_code instance-attribute 
postal_code
street instance-attribute 
street

CustomerData dataclass 

CustomerData(
    *, entity_name, siren, siret=None, postal_address
)

Customer data needed for billing purposes.

billing_strategy property 
billing_strategy

Return the billing strategy for this customer.

entity_name instance-attribute 
entity_name
postal_address instance-attribute 
postal_address
siren instance-attribute 
siren
siret class-attribute instance-attribute 
siret = None

InvoiceToGenerateData dataclass 

InvoiceToGenerateData(
    *,
    billing_period,
    billing_year,
    contract_type,
    contract_ref,
    contract_yearly_price_per_employee_in_cents,
    contract_needs_installment_plan_for_employees_count,
    references_of_employees_affiliated_over_the_year
)

Data needed to generate an invoice for a customer.

billing_period instance-attribute 
billing_period
billing_year instance-attribute 
billing_year

The year of the billing period. END_OF_YEAR_REGUL invoices are issued on 01/01 of the year Y+1 but will hold the original year Y in billing_year.

contract_needs_installment_plan property 
contract_needs_installment_plan

Whether the contract needs an installment plan.

contract_needs_installment_plan_for_employees_count instance-attribute 
contract_needs_installment_plan_for_employees_count
contract_ref instance-attribute 
contract_ref
contract_type instance-attribute 
contract_type
contract_yearly_price_per_employee_in_cents instance-attribute 
contract_yearly_price_per_employee_in_cents
references_of_employees_affiliated_over_the_year instance-attribute 
references_of_employees_affiliated_over_the_year

List of opaque references of all employees affiliated on this subscription/contract, during the entire year of the billing period (not just the specified billing period).

OccupationalHealthBillingPeriod

Bases: AlanBaseEnum

Billing period for occupational health services.

END_OF_YEAR_REGUL class-attribute instance-attribute 
END_OF_YEAR_REGUL = 'END_OF_YEAR_REGUL'
Q1 class-attribute instance-attribute 
Q1 = 'Q1'
Q2 class-attribute instance-attribute 
Q2 = 'Q2'
Q3 class-attribute instance-attribute 
Q3 = 'Q3'
Q4 class-attribute instance-attribute 
Q4 = 'Q4'

dmst

CurriculumLaboris dataclass 

CurriculumLaboris(*, jobs)

Bases: DataClassJsonMixin

Member curriculum laboris (work information): previous job, description of work, tasks, etc.

Keep in sync with: frontend/apps/medical-software/app/hooks/useProfileDmstQuery.ts

Used in the HP medical app (DMST) for the "Professional" tab

jobs instance-attribute 
jobs

Dmst dataclass 

Dmst(
    *,
    occupational_health_profile_id,
    first_name,
    last_name,
    gender,
    birthdate,
    curriculum_laboris
)

Bases: DataClassJsonMixin

Information about a member, such as their name, gender, birthdate, etc.

Keep in sync with: frontend/apps/medical-software/app/hooks/useProfileDmstQuery.ts

Used in the HP medical app (DMST)

birthdate instance-attribute 
birthdate
curriculum_laboris instance-attribute 
curriculum_laboris
first_name instance-attribute 
first_name
gender instance-attribute 
gender
last_name instance-attribute 
last_name
occupational_health_profile_id instance-attribute 
occupational_health_profile_id

MemberJob dataclass 

MemberJob(
    *,
    id,
    title,
    start_date,
    end_date,
    employer,
    description
)

Bases: DataClassJsonMixin

Represents a single job held by a member, including employment details and duration.

Keep in sync with: frontend/apps/medical-software/app/hooks/useProfileDmstQuery.ts

Used in the HP medical app (DMST) for the "Professional" tab

description instance-attribute 
description
employer instance-attribute 
employer
end_date instance-attribute 
end_date
id instance-attribute 
id
start_date instance-attribute 
start_date
title instance-attribute 
title

member_info

MemberInfo dataclass 

MemberInfo(
    *,
    occupational_health_profile_id,
    user_id,
    member_first_name,
    member_last_name,
    member_gender,
    member_birthdate
)

Bases: DataClassJsonMixin

Information about a member, such as their name, gender, birthdate, etc.

Note: we'll be using other structures for the rest of the data e.g. MemberActivity, etc.

Used in the HP medical app.

member_birthdate instance-attribute 
member_birthdate
member_first_name instance-attribute 
member_first_name
member_gender instance-attribute 
member_gender
member_last_name instance-attribute 
member_last_name
occupational_health_profile_id instance-attribute 
occupational_health_profile_id
user_id instance-attribute 
user_id

visit

VisitInfo dataclass 

VisitInfo(
    *,
    member_first_name,
    member_last_name,
    member_gender,
    member_birthdate,
    visit_date,
    visit_type,
    health_professional_name,
    health_professional_id,
    visit_status
)

Bases: DataClassJsonMixin

Information about a visit scheduled for a specific date.

Used in the HP medical app.

health_professional_id instance-attribute 
health_professional_id
health_professional_name instance-attribute 
health_professional_name
member_birthdate instance-attribute 
member_birthdate
member_first_name instance-attribute 
member_first_name
member_gender instance-attribute 
member_gender
member_last_name instance-attribute 
member_last_name
visit_date instance-attribute 
visit_date
visit_status instance-attribute 
visit_status
visit_type instance-attribute 
visit_type

components.occupational_health.public.enums

AffiliationDecision

Bases: AlanBaseEnum

Possible decisions for occupational health affiliation.

AFFILIATED class-attribute instance-attribute 

AFFILIATED = 'affiliated'

NOT_AFFILIATED class-attribute instance-attribute 

NOT_AFFILIATED = 'not_affiliated'

REQUIRE_MANUAL_REVIEW class-attribute instance-attribute 

REQUIRE_MANUAL_REVIEW = 'require_manual_review'

components.occupational_health.public.events

subscription

subscribe_to_events 

subscribe_to_events()

All event subscriptions for the occupational health component should be done here.

Source code in components/occupational_health/public/events/subscription.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def subscribe_to_events() -> None:
    """
    All event subscriptions for the occupational health component should be done here.
    """
    from shared.messaging.broker import get_message_broker

    message_broker = get_message_broker()

    # This is on top of the existing global profile profile merge event
    # See: the components.global_profile.internal.events.subscribers.update_links_to_profile_when_merging
    message_broker.subscribe_async(
        ProfilesMerged,
        update_occupational_health_profile_merge,
        queue_name=LOW_PRIORITY_QUEUE,
    )

components.occupational_health.public.marmot

actions

Actions for use in Marmot controllers.

affiliate_new_member_for_marmot 

affiliate_new_member_for_marmot(
    *,
    user_id,
    account_id,
    start_date,
    end_date=None,
    actor_user_id=None
)

Affiliate a new member to occupational health.

Source code in components/occupational_health/public/marmot/actions.py 
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
def affiliate_new_member_for_marmot(
    *,
    user_id: str,
    account_id: UUID,
    start_date: date,
    end_date: date | None = None,
    actor_user_id: str | None = None,
) -> None:
    """
    Affiliate a new member to occupational health.
    """
    current_logger.info(
        "Affiliating new member from Marmot",
        extra={
            "user_id": user_id,
            "account_id": account_id,
            "start_date": start_date,
            "end_date": end_date,
        },
    )

    profile_id = get_or_create_profile_id(user_id)

    current_logger.info(
        "Affiliating new member from Marmot",
        extra={
            "profile_id": profile_id,
            "account_id": account_id,
            "start_date": start_date,
            "end_date": end_date,
        },
    )

    account_name = get_account_name(account_id)
    ts_thread = log_and_post_message(
        f"Affiliating new member {user_id} from Marmot (account {account_name})",
        username="Alaner activity in Marmot",
    )
    slack_id = get_actor_slack_handle(actor_user_id=actor_user_id)
    log_and_post_message(
        f"<{slack_id}> can you share context on why this was required?",
        username="Alaner activity in Marmot",
        ts_thread=ts_thread,
    )

    affiliate_member(
        occupational_health_profile_id=profile_id,
        start_date=start_date,
        end_date=end_date,
        account_id=account_id,
    )

cancel_affiliation_for_marmot 

cancel_affiliation_for_marmot(
    affiliation_id, actor_user_id
)

Cancel an affiliation for a profile.

This is a wrapper around cancel_affiliation that adds logging.

Source code in components/occupational_health/public/marmot/actions.py 
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
def cancel_affiliation_for_marmot(
    affiliation_id: UUID,
    actor_user_id: str | None,
) -> None:
    """
    Cancel an affiliation for a profile.

    This is a wrapper around cancel_affiliation that adds logging.
    """
    from components.occupational_health.internal.business_logic.actions.affiliation import (
        cancel_affiliation,
    )

    current_logger.info(
        f"Cancelling affiliation {affiliation_id} from Marmot",
        extra={
            "actor_user_id": actor_user_id,
        },
    )

    ts_thread = log_and_post_message(
        f"Cancelling affiliation `{affiliation_id}` from Marmot",
        username="Alaner activity in Marmot",
    )
    slack_id = get_actor_slack_handle(actor_user_id=actor_user_id)
    log_and_post_message(
        f"<{slack_id}> can you share context on why this was required?",
        username="Alaner activity in Marmot",
        ts_thread=ts_thread,
    )

    cancel_affiliation(affiliation_id=affiliation_id)

change_affiliation_decision 

change_affiliation_decision(
    affiliation_decision_id, manual_decision
)

Change the decision of an affiliation decision. If the decision is to affiliate, the profile will be affiliated.

Source code in components/occupational_health/public/marmot/actions.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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def change_affiliation_decision(
    affiliation_decision_id: UUID, manual_decision: AffiliationDecision
) -> None:
    """
    Change the decision of an affiliation decision.
    If the decision is to affiliate, the profile will be affiliated.
    """
    from components.occupational_health.external.company import get_account_id
    from components.occupational_health.internal.business_logic.actions.affiliation import (
        affiliate_member,
    )

    if manual_decision == AffiliationDecision.REQUIRE_MANUAL_REVIEW:
        raise BaseErrorCode.invalid_arguments(
            message="manual_decision cannot be require_manual_review"
        )

    decision = get_or_raise_missing_resource(
        OccupationalHealthAffiliationDecision, affiliation_decision_id
    )

    if manual_decision == AffiliationDecision.AFFILIATED:
        # Extract company_id from the employment change payload to get the account_id
        company_id = decision.employment_change_payload["core_employment_version"][
            "company_id"
        ]
        account_id = get_account_id(company_id)

        # Get start_date and end_date from the employment change payload
        employment_version = decision.employment_change_payload[
            "core_employment_version"
        ]
        start_date = employment_version["start_date"]
        end_date = employment_version.get("end_date")  # Optional

        affiliate_member(
            occupational_health_profile_id=mandatory(decision.profile_id),
            start_date=start_date,
            end_date=end_date,
            account_id=account_id,
            commit=False,  # We'll commit at the end of this function
        )

    decision.manual_decision = manual_decision
    current_session.add(decision)

    current_session.commit()

reprocess_affiliation_decision_for_marmot 

reprocess_affiliation_decision_for_marmot(
    affiliation_decision_id,
)

Reprocess an affiliation decision for Marmot. This will trigger a new affiliation decision based on the same employment change payload.

Source code in components/occupational_health/public/marmot/actions.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
def reprocess_affiliation_decision_for_marmot(affiliation_decision_id: UUID) -> None:
    """
    Reprocess an affiliation decision for Marmot.
    This will trigger a new affiliation decision based on the same employment change payload.
    """
    from components.occupational_health.internal.business_logic.actions.employment.reprocess_employee_movement import (
        reprocess_employee_movement as _reprocess_employee_movement,
    )

    _reprocess_employee_movement(
        affiliation_decision_id=affiliation_decision_id,
        replace_previous_decision=True,
    )

terminate_affiliation_for_marmot 

terminate_affiliation_for_marmot(
    account_id,
    profile_id,
    end_date,
    termination_type,
    actor_user_id,
)

Terminate an affiliation for a profile. This is a wrapper around terminate_member_affiliation that adds logging.

Source code in components/occupational_health/public/marmot/actions.py 
 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
def terminate_affiliation_for_marmot(
    account_id: UUID,
    profile_id: UUID,
    end_date: date,
    termination_type: str,
    actor_user_id: str | None,
) -> None:
    """
    Terminate an affiliation for a profile.
    This is a wrapper around terminate_member_affiliation that adds logging.
    """
    from components.occupational_health.internal.business_logic.actions.affiliation import (
        terminate_member_affiliation,
    )

    current_logger.info(
        "Terminating affiliation for profile from Marmot",
        extra={
            "account_id": account_id,
            "profile_id": profile_id,
            "end_date": end_date,
            "termination_type": termination_type,
            "actor_user_id": actor_user_id,
        },
    )

    account_name = get_account_name(account_id)

    ts_thread = log_and_post_message(
        f"Terminating affiliation for profile `{profile_id}` from Marmot (account <https://alan.com/marmot-v2/occupational-health/affiliations?account_id={account_id}|{account_name}>) on date {end_date}",
        username="Alaner activity in Marmot",
    )
    slack_id = get_actor_slack_handle(actor_user_id=actor_user_id)
    log_and_post_message(
        f"<{slack_id}> can you share context on why this was required?",
        username="Alaner activity in Marmot",
        ts_thread=ts_thread,
    )

    terminate_member_affiliation(
        occupational_health_profile_id=profile_id,
        end_date=end_date,
        account_id=account_id,
        termination_type=termination_type,
    )

entities

AffiliatedMemberEmploymentDataForMarmot dataclass 

AffiliatedMemberEmploymentDataForMarmot(
    start_date, end_date, company_id, company_name
)

Bases: DataClassJsonMixin

Represent the employment of a member affiliated, to be displayed in Marmot

company_id instance-attribute 
company_id
company_name instance-attribute 
company_name
end_date instance-attribute 
end_date
start_date instance-attribute 
start_date

AffiliatedMemberForMarmot dataclass 

AffiliatedMemberForMarmot(
    affiliation_id,
    start_date,
    end_date,
    is_cancelled,
    user_id,
    occupational_health_profile_id,
    global_profile_id,
    first_name,
    last_name,
    birth_date,
    email,
    phone_number,
    is_vip,
    employments,
    account_id,
)

Bases: DataClassJsonMixin

Represent a member affiliated to Prévenir, to be displayed in Marmot

account_id instance-attribute 
account_id
affiliation_id instance-attribute 
affiliation_id
birth_date instance-attribute 
birth_date
email instance-attribute 
email
employments instance-attribute 
employments
end_date instance-attribute 
end_date
first_name instance-attribute 
first_name
global_profile_id instance-attribute 
global_profile_id
is_cancelled instance-attribute 
is_cancelled
is_vip instance-attribute 
is_vip
last_name instance-attribute 
last_name
occupational_health_profile_id instance-attribute 
occupational_health_profile_id
phone_number instance-attribute 
phone_number
start_date instance-attribute 
start_date
user_id instance-attribute 
user_id

AffiliationMovementForMarmot dataclass 

AffiliationMovementForMarmot(
    movement_id,
    account_id,
    profile_id,
    movement_type,
    date_of_movement,
    first_name,
    last_name,
    birth_date,
    created_at,
    decision_id,
    company_id,
    company_name,
    company_name_enriched_from_dsn,
)

Bases: DataClassJsonMixin

Represent an affiliation movement for display in Marmot

account_id instance-attribute 
account_id
birth_date instance-attribute 
birth_date
company_id instance-attribute 
company_id
company_name instance-attribute 
company_name
company_name_enriched_from_dsn instance-attribute 
company_name_enriched_from_dsn
created_at instance-attribute 
created_at
date_of_movement instance-attribute 
date_of_movement
decision_id instance-attribute 
decision_id
first_name instance-attribute 
first_name
last_name instance-attribute 
last_name
movement_id instance-attribute 
movement_id
movement_type instance-attribute 
movement_type
profile_id instance-attribute 
profile_id

OccupationalHealthAffiliationDecisionForMarmot dataclass 

OccupationalHealthAffiliationDecisionForMarmot(
    id,
    created_at,
    decision,
    employment_change_payload,
    core_employment_id,
    rule_id,
    rule_name,
)

Bases: DataClassJsonMixin

Affiliation decision data for display in Marmot.

core_employment_id instance-attribute 
core_employment_id
created_at instance-attribute 
created_at
decision instance-attribute 
decision
employment_change_payload instance-attribute 
employment_change_payload
id instance-attribute 
id
rule_id instance-attribute 
rule_id
rule_name instance-attribute 
rule_name

OccupationalHealthAffiliationDecisionWithProfileForMarmot dataclass 

OccupationalHealthAffiliationDecisionWithProfileForMarmot(
    id,
    created_at,
    decision,
    manual_decision,
    employment_change_payload,
    core_employment_id,
    rule_id,
    rule_name,
    first_name,
    last_name,
    birth_date,
    company_id,
    company_name,
    company_name_enriched_from_dsn,
)

Bases: DataClassJsonMixin

Affiliation decision data with profile information for display in Marmot.

birth_date instance-attribute 
birth_date
company_id instance-attribute 
company_id
company_name instance-attribute 
company_name
company_name_enriched_from_dsn instance-attribute 
company_name_enriched_from_dsn
core_employment_id instance-attribute 
core_employment_id
created_at instance-attribute 
created_at
decision instance-attribute 
decision
employment_change_payload instance-attribute 
employment_change_payload
first_name instance-attribute 
first_name
id instance-attribute 
id
last_name instance-attribute 
last_name
manual_decision instance-attribute 
manual_decision
rule_id instance-attribute 
rule_id
rule_name instance-attribute 
rule_name

OccupationalHealthProfileData dataclass 

OccupationalHealthProfileData(
    profile_id,
    visits,
    affiliations,
    affiliation_decisions,
    next_visit_result,
)

Bases: DataClassJsonMixin

All occupational health data for a profile

affiliation_decisions instance-attribute 
affiliation_decisions
affiliations instance-attribute 
affiliations
next_visit_result instance-attribute 
next_visit_result
profile_id instance-attribute 
profile_id
visits instance-attribute 
visits

OccupationalHealthVisitForMarmot dataclass 

OccupationalHealthVisitForMarmot(
    id,
    visit_date,
    visit_type,
    visit_setup,
    managed_by_prevenir,
    took_place,
    cancellation_reason,
    work_stoppage_reason,
    health_professional_first_name,
    health_professional_last_name,
)

Bases: DataClassJsonMixin

Represent a visit for display in Marmot

cancellation_reason instance-attribute 
cancellation_reason
health_professional_first_name instance-attribute 
health_professional_first_name
health_professional_last_name instance-attribute 
health_professional_last_name
id instance-attribute 
id
managed_by_prevenir instance-attribute 
managed_by_prevenir
took_place instance-attribute 
took_place
visit_date instance-attribute 
visit_date
visit_setup instance-attribute 
visit_setup
visit_type instance-attribute 
visit_type
work_stoppage_reason instance-attribute 
work_stoppage_reason

enums

MovementType

Bases: AlanBaseEnum

The recorded movement: either an affiliation, or a termination, or a cancellation.

Transfers are composed by a termination and an affiliation, if they moved between two distinct accounts.

AFFILIATION class-attribute instance-attribute 
AFFILIATION = 'affiliation'
CANCELLATION class-attribute instance-attribute 
CANCELLATION = 'cancellation'
TERMINATION class-attribute instance-attribute 
TERMINATION = 'termination'

queries

Queries for use in Marmot controllers.

get_affiliated_members_for_marmot 

get_affiliated_members_for_marmot(
    filter_by_account_id=None,
)

Return the list of affiliated members (including past ones) for display in Marmot.

Source code in components/occupational_health/public/marmot/queries.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
@tracer_wrap()
def get_affiliated_members_for_marmot(
    filter_by_account_id: UUID | None = None,
) -> list[AffiliatedMemberForMarmot]:
    """
    Return the list of affiliated members (including past ones) for display in Marmot.
    """
    from components.affiliation_occ_health.public.api import (
        get_affiliation_list,
    )
    from components.occupational_health.internal.business_logic.queries.marmot import (
        get_affiliated_members_from_affiliations_for_marmot,
    )
    from components.occupational_health.internal.helpers.affiliation import (
        build_subscription_ref,
    )

    if filter_by_account_id:
        affiliations = get_affiliation_list(
            AffiliationService.OCCUPATIONAL_HEALTH,
            subscription_ref=build_subscription_ref(filter_by_account_id),
        )
    else:
        affiliations = get_affiliation_list(AffiliationService.OCCUPATIONAL_HEALTH)

    return get_affiliated_members_from_affiliations_for_marmot(
        affiliations=affiliations
    )

get_affiliation_decisions_for_marmot 

get_affiliation_decisions_for_marmot(
    profile_service,
    account_id,
    decision=None,
    not_older_than=None,
)

Return the list of affiliation decisions, optionally filtered by decision type and date.

Source code in components/occupational_health/public/marmot/queries.py 
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
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
375
376
@tracer_wrap()
@inject_profile_service
def get_affiliation_decisions_for_marmot(
    profile_service: ProfileService,
    account_id: UUID,
    decision: AffiliationDecision | None = None,
    not_older_than: date | None = None,
) -> list[OccupationalHealthAffiliationDecisionWithProfileForMarmot]:
    """
    Return the list of affiliation decisions, optionally filtered by decision type and date.
    """
    company_ids = get_company_ids_in_account(account_id)
    query = (
        current_session.query(OccupationalHealthAffiliationDecision)  # noqa: ALN085
        .join(OccupationalHealthAffiliationDecision.rule, isouter=True)
        .join(OccupationalHealthAffiliationDecision.profile)
        .options(
            joinedload(OccupationalHealthAffiliationDecision.rule),
            joinedload(OccupationalHealthAffiliationDecision.profile),
            raiseload("*"),
        )
        .filter(
            cast(
                OccupationalHealthAffiliationDecision.employment_change_payload[
                    "core_employment_version"
                ]["company_id"].astext,
                Integer,
            ).in_(company_ids),
        )
        .order_by(OccupationalHealthAffiliationDecision.created_at.desc())
    )

    if decision:
        query = query.filter(OccupationalHealthAffiliationDecision.decision == decision)

    if not_older_than:
        query = query.filter(
            OccupationalHealthAffiliationDecision.created_at >= not_older_than
        )

    decisions = query.all()

    # Get all global profile IDs
    global_profile_ids = {decision.profile.global_profile_id for decision in decisions}  # type: ignore[union-attr]
    profiles = profile_service.get_profiles(profile_ids=global_profile_ids)
    profiles_per_global_id = {profile.id: profile for profile in profiles}

    # Get company names
    company_names = get_company_names_by_ids(set(company_ids))
    company_names_from_company_id_and_nic = _build_company_names_by_company_id_and_nic(
        [decision.employment_change_payload for decision in decisions]
    )

    return [
        _build_decision(
            decision,
            profiles_per_global_id.get(decision.profile.global_profile_id),  # type: ignore[union-attr]
            company_names,
            company_names_from_company_id_and_nic,
        )
        for decision in decisions
    ]

get_occupational_health_profile_data_for_marmot 

get_occupational_health_profile_data_for_marmot(profile_id)

Return all occupational health data for a profile, for display in Marmot.

Source code in components/occupational_health/public/marmot/queries.py 
 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
@tracer_wrap()
def get_occupational_health_profile_data_for_marmot(
    profile_id: UUID,
) -> OccupationalHealthProfileData:
    """
    Return all occupational health data for a profile, for display in Marmot.
    """
    from components.affiliation_occ_health.public.api import get_affiliation_list
    from components.occupational_health.internal.business_logic.queries.marmot import (
        get_affiliated_members_from_affiliations_for_marmot,
    )
    from components.occupational_health.internal.business_logic.queries.visits.visits import (
        get_predictable_visits,
    )
    from components.occupational_health.internal.helpers.affiliation import (
        build_affiliable_ref,
    )
    from components.occupational_health.internal.models.occupational_health_affiliation_decision import (
        OccupationalHealthAffiliationDecision,
    )
    from components.occupational_health.internal.models.occupational_health_visit import (
        OccupationalHealthVisit,
    )
    from shared.helpers.db import current_session

    # Get all affiliations for this profile
    affiliable_ref = build_affiliable_ref(profile_id)
    affiliations = get_affiliation_list(
        AffiliationService.OCCUPATIONAL_HEALTH,
        affiliable_ref=affiliable_ref,
    )
    affiliated_members = get_affiliated_members_from_affiliations_for_marmot(
        affiliations=affiliations
    )

    # Get all visits
    visits = (
        current_session.query(OccupationalHealthVisit)  # noqa: ALN085
        .filter(OccupationalHealthVisit.profile_id == profile_id)
        .options(
            joinedload(OccupationalHealthVisit.health_professional),
            raiseload("*"),
        )
    )

    # Get all affiliation decisions
    affiliation_decisions = (
        current_session.query(OccupationalHealthAffiliationDecision)  # noqa: ALN085
        .filter(OccupationalHealthAffiliationDecision.profile_id == profile_id)
        .options(
            joinedload(OccupationalHealthAffiliationDecision.rule),
            raiseload("*"),
        )
    )

    # Serialize visits
    serialized_visits = []
    for visit in visits:
        serialized_visits.append(
            OccupationalHealthVisitForMarmot(
                id=visit.id,
                visit_date=visit.visit_date,
                visit_type=visit.visit_type,
                visit_setup=visit.visit_setup,
                managed_by_prevenir=visit.managed_by_prevenir,
                took_place=visit.took_place,
                cancellation_reason=visit.cancellation_reason,
                work_stoppage_reason=visit.work_stoppage_reason,
                health_professional_first_name=visit.health_professional.first_name
                if visit.health_professional
                else None,
                health_professional_last_name=visit.health_professional.last_name
                if visit.health_professional
                else None,
            )
        )

    # Serialize affiliation decisions
    serialized_decisions = []
    for decision in affiliation_decisions:
        serialized_decisions.append(
            OccupationalHealthAffiliationDecisionForMarmot(
                id=decision.id,
                created_at=decision.created_at.date(),
                decision=decision.decision,
                employment_change_payload=decision.employment_change_payload,
                core_employment_id=decision.core_employment_id,
                rule_id=decision.rule_id,
                rule_name=decision.rule.name if decision.rule else None,
            )
        )

    predictable_visits_result = get_predictable_visits(
        profile_id=profile_id,
        affiliations=affiliations,
    )

    return OccupationalHealthProfileData(
        profile_id=profile_id,
        visits=serialized_visits,
        affiliations=affiliated_members,
        affiliation_decisions=serialized_decisions,
        next_visit_result=predictable_visits_result,
    )

get_recent_affiliation_movements 

get_recent_affiliation_movements(
    profile_service, account_id, not_older_than
)

Return the list of affiliation decisions made since the given date for the given account.

Source code in components/occupational_health/public/marmot/queries.py 
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
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
@inject_profile_service
def get_recent_affiliation_movements(
    profile_service: ProfileService,
    account_id: UUID,
    not_older_than: date,
) -> list[AffiliationMovementForMarmot]:
    """
    Return the list of affiliation decisions made since the given date for the given account.
    """
    movements = (
        current_session.query(  # noqa: ALN085
            OccupationalHealthProfile.global_profile_id,
            OccupationalHealthAffiliationMovement.id,
            OccupationalHealthAffiliationMovement.account_id,
            OccupationalHealthAffiliationMovement.created_at,
            OccupationalHealthAffiliationMovement.movement_type,
            OccupationalHealthAffiliationMovement.affiliation_decision_id,
            OccupationalHealthAffiliationMovement.profile_id,
            OccupationalHealthAffiliationMovement.employment_change_payload,
        )
        .select_from(OccupationalHealthAffiliationMovement)
        .join(OccupationalHealthAffiliationMovement.profile)
        .filter(
            OccupationalHealthAffiliationMovement.account_id == account_id,
            OccupationalHealthAffiliationMovement.created_at >= not_older_than,
        )
        .options(raiseload("*"))
        .all()
    )

    global_profile_ids = {movement.global_profile_id for movement in movements}
    profiles = profile_service.get_profiles(
        profile_ids=global_profile_ids,
    )
    profiles_per_global_id = {profile.id: profile for profile in profiles}

    company_ids = {
        movement.employment_change_payload["core_employment_version"]["company_id"]
        for movement in movements
        if movement.employment_change_payload
    }
    company_names = get_company_names_by_ids(company_ids)

    company_names_from_company_id_and_nic = _build_company_names_by_company_id_and_nic(
        [movement.employment_change_payload for movement in movements]
    )

    affiliation_movements_to_return: list[AffiliationMovementForMarmot] = []

    for movement in movements:
        if not movement.employment_change_payload:
            movement_company_id = None
            movement_company_name = None
            company_name_enriched_from_dsn = False
        else:
            movement_company_id = movement.employment_change_payload[
                "core_employment_version"
            ]["company_id"]
            nic = extract_nic_from_employment_change(
                EmploymentChange.from_dict(movement.employment_change_payload)
            )
            company_name_from_dsn = (
                company_names_from_company_id_and_nic.get((movement_company_id, nic))
                if nic
                else None
            )
            movement_company_name = (
                company_name_from_dsn
                # Fall back on the usual Company.name data
                or company_names[movement_company_id]
            )
            company_name_enriched_from_dsn = company_name_from_dsn is not None

        affiliation_movements_to_return.append(
            AffiliationMovementForMarmot(
                movement_id=movement.id,
                account_id=account_id,
                birth_date=profiles_per_global_id[
                    movement.global_profile_id
                ].birth_date,
                first_name=profiles_per_global_id[movement.global_profile_id].first_name
                or "",
                last_name=profiles_per_global_id[movement.global_profile_id].last_name
                or "",
                created_at=movement.created_at,
                profile_id=movement.profile_id,
                movement_type=movement.movement_type,
                date_of_movement=_get_date_of_movement(movement),
                decision_id=movement.affiliation_decision_id,
                company_id=movement_company_id,
                company_name=movement_company_name,
                company_name_enriched_from_dsn=company_name_enriched_from_dsn,
            )
        )

    return affiliation_movements_to_return

components.occupational_health.public.queries

billing

Set of queries used by the Billing stack.

See https://github.com/alan-eu/Topics/discussions/30344?sort=old#discussioncomment-14080373 ⧉

ContractRefNotFound

Bases: Exception

A contract reference was not found.

UnableToDetermineCustomerToBill

Bases: Exception

Unable to determine which customer to bill.

build_invoices_data_for_period 

build_invoices_data_for_period(
    billing_period, billing_year
)

Build the data required to generate all invoices for the given billing period and invoice year.

Parameters:

Name Type Description Default
billing_period OccupationalHealthBillingPeriod

The billing period to build data for.

 required
billing_year int

The year of the billing period to build data for.

 required

The data is not guaranteed to be stable over time for past billing periods, as retroactive affiliation changes can occur.

Returns:

Type Description
list[InvoiceToGenerateData]

list[InvoiceToGenerateData]: The data about all invoices to generate for the given period.
Source code in components/occupational_health/public/queries/billing.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
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def build_invoices_data_for_period(
    billing_period: OccupationalHealthBillingPeriod,
    billing_year: int,
) -> list[InvoiceToGenerateData]:
    """
    Build the data required to generate all invoices for the given billing period and invoice year.

    Args:
        billing_period: The billing period to build data for.
        billing_year: The year of the billing period to build data for.

    The data is not guaranteed to be stable over time for past billing periods, as
    retroactive affiliation changes can occur.

    Returns:
        list[InvoiceToGenerateData]: The data about all invoices to generate for the given period.
    """
    if billing_year != 2025:
        raise NotImplementedError("This function is only implemented for the year 2025")

    with read_only_session():
        # Get all customers with an active contract during the given billing period
        account_ids = get_app_dependency().get_account_id_customers_2025()

        invoice_to_generate_data_list: list[InvoiceToGenerateData] = []

        for account_id in account_ids:
            invoice_to_generate_data_list.extend(
                build_invoices_data_for_period_and_account(
                    billing_period=billing_period,
                    billing_year=billing_year,
                    account_id=account_id,
                )
            )

        return invoice_to_generate_data_list

build_invoices_data_for_period_and_account 

build_invoices_data_for_period_and_account(
    billing_period, billing_year, account_id
)

Build the data required to generate all invoices for the given billing period, invoice year and account.

Parameters:

Name Type Description Default
billing_period OccupationalHealthBillingPeriod

The billing period to build data for.

 required
billing_year int

The year of the billing period to build data for.

 required
account_id UUID

The account ID to build invoices for.

 required

The data is not guaranteed to be stable over time for past billing periods, as retroactive affiliation changes can occur.

Returns:

Type Description
list[InvoiceToGenerateData]

list[InvoiceToGenerateData]: The data about all invoices to generate for the given period.
Source code in components/occupational_health/public/queries/billing.py 
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
231
def build_invoices_data_for_period_and_account(
    billing_period: OccupationalHealthBillingPeriod,
    billing_year: int,
    account_id: UUID,
) -> list[InvoiceToGenerateData]:
    """
    Build the data required to generate all invoices for the given billing period, invoice year and account.

    Args:
        billing_period: The billing period to build data for.
        billing_year: The year of the billing period to build data for.
        account_id: The account ID to build invoices for.

    The data is not guaranteed to be stable over time for past billing periods, as
    retroactive affiliation changes can occur.

    Returns:
        list[InvoiceToGenerateData]: The data about all invoices to generate for the given period.
    """
    from components.occupational_health.internal.business_logic.queries.affiliation import (
        get_all_affiliations_for_account,
    )
    from components.occupational_health.internal.business_logic.queries.billing import (
        CUSTOMER_DATA_LIST,
        YEARLY_PRICE_PER_EMPLOYEE_IN_CENTS,
    )
    from components.occupational_health.internal.helpers.affiliation import (
        build_profile_id_from_affiliable_ref,
    )

    # 1. Get the customer data, including the billable entities.
    customers = []
    for customer in CUSTOMER_DATA_LIST:
        if (
            customer.account_id == account_id
            # Only consider "contracts" active during the billing year
            and customer.contract_start_date.year <= billing_year
            and (
                customer.contract_end_date is None
                # (we'll bill for contracts even if they end before the end of the year)
                or customer.contract_end_date.year >= billing_year
            )
        ):
            customers.append(customer)

    # 2. Get all affiliated members overlapping with billing period for the given account, spread them over the billable entities depending on their SIRET.
    affiliations_for_whole_account = get_all_affiliations_for_account(
        account_id=account_id
    )

    # Compute the start and end dates, between which we'll consider affiliations
    match billing_period:
        case OccupationalHealthBillingPeriod.Q1:
            start_date = date(billing_year, 1, 1)
            end_date = date(billing_year, 3, 31)
        case OccupationalHealthBillingPeriod.Q2:
            # Yes, we consider all affiliations from the start of the year, as the Q1 one might have changed in-between.
            start_date = date(billing_year, 1, 1)
            end_date = date(billing_year, 6, 30)
        case OccupationalHealthBillingPeriod.Q3:
            start_date = date(billing_year, 1, 1)
            end_date = date(billing_year, 9, 30)
        case OccupationalHealthBillingPeriod.Q4:
            start_date = date(billing_year, 1, 1)
            end_date = date(billing_year, 12, 31)

    customer_by_contract_ref = {
        customer.contract_ref: customer for customer in customers
    }
    affiliable_refs_per_customer_ref: dict[str, list[str]] = {
        c.contract_ref: [] for c in customers
    }
    errors = 0
    for affiliation in affiliations_for_whole_account:
        if affiliation.is_ever_active_between(
            period_start=start_date,
            period_end=end_date,
        ):
            # This affiliation is active at some point during the billing period:
            # take it into account for billing.

            try:
                # Find the right customer for this affiliation
                matching_customer = _find_customer_to_bill_for_affiliation(
                    account_id=account_id,
                    affiliation=affiliation,
                    customers=customers,
                )
            except UnableToDetermineCustomerToBill:
                current_logger.exception(
                    f"Unable to determine customer to bill for {affiliation}"
                )
                errors += 1
                # Log and skip for now...
            else:
                affiliable_refs_per_customer_ref[matching_customer.contract_ref].append(
                    affiliation.affiliable_ref
                )

    # TODO: ensure transfers are properly handled (eg unit test?)

    if errors:
        current_logger.warning(
            f"Unable to find the customer to bill for {errors} affiliations in account {account_id}"
        )

    # 3. Package the data
    return [
        InvoiceToGenerateData(
            billing_period=billing_period,
            billing_year=billing_year,
            contract_type="occupational_health",
            contract_ref=customer_contract_ref,
            contract_yearly_price_per_employee_in_cents=YEARLY_PRICE_PER_EMPLOYEE_IN_CENTS,
            contract_needs_installment_plan_for_employees_count=customer_by_contract_ref[
                customer_contract_ref
            ].installment_plan_for_number_of_employees,
            references_of_employees_affiliated_over_the_year={
                str(build_profile_id_from_affiliable_ref(affiliable_ref))
                for affiliable_ref in affiliable_refs
            },
        )
        for customer_contract_ref, affiliable_refs in affiliable_refs_per_customer_ref.items()
    ]

check_data 

check_data(invoices_data)

Temporary function to display the data produced by Occupational Health.

Source code in components/occupational_health/public/queries/billing.py 
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
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
def check_data(invoices_data: list[InvoiceToGenerateData]) -> None:
    """
    Temporary function to display the data produced by Occupational Health.
    """
    for invoice in invoices_data:
        customer_data = get_customer_data(invoice.contract_ref)
        click.secho(
            f"{invoice.billing_period} {invoice.billing_year} invoice for {customer_data.entity_name} ({customer_data.postal_address.city})",
            bold=True,
        )
        click.secho(invoice, dim=True)

        employees_count_to_date = len(
            invoice.references_of_employees_affiliated_over_the_year
        )
        click.secho(f" 👉 {employees_count_to_date} employees to bill to date")

        if invoice.contract_needs_installment_plan:
            click.secho("Installment plan", fg="red")

            ratio = {
                OccupationalHealthBillingPeriod.Q1: 0.25,
                OccupationalHealthBillingPeriod.Q2: 0.5,
                OccupationalHealthBillingPeriod.Q3: 0.75,
                OccupationalHealthBillingPeriod.Q4: 1.0,
                OccupationalHealthBillingPeriod.END_OF_YEAR_REGUL: 0.0,
            }

            # Amount charge _to date_ for the installment plan
            installment_plan_in_cents = int(
                ratio[invoice.billing_period]
                * (
                    invoice.contract_yearly_price_per_employee_in_cents
                    * employees_count_to_date
                )
            )

            total_amount_in_cents = installment_plan_in_cents + (
                invoice.contract_yearly_price_per_employee_in_cents
                * (
                    employees_count_to_date
                    - mandatory(
                        invoice.contract_needs_installment_plan_for_employees_count
                    )
                )
            )
            click.secho(f"   Total amount: {total_amount_in_cents / 100:.2f}€ to date")
            continue

        total_amount_in_cents = (
            invoice.contract_yearly_price_per_employee_in_cents
            * employees_count_to_date
        )
        click.secho(f"   Total amount: {total_amount_in_cents / 100:.2f}€")

get_customer_data 

get_customer_data(contract_ref)

Fetch customer data for the billing system.

Until we decide how to store this customer data (on the Billing side?), here is a query to serve it.

Parameters:

Name Type Description Default
contract_ref str

The contract reference to look up customer data for

 required

Returns:

Name Type Description
CustomerData CustomerData

The customer data for the given contract reference

Raises:

Type Description
ContractRefNotFound

If the contract reference is not found in the customer data mapping
Source code in components/occupational_health/public/queries/billing.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def get_customer_data(contract_ref: str) -> CustomerData:
    """
    Fetch customer data for the billing system.

    Until we decide how to store this customer data (on the Billing side?), here is a query to serve it.

    Args:
        contract_ref: The contract reference to look up customer data for

    Returns:
        CustomerData: The customer data for the given contract reference

    Raises:
        ContractRefNotFound: If the contract reference is not found in the customer data mapping
    """
    from components.occupational_health.internal.business_logic.queries.billing import (
        CUSTOMER_DATA_LIST,
    )

    # Very naive implementation for now
    for customer in CUSTOMER_DATA_LIST:
        if customer.contract_ref == contract_ref:
            return customer

    raise ContractRefNotFound(
        f"Customer data not found for contract reference: {contract_ref}"
    )

has_active_occupational_health_contract

has_active_occupational_health_contract 

has_active_occupational_health_contract(account_id)

Return whether the given account ID has an active occupational health contract at the moment.

Parameters:

Name Type Description Default
account_id UUID

The ID of the account to check.

 required

Returns:

Name Type Description
bool bool

True if the account has an active occupational health contract, False otherwise.
Source code in components/occupational_health/public/queries/has_active_occupational_health_contract.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
def has_active_occupational_health_contract(account_id: UUID) -> bool:
    """
    Return whether the given account ID has an active occupational health contract at the moment.

    Args:
        account_id: The ID of the account to check.

    Returns:
        bool: True if the account has an active occupational health contract, False otherwise.
    """
    from components.occupational_health.internal.business_logic.queries.customers.contract import (
        has_active_contract as _has_active_contract,
    )

    return _has_active_contract(account_id)

is_currently_affiliated_to_prevenir

is_currently_affiliated_to_occupational_health 

is_currently_affiliated_to_occupational_health(
    global_profile_id, account_id, commit=True
)

Check if the user is currently affiliated to Prevenir.

Will create a OccupationalHealthProfile if it doesn't exist, hence the commit parameter.

Returns True if the user has active affiliations, otherwise False.

Source code in components/occupational_health/public/queries/is_currently_affiliated_to_prevenir.py 
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
def is_currently_affiliated_to_occupational_health(
    global_profile_id: UUID,
    account_id: UUID,
    commit: bool = True,
) -> bool:
    """
    Check if the user is currently affiliated to Prevenir.

    Will create a OccupationalHealthProfile if it doesn't exist, hence the commit parameter.

    Returns True if the user has active affiliations, otherwise False.
    """
    occupational_health_profile_id = get_or_create_profile_id_from_global_id(
        global_profile_id,
        commit=commit,
    )

    # If you pass an user (a global profile ID) and an account ID, you can get at most one affiliation.
    active_affiliations = get_active_affiliations(
        occupational_health_profile_id=occupational_health_profile_id,
        for_account_id=account_id,
    )

    if active_affiliations:
        return True

    return False