Skip to content

Api reference

components.occupational_health.public.actions

account

move_occupational_health_account_models_from_source_to_target_account 

move_occupational_health_account_models_from_source_to_target_account(
    source_account_id, target_account_id
)

Function to move all occupational health models linked to an account from a source account to a target account in case of an account being deleted/merged

Source code in components/occupational_health/public/actions/account.py 
 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
def move_occupational_health_account_models_from_source_to_target_account(
    source_account_id: UUID,
    target_account_id: UUID,
) -> None:
    """
    Function to move all occupational health models linked to an account from a source account to a target account
    in case of an account being deleted/merged
    """
    error = None
    messages: list[str] = []
    source_account_id_typed = AccountId(source_account_id)
    target_account_id_typed = AccountId(target_account_id)

    with transaction(propagation=Propagation.REQUIRES_NEW) as session:
        try:
            messages += _move_subscription_model(
                source_account_id_typed, target_account_id_typed
            )
            messages += _move_affiliations(
                source_account_id_typed, target_account_id_typed
            )
            messages += _move_affiliation_strategy_rules(
                source_account_id_typed, target_account_id_typed
            )
            messages += _move_affiliation_movements(
                source_account_id_typed, target_account_id_typed
            )
            messages += _move_dashboard_admins(
                source_account_id_typed, target_account_id_typed
            )
            messages += _move_mass_affiliation_logs(
                source_account_id_typed, target_account_id_typed
            )
            messages += _move_profile_situations(
                source_account_id_typed, target_account_id_typed
            )
            messages += _move_visits(source_account_id_typed, target_account_id_typed)
            messages += _move_workspace_actions(
                source_account_id_typed, target_account_id_typed
            )
            messages += _move_profile_employment_data(
                source_account_id_typed, target_account_id_typed
            )

            # TODO: occupational_health_billed_entity => check logic?
        except Exception as e:
            error = e
            current_logger.exception(
                "Failed to move occupational health models from source account to target account, rollbacking changes",
                source_account_id=source_account_id,
                target_account_id=target_account_id,
            )
            messages.append(
                " ⚠️ Exception occurred while moving occupational health models from source account to target account"
            )
            messages.append(str(e))
            session.rollback()

    if len(messages) > 0:
        _post_slack_notification(source_account_id, target_account_id, messages)

    if error is not None:
        raise error

billing_entity

Public actions for managing occupational health billed entities.

create_billed_entity 

create_billed_entity(
    entity_name,
    siren,
    postal_street,
    postal_code,
    postal_city,
    account_id,
    subscription_ref,
    emails_to_notify,
    siret=None,
    number_of_employees_per_year_for_installment_plan=None,
    save=True,
)

Create a new billed entity for occupational health billing.

Parameters:

Name Type Description Default
entity_name str

Legal name of the entity

 required
siren str

French company identifier (9 digits)

 required
postal_street str

Street address

 required
postal_code str

Postal code

 required
postal_city str

City name

 required
account_id AccountId

Account ID this entity belongs to

 required
subscription_ref UUID

Reference of the subscription

 required
emails_to_notify list[str]

List of HR email addresses to notify

 required
siret str | None

French establishment identifier (14 digits, optional)

 None
number_of_employees_per_year_for_installment_plan dict[str, int] | None

Year-specific employee counts

 None
save bool

Whether to commit the transaction

 True

Returns:

Type Description
OccupationalHealthBilledEntity

The created billed entity
Source code in components/occupational_health/public/actions/billing_entity.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
def create_billed_entity(
    entity_name: str,
    siren: str,
    postal_street: str,
    postal_code: str,
    postal_city: str,
    account_id: AccountId,
    subscription_ref: UUID,
    emails_to_notify: list[str],
    siret: str | None = None,
    number_of_employees_per_year_for_installment_plan: dict[str, int] | None = None,
    save: bool = True,
) -> OccupationalHealthBilledEntity:
    """
    Create a new billed entity for occupational health billing.

    Args:
        entity_name: Legal name of the entity
        siren: French company identifier (9 digits)
        postal_street: Street address
        postal_code: Postal code
        postal_city: City name
        account_id: Account ID this entity belongs to
        subscription_ref: Reference of the subscription
        emails_to_notify: List of HR email addresses to notify
        siret: French establishment identifier (14 digits, optional)
        number_of_employees_per_year_for_installment_plan: Year-specific employee counts
        save: Whether to commit the transaction

    Returns:
        The created billed entity
    """
    return create_billed_entity_internal(
        entity_name=entity_name,
        siren=siren,
        postal_street=postal_street,
        postal_code=postal_code,
        postal_city=postal_city,
        account_id=account_id,
        subscription_ref=subscription_ref,
        emails_to_notify=emails_to_notify,
        siret=siret,
        number_of_employees_per_year_for_installment_plan=number_of_employees_per_year_for_installment_plan,
        save=save,
    )

update_billed_entity 

update_billed_entity(
    billed_entity_id,
    entity_name=NOT_SET,
    siren=NOT_SET,
    siret=NOT_SET,
    postal_street=NOT_SET,
    postal_code=NOT_SET,
    postal_city=NOT_SET,
    subscription_ref=NOT_SET,
    emails_to_notify=NOT_SET,
    number_of_employees_per_year_for_installment_plan=NOT_SET,
    save=True,
)

Update an existing billed entity.

Parameters:

Name Type Description Default
billed_entity_id UUID

ID of the billed entity to update

 required
entity_name NotSet[str]

Legal name of the entity

 NOT_SET
siren NotSet[str]

French company identifier (9 digits)

 NOT_SET
siret NotSet[str | None]

French establishment identifier (14 digits, optional)

 NOT_SET
postal_street NotSet[str]

Street address

 NOT_SET
postal_code NotSet[str]

Postal code

 NOT_SET
postal_city NotSet[str]

City name

 NOT_SET
subscription_ref NotSet[UUID]

Reference of the subscription

 NOT_SET
emails_to_notify NotSet[list[str]]

List of HR email addresses to notify

 NOT_SET
number_of_employees_per_year_for_installment_plan NotSet[dict[str, int] | None]

Year-specific employee counts

 NOT_SET
save bool

Whether to commit the transaction

 True

Returns:

Type Description
OccupationalHealthBilledEntity

The updated billed entity
Source code in components/occupational_health/public/actions/billing_entity.py 
 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
def update_billed_entity(
    billed_entity_id: UUID,
    entity_name: NotSet[str] = NOT_SET,
    siren: NotSet[str] = NOT_SET,
    siret: NotSet[str | None] = NOT_SET,
    postal_street: NotSet[str] = NOT_SET,
    postal_code: NotSet[str] = NOT_SET,
    postal_city: NotSet[str] = NOT_SET,
    subscription_ref: NotSet[UUID] = NOT_SET,
    emails_to_notify: NotSet[list[str]] = NOT_SET,
    number_of_employees_per_year_for_installment_plan: NotSet[
        dict[str, int] | None
    ] = NOT_SET,
    save: bool = True,
) -> OccupationalHealthBilledEntity:
    """
    Update an existing billed entity.

    Args:
        billed_entity_id: ID of the billed entity to update
        entity_name: Legal name of the entity
        siren: French company identifier (9 digits)
        siret: French establishment identifier (14 digits, optional)
        postal_street: Street address
        postal_code: Postal code
        postal_city: City name
        subscription_ref: Reference of the subscription
        emails_to_notify: List of HR email addresses to notify
        number_of_employees_per_year_for_installment_plan: Year-specific employee counts
        save: Whether to commit the transaction

    Returns:
        The updated billed entity
    """
    return update_billed_entity_internal(
        billed_entity_id=billed_entity_id,
        entity_name=entity_name,
        siren=siren,
        siret=siret,
        postal_street=postal_street,
        postal_code=postal_code,
        postal_city=postal_city,
        subscription_ref=subscription_ref,
        emails_to_notify=emails_to_notify,
        number_of_employees_per_year_for_installment_plan=number_of_employees_per_year_for_installment_plan,
        save=save,
    )

members

update_dmst 

update_dmst(
    occupational_health_profile_id, dmst, commit=True
)

Updates the DMST data for a given occupational health profile. We also create the medical secrecy worker record ID, which will be created by the caller on the next request using this ID

See update_curriculum_laboris_job: also creates or updates jobs for the profile

Source code in components/occupational_health/internal/business_logic/actions/members/update_dmst.py 
 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
def update_dmst(
    occupational_health_profile_id: UUID,
    dmst: Dmst,
    commit: bool = True,
) -> None:
    """
    Updates the DMST data for a given occupational health profile. We also create the
    medical secrecy worker record ID, which will be created by the caller on the next
    request using this ID

    See update_curriculum_laboris_job: also creates or updates jobs for the profile
    """

    occupational_health_profile = current_session.get_one(
        OccupationalHealthProfile, occupational_health_profile_id
    )

    if not dmst.medical_secrecy_worker_record_id:
        # Note: we're pre-creating the ID, the caller needs to create the medical secrecy data
        # Dirty hack: we modify the dmst param so that the caller can get the new ID
        occupational_health_profile.medical_secrecy_worker_record_id = uuid.uuid4()
        dmst.medical_secrecy_worker_record_id = (
            occupational_health_profile.medical_secrecy_worker_record_id
        )

    existing_job_ids = {
        job.id
        for job in OccupationalHealthJobBroker.get_existing_jobs(
            occupational_health_profile_id,
        ).all()
    }
    updated_job_ids = {
        job.id for job in dmst.curriculum_laboris.jobs if job.id or set()
    }
    deleted_ids = set(existing_job_ids) - set(updated_job_ids)

    for deleted_id in deleted_ids:
        delete_curriculum_laboris_job(
            occupational_health_profile_id=occupational_health_profile.id,
            job_id=deleted_id,
            commit=False,
        )

    if dmst.curriculum_laboris.jobs:
        for job in dmst.curriculum_laboris.jobs:
            update_curriculum_laboris_job(
                occupational_health_profile_id=occupational_health_profile.id,
                job=job,
                commit=False,
            )

    if dmst.administrative_profile:
        update_administrative_profile(
            occupational_health_profile_id=occupational_health_profile.id,
            health_statuses=dmst.administrative_profile.health_statuses,
            risk_category=dmst.administrative_profile.risk_category,
            notes=dmst.administrative_profile.notes,
            medical_record_sharing_consent=dmst.administrative_profile.medical_record_sharing_consent,
            medical_record_access_consent=dmst.administrative_profile.medical_record_access_consent,
            health_data_sharing_consent=dmst.administrative_profile.health_data_sharing_consent,
            video_consultation_consent=dmst.administrative_profile.video_consultation_consent,
            commit=False,
        )

        update_personal_email(
            profile_id=ProfileId(occupational_health_profile.id),
            personal_email=dmst.administrative_profile.personal_email,
            commit=False,
        )

        update_phone_number(
            profile_id=ProfileId(occupational_health_profile.id),
            phone_number=dmst.administrative_profile.phone_number,
            commit=False,
        )

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

workspace_actions

create_workspace_action 

create_workspace_action(
    thesaurus_mean_id,
    target_type,
    target_id,
    status,
    creator_full_name,
    prevention_type,
    eta_date=None,
    note=None,
    commit=True,
)

Create a new WorkspaceAction for a company or a member

Returns:

Type Description
UUID

The ID of the newly created AMT
Source code in components/occupational_health/internal/business_logic/actions/workspace_actions.py 
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
def create_workspace_action(
    thesaurus_mean_id: uuid.UUID,
    target_type: WorkspaceActionType,
    target_id: uuid.UUID,
    status: WorkspaceActionStatus,
    creator_full_name: str,
    prevention_type: WorkspaceActionPreventionType,
    eta_date: Optional[datetime] = None,
    note: Optional[str] = None,
    commit: bool = True,
) -> uuid.UUID:
    """
    Create a new WorkspaceAction for a company or a member

    Returns:
        The ID of the newly created AMT
    """

    # TODO david.barthelemy: the creator_full_name must be replace as soon as you have a reference to the global profile. For now it's ok to live like this
    thesaurus_mean = get_or_raise_missing_resource(ThesaurusMean, thesaurus_mean_id)
    if target_type == WorkspaceActionType.MEMBER:
        account_id = None
        profile = get_or_raise_missing_resource(OccupationalHealthProfile, target_id)
    elif target_type == WorkspaceActionType.COMPANY:
        account_id = target_id
        profile = None
    else:
        raise NotImplementedError(f"unsupported target type: {target_type}")

    workspace_action = OccupationalHealthWorkspaceAction(
        title=thesaurus_mean.label,
        thesaurus_mean=thesaurus_mean,
        type=target_type,
        profile=profile,
        account_id=account_id,
        prevention_type=prevention_type,
        eta_date=eta_date,
        status=status,
        note=note,
        actor_full_name=creator_full_name,
    )

    current_session.add(workspace_action)

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

    current_logger.info(
        "Created new WorkspaceAction",
        workspace_action_d=str(workspace_action.id),
        target_type=str(target_type),
        target_id=str(target_id),
        status=str(status),
    )

    return workspace_action.id

delete_workspace_action 

delete_workspace_action(workspace_action_id, commit=True)

Delete an existing WorkspaceAction

Source code in components/occupational_health/internal/business_logic/actions/workspace_actions.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def delete_workspace_action(
    workspace_action_id: uuid.UUID,
    commit: bool = True,
) -> None:
    """
    Delete an existing WorkspaceAction
    """
    workspace_action = get_or_raise_missing_resource(
        OccupationalHealthWorkspaceAction, workspace_action_id
    )

    current_session.delete(workspace_action)

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

    current_logger.info(
        "Deleted WorkspaceAction",
        workspace_action_id=str(workspace_action_id),
    )

update_workspace_action 

update_workspace_action(
    workspace_action_id,
    status,
    prevention_type=None,
    eta_date=None,
    note=None,
    commit=True,
)

Update an existing WorkspaceAction

Only updates the fields: status, prevention_type, eta_date, note The target and action (thesaurus_mean) cannot be changed

Source code in components/occupational_health/internal/business_logic/actions/workspace_actions.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
def update_workspace_action(
    workspace_action_id: uuid.UUID,
    status: WorkspaceActionStatus,
    prevention_type: Optional[WorkspaceActionPreventionType] = None,
    eta_date: Optional[datetime] = None,
    note: Optional[str] = None,
    commit: bool = True,
) -> None:
    """
    Update an existing WorkspaceAction

    Only updates the fields: status, prevention_type, eta_date, note
    The target and action (thesaurus_mean) cannot be changed
    """
    workspace_action = get_or_raise_missing_resource(
        OccupationalHealthWorkspaceAction, workspace_action_id
    )

    workspace_action.status = status
    workspace_action.prevention_type = prevention_type
    workspace_action.eta_date = eta_date
    workspace_action.note = note

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

    current_logger.info(
        "Updated WorkspaceAction",
        workspace_action_id=str(workspace_action.id),
        status=str(status),
    )

components.occupational_health.public.dependencies

AccountSearchResult dataclass 

AccountSearchResult(id, name)

Represent a basic Account object retrieved from the Search

id instance-attribute 

id

name instance-attribute 

name

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 
151
152
153
154
155
156
@abstractmethod
def create_profile_with_user(
    self, profile_data: ProfileData
) -> tuple[UserId, GlobalProfileId]:
    """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 
172
173
174
175
176
177
@abstractmethod
def get_account_admins_profile_ids(
    self, account_id: AccountId
) -> set[GlobalProfileId]:
    """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 
58
59
60
61
@abstractmethod
def get_account_id(self, company_id: CompanyId) -> AccountId:
    """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 
68
69
70
71
72
73
@abstractmethod
def get_account_id_and_siret(
    self, company_id: CompanyId
) -> tuple[AccountId, str | None]:
    """Get account ID and SIRET for a company."""
    raise NotImplementedError()

get_account_name abstractmethod 

get_account_name(account_id)

Get account name.

Source code in components/occupational_health/public/dependencies.py 
101
102
103
104
@abstractmethod
def get_account_name(self, account_id: AccountId) -> 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 
179
180
181
182
183
184
@abstractmethod
def get_company_admins_profile_ids(
    self, company_id: CompanyId
) -> set[GlobalProfileId]:
    """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 
94
95
96
97
98
99
@abstractmethod
def get_company_display_name_and_account_id(
    self, company_id: CompanyId
) -> tuple[str, AccountId]:
    """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 
123
124
125
126
@abstractmethod
def get_company_from_siret(self, siret: str) -> Optional["DSNCompanyData"]:
    """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 
63
64
65
66
@abstractmethod
def get_company_ids_in_account(self, account_id: AccountId) -> list[CompanyId]:
    """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 
87
88
89
90
91
92
@abstractmethod
def get_company_names_by_ids(
    self, company_ids: set[CompanyId]
) -> dict[CompanyId, 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 
82
83
84
85
@abstractmethod
def get_company_siren(self, company_id: CompanyId) -> 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 
75
76
77
78
79
80
@abstractmethod
def get_company_siret(
    self, company_id: CompanyId, 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.

Raises:

Type Description
UserIdNotFound

If the user ID does not exist.
Source code in components/occupational_health/public/dependencies.py 
107
108
109
110
111
112
113
114
@abstractmethod
def get_global_profile_id(self, user_id: UserId) -> GlobalProfileId | None:
    """Get global profile ID for a user.

    Raises:
        UserIdNotFound: If the user ID does not exist.
    """
    raise NotImplementedError()

get_ssn_and_ntt_for_user abstractmethod 

get_ssn_and_ntt_for_user(user_id)

Set SSN and NTT on user.

Source code in components/occupational_health/public/dependencies.py 
165
166
167
168
169
170
@abstractmethod
def get_ssn_and_ntt_for_user(
    self, user_id: UserId
) -> tuple[str | None, str | None]:
    """Set SSN and NTT on user."""
    raise NotImplementedError()

get_user_id_by_global_profile_id_mapping_from_global_profile_ids abstractmethod 

get_user_id_by_global_profile_id_mapping_from_global_profile_ids(
    global_profile_ids,
)

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

Source code in components/occupational_health/public/dependencies.py 
128
129
130
131
132
133
@abstractmethod
def get_user_id_by_global_profile_id_mapping_from_global_profile_ids(
    self, global_profile_ids: Iterable[GlobalProfileId]
) -> dict[GlobalProfileId, UserId]:
    """Get global profile ID -> user ID mapping from global health profile IDs."""
    raise NotImplementedError()

get_user_id_by_global_profile_id_mapping_from_user_ids abstractmethod 

get_user_id_by_global_profile_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 
135
136
137
138
139
140
@abstractmethod
def get_user_id_by_global_profile_id_mapping_from_user_ids(
    self, user_ids: Iterable[UserId]
) -> dict[GlobalProfileId, UserId]:
    """Get global profile ID -> user ID mapping from user 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 
116
117
118
119
120
121
@abstractmethod
def get_user_id_from_global_profile_id(
    self, global_profile_id: GlobalProfileId
) -> UserId:
    """Get user ID from global profile ID."""
    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 
142
143
144
145
146
147
148
149
@abstractmethod
def get_work_stoppages_for_users(
    self,
    user_ids: Iterable[UserId],
    ever_active_during_period: tuple[date, date] | None = None,
) -> dict[UserId, list["WorkStoppage"]]:
    """Get work stoppages for users."""
    raise NotImplementedError()

search_accounts_by_name abstractmethod 

search_accounts_by_name(query, account_ids)

Search accounts by name.

Source code in components/occupational_health/public/dependencies.py 
186
187
188
189
190
191
@abstractmethod
def search_accounts_by_name(
    self, query: str, account_ids: Iterable[UUID]
) -> list[AccountSearchResult]:
    """Search accounts by name."""
    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 
158
159
160
161
162
163
@abstractmethod
def set_ssn_ntt_on_user(
    self, user_id: UserId, 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 
203
204
205
206
207
208
209
210
211
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 
194
195
196
197
198
199
200
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 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def occupational_health_employment_change_consumer(  # noqa: D103
    employment_change: "EmploymentChange[FrExtendedValues]",
    event_bus_orchestrator: "EventBusOrchestrator",
) -> None:
    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,
    )

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 
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
def requires_siret_information(account_id: UUID) -> bool:
    """
    Return whether the given account ID requires SIRET information to be provided when tracking employment movements.
    """
    from components.occupational_health.public.queries.has_active_occupational_health_contract import (
        has_active_or_upcoming_occupational_health_contract,
    )

    account_id = AccountId(account_id)

    if not has_active_or_upcoming_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)

Bases: DataClassJsonMixin

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
)

Bases: DataClassJsonMixin

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,
    entity_name,
    siren,
    siret,
    contract_yearly_price_per_employee_in_cents,
    contract_needs_installment_plan_for_employees_count,
    references_of_employees_affiliated_over_the_year
)

Bases: DataClassJsonMixin

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
entity_name instance-attribute 
entity_name
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).

siren instance-attribute 
siren
siret instance-attribute 
siret

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'
to_validity_period 
to_validity_period(year)

The period during which we'd consider affiliations for billing purposes.

Source code in components/occupational_health/public/entities/billing.py 
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
def to_validity_period(self, year: int) -> ValidityPeriod:
    """The period during which we'd consider affiliations for billing purposes."""
    start_date = date(year, 1, 1)
    match self:
        case OccupationalHealthBillingPeriod.Q1:
            # Only consider the affiliations active on 01/01
            end_date = date(year, 1, 1)
        case OccupationalHealthBillingPeriod.Q2:
            # Bill for all affiliations active between 01/01 and 31/03
            end_date = date(year, 3, 31)
        case OccupationalHealthBillingPeriod.Q3:
            # In Q3 we bill for affiliations until the end of Q2
            end_date = date(year, 6, 30)
        case OccupationalHealthBillingPeriod.Q4:
            # In Q4 we bill for affiliations until the end of Q3
            end_date = date(year, 9, 30)
        case OccupationalHealthBillingPeriod.END_OF_YEAR_REGUL:
            # Then at the end of the year, we bill for all affiliations during the year
            end_date = date(year, 12, 31)
        case _:
            raise NotImplementedError(f"Unsupported billing period: {self}")

    return ValidityPeriod(
        start_date=start_date,
        end_date=end_date,
    )

dmst

AdministrativeProfile dataclass 

AdministrativeProfile(
    *,
    ssn,
    health_statuses,
    risk_category,
    notes,
    medical_record_sharing_consent,
    medical_record_access_consent,
    health_data_sharing_consent,
    video_consultation_consent,
    personal_email=None,
    professional_email=None,
    phone_number=None
)

Bases: DataClassJsonMixin

Administrative information about a member's occupational health status.

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

Used in the HP medical app (DMST) for the "Administrative side modal"

health_data_sharing_consent
health_statuses instance-attribute 
health_statuses

medical_record_access_consent

medical_record_sharing_consent
notes instance-attribute 
notes
personal_email class-attribute instance-attribute 
personal_email = None
phone_number class-attribute instance-attribute 
phone_number = None
professional_email class-attribute instance-attribute 
professional_email = None
risk_category instance-attribute 
risk_category
ssn instance-attribute 
ssn

video_consultation_consent

ContractType

Bases: AlanBaseEnum

Contract types from Présanse thesaurus level 1.

cdd class-attribute instance-attribute 
cdd = 'cdd'
cdi class-attribute instance-attribute 
cdi = 'cdi'
contrat_accompagnement_emploi class-attribute instance-attribute 
contrat_accompagnement_emploi = (
    "contrat_accompagnement_emploi"
)
contrat_apprentissage class-attribute instance-attribute 
contrat_apprentissage = 'contrat_apprentissage'
contrat_groupements_employeurs class-attribute instance-attribute 
contrat_groupements_employeurs = (
    "contrat_groupements_employeurs"
)
contrat_initiative_emploi class-attribute instance-attribute 
contrat_initiative_emploi = 'contrat_initiative_emploi'
contrat_intermittent class-attribute instance-attribute 
contrat_intermittent = 'contrat_intermittent'
contrat_professionnalisation class-attribute instance-attribute 
contrat_professionnalisation = (
    "contrat_professionnalisation"
)
contrat_saisonnier class-attribute instance-attribute 
contrat_saisonnier = 'contrat_saisonnier'
contrat_temporaire class-attribute instance-attribute 
contrat_temporaire = 'contrat_temporaire'
contrat_unique_insertion class-attribute instance-attribute 
contrat_unique_insertion = 'contrat_unique_insertion'

CurriculumLaboris dataclass 

CurriculumLaboris(*, occupational_medical_history, 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
occupational_medical_history instance-attribute 
occupational_medical_history

Dmst dataclass 

Dmst(
    *,
    occupational_health_profile_id,
    first_name,
    last_name,
    gender,
    birthdate,
    medical_secrecy_worker_record_id,
    curriculum_laboris,
    administrative_profile,
    health_history=None
)

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)

administrative_profile instance-attribute 
administrative_profile
birthdate instance-attribute 
birthdate
curriculum_laboris instance-attribute 
curriculum_laboris
first_name instance-attribute 
first_name
gender instance-attribute 
gender
health_history class-attribute instance-attribute 
health_history = None
last_name instance-attribute 
last_name
medical_secrecy_worker_record_id instance-attribute 
medical_secrecy_worker_record_id
occupational_health_profile_id instance-attribute 
occupational_health_profile_id

MemberJob dataclass 

MemberJob(
    *,
    id,
    title,
    start_date,
    end_date,
    employer,
    is_past_job,
    medical_secrecy_worker_job_id,
    description,
    working_hours,
    missions,
    physical_conditions,
    organizational_conditions,
    mental_conditions,
    tools,
    worn_equipment,
    risks_and_advice,
    contract_type,
    exposition_notations
)

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

contract_type instance-attribute 
contract_type
description instance-attribute 
description
employer instance-attribute 
employer
end_date instance-attribute 
end_date
exposition_notations instance-attribute 
exposition_notations
id instance-attribute 
id
is_past_job instance-attribute 
is_past_job
medical_secrecy_worker_job_id instance-attribute 
medical_secrecy_worker_job_id
mental_conditions instance-attribute 
mental_conditions
missions instance-attribute 
missions
organizational_conditions instance-attribute 
organizational_conditions
physical_conditions instance-attribute 
physical_conditions
risks_and_advice instance-attribute 
risks_and_advice
start_date instance-attribute 
start_date
title instance-attribute 
title
tools instance-attribute 
tools
working_hours instance-attribute 
working_hours
worn_equipment instance-attribute 
worn_equipment

OccupationalMedicalHistory dataclass 

OccupationalMedicalHistory(*, types, description)

Bases: DataClassJsonMixin

Represents the medical history pertaining to occupational health.

Note: this is different from the general HealthHistory, which is not tied to occupational (work) history

description instance-attribute 
description
types instance-attribute 
types

OccupationalMedicalHistoryType

Bases: AlanBaseEnum

Representing types of occupational medical history, can be multiple

occupational_accident class-attribute instance-attribute 
occupational_accident = 'occupational_accident'
occupational_disease class-attribute instance-attribute 
occupational_disease = 'occupational_disease'

health_history

Health data entities for occupational health worker medical records.

These entities correspond to the TypeScript interfaces in: frontend/apps/medical-software/app/hooks/types.ts

NOTE: This is a duplicate of components.medical_secrecy.internal.entities.occupational_health_worker_medical_record_health_history to avoid cross-component dependencies. Keep in sync manually. TODO: @david.barthelemy: clean this after talking with Alexandre and Mickael

AlcoholStatus

Bases: AlanBaseEnum

Alcohol consumption status.

addicted class-attribute instance-attribute 
addicted = 'addicted'
casual class-attribute instance-attribute 
casual = 'casual'
deprived class-attribute instance-attribute 
deprived = 'deprived'
no class-attribute instance-attribute 
no = 'no'

AllergyItem dataclass 

AllergyItem(
    *,
    label=None,
    status=None,
    start_date_year=None,
    observation=None
)

Bases: DataClassJsonMixin

Allergy item.

Corresponds to AllergyItem TypeScript interface.

label class-attribute instance-attribute 
label = None
observation class-attribute instance-attribute 
observation = None
start_date_year class-attribute instance-attribute 
start_date_year = None
status class-attribute instance-attribute 
status = None

AllergyStatus

Bases: AlanBaseEnum

Status of an allergy.

cured class-attribute instance-attribute 
cured = 'cured'
in_treatment class-attribute instance-attribute 
in_treatment = 'in_treatment'
untreated class-attribute instance-attribute 
untreated = 'untreated'

EyeHealthItem dataclass 

EyeHealthItem(
    *,
    optical_correction=None,
    ophthalmologic_follow_up=None,
    observation=None
)

Bases: DataClassJsonMixin

Corresponds to EyeHealthItem TypeScript interface.

observation class-attribute instance-attribute 
observation = None
ophthalmologic_follow_up class-attribute instance-attribute 
ophthalmologic_follow_up = None
optical_correction class-attribute instance-attribute 
optical_correction = None

FamilyHistoryItem dataclass 

FamilyHistoryItem(
    *, label=None, family_members=list(), observation=None
)

Bases: DataClassJsonMixin

Family history item.

Corresponds to FamilyHistoryItem TypeScript interface.

family_members class-attribute instance-attribute 
family_members = field(default_factory=list)
label class-attribute instance-attribute 
label = None
observation class-attribute instance-attribute 
observation = None

FamilyHistoryMember dataclass 

FamilyHistoryMember(*, age=None, status=None)

Bases: DataClassJsonMixin

Family history member information.

Corresponds to FamilyHistoryMember TypeScript interface.

age class-attribute instance-attribute 
age = None
status class-attribute instance-attribute 
status = None

FamilyStatus

Bases: AlanBaseEnum

Family member status.

aunt class-attribute instance-attribute 
aunt = 'aunt'
brother class-attribute instance-attribute 
brother = 'brother'
father class-attribute instance-attribute 
father = 'father'
grand_father class-attribute instance-attribute 
grand_father = 'grand_father'
grand_mother class-attribute instance-attribute 
grand_mother = 'grand_mother'
great_grand_father class-attribute instance-attribute 
great_grand_father = 'great_grand_father'
great_grand_mother class-attribute instance-attribute 
great_grand_mother = 'great_grand_mother'
mother class-attribute instance-attribute 
mother = 'mother'
other class-attribute instance-attribute 
other = 'other'
sister class-attribute instance-attribute 
sister = 'sister'
uncle class-attribute instance-attribute 
uncle = 'uncle'

HealthHistory dataclass 

HealthHistory(
    *,
    medical_surgical_history=list(),
    allergies=list(),
    treatment=None,
    vaccine=None,
    family_history=list(),
    lifestyle=None,
    transport_mode=None,
    personal_situation=None,
    gynecological_follow_up=None,
    gynecological_observation=None,
    contraceptives=None,
    eye_health=None,
    medical_observation=None
)

Bases: DataClassJsonMixin

Health history for occupational health worker medical record.

This dataclass contains all health-related information that needs to be encrypted and stored in the OccupationalHealthWorkerMedicalRecord.health_history field.

All fields correspond to the forms defined in: frontend/apps/medical-software/app/pages/MemberHealthTab.tsx

allergies class-attribute instance-attribute 
allergies = field(default_factory=list)
contraceptives class-attribute instance-attribute 
contraceptives = None
eye_health class-attribute instance-attribute 
eye_health = None
family_history class-attribute instance-attribute 
family_history = field(default_factory=list)
gynecological_follow_up class-attribute instance-attribute 
gynecological_follow_up = None
gynecological_observation class-attribute instance-attribute 
gynecological_observation = None
lifestyle class-attribute instance-attribute 
lifestyle = None
medical_observation class-attribute instance-attribute 
medical_observation = None
medical_surgical_history class-attribute instance-attribute 
medical_surgical_history = field(default_factory=list)
personal_situation class-attribute instance-attribute 
personal_situation = None
transport_mode class-attribute instance-attribute 
transport_mode = None
treatment class-attribute instance-attribute 
treatment = None
vaccine class-attribute instance-attribute 
vaccine = None

LifestyleItem dataclass 

LifestyleItem(
    *,
    alcohol_status=None,
    alcohol_observation=None,
    smoking_status=None,
    smoking_observation=None,
    other_addictive_behaviors=None,
    physical_activity_status=None,
    physical_activity_observations=None,
    food=None,
    sleep=None,
    lifestyle_observations=None
)

Bases: DataClassJsonMixin

Lifestyle information.

Corresponds to LifestyleItem TypeScript interface.

alcohol_observation class-attribute instance-attribute 
alcohol_observation = None
alcohol_status class-attribute instance-attribute 
alcohol_status = None
food class-attribute instance-attribute 
food = None
lifestyle_observations class-attribute instance-attribute 
lifestyle_observations = None
other_addictive_behaviors class-attribute instance-attribute 
other_addictive_behaviors = None
physical_activity_observations class-attribute instance-attribute 
physical_activity_observations = None
physical_activity_status class-attribute instance-attribute 
physical_activity_status = None
sleep class-attribute instance-attribute 
sleep = None
smoking_observation class-attribute instance-attribute 
smoking_observation = None
smoking_status class-attribute instance-attribute 
smoking_status = None

LivingStatus

Bases: AlanBaseEnum

Living status.

alone class-attribute instance-attribute 
alone = 'alone'
couple class-attribute instance-attribute 
couple = 'couple'
family class-attribute instance-attribute 
family = 'family'
house_share class-attribute instance-attribute 
house_share = 'house_share'
other class-attribute instance-attribute 
other = 'other'
single_parent_family class-attribute instance-attribute 
single_parent_family = 'single_parent_family'

MedicalSurgicalHistoryItem dataclass 

MedicalSurgicalHistoryItem(
    *,
    label=None,
    status=None,
    start_date_year=None,
    observation=None
)

Bases: DataClassJsonMixin

Medical and surgical history item.

Corresponds to MedicalSurgicalHistoryItem TypeScript interface.

label class-attribute instance-attribute 
label = None
observation class-attribute instance-attribute 
observation = None
start_date_year class-attribute instance-attribute 
start_date_year = None
status class-attribute instance-attribute 
status = None

MedicalSurgicalHistoryStatus

Bases: AlanBaseEnum

Status of a medical or surgical history item.

cured class-attribute instance-attribute 
cured = 'cured'
in_treatment class-attribute instance-attribute 
in_treatment = 'in_treatment'
relapse class-attribute instance-attribute 
relapse = 'relapse'
remission class-attribute instance-attribute 
remission = 'remission'

PersonalSituationItem dataclass 

PersonalSituationItem(
    *, living_status=None, number_of_children=None
)

Bases: DataClassJsonMixin

Corresponds to PersonalSituationItem TypeScript interface.

living_status class-attribute instance-attribute 
living_status = None
number_of_children class-attribute instance-attribute 
number_of_children = None

PhysicalActivityStatus

Bases: AlanBaseEnum

Physical activity status.

less_than_1 class-attribute instance-attribute 
less_than_1 = 'less_than_1'
less_than_2 class-attribute instance-attribute 
less_than_2 = 'less_than_2'
more_than_2 class-attribute instance-attribute 
more_than_2 = 'more_than_2'
no class-attribute instance-attribute 
no = 'no'

SmokingStatus

Bases: AlanBaseEnum

Smoking status.

deprived class-attribute instance-attribute 
deprived = 'deprived'
less_than_5 class-attribute instance-attribute 
less_than_5 = 'less_than_5'
less_than_9 class-attribute instance-attribute 
less_than_9 = 'less_than_9'
more_than_9 class-attribute instance-attribute 
more_than_9 = 'more_than_9'
no class-attribute instance-attribute 
no = 'no'
passive class-attribute instance-attribute 
passive = 'passive'

TransportModeItem dataclass 

TransportModeItem(
    *,
    travel_time=None,
    transport_mode=None,
    is_full_remote=None
)

Bases: DataClassJsonMixin

Corresponds to TransportModeItem TypeScript interface.

is_full_remote class-attribute instance-attribute 
is_full_remote = None
transport_mode class-attribute instance-attribute 
transport_mode = None
travel_time class-attribute instance-attribute 
travel_time = None

TreatmentItem dataclass 

TreatmentItem(*, description=None)

Bases: DataClassJsonMixin

Treatment item.

Corresponds to TreatmentItem TypeScript interface.

description class-attribute instance-attribute 
description = None

VaccineItem dataclass 

VaccineItem(*, status=None, description=None)

Bases: DataClassJsonMixin

Vaccination item.

Corresponds to VaccineItem TypeScript interface.

description class-attribute instance-attribute 
description = None
status class-attribute instance-attribute 
status = None

VaccineStatus

Bases: AlanBaseEnum

Status of vaccination.

need_update class-attribute instance-attribute 
need_update = 'need_update'
unknown class-attribute instance-attribute 
unknown = 'unknown'
up_to_date class-attribute instance-attribute 
up_to_date = 'up_to_date'

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

thesaurus

ThesaurusMean dataclass 

ThesaurusMean(*, id, label)

Bases: DataClassJsonMixin

The public dataclass of a ThesaurusMean from Presanse

id instance-attribute 
id
label instance-attribute 
label

ThesaurusOccupationalExposure dataclass 

ThesaurusOccupationalExposure(
    *,
    notation,
    label,
    dc_type,
    status,
    created_at,
    version,
    line_number,
    risk_category,
    parent_notation,
    class_label=None,
    subclass_label=None,
    class_notation=None,
    subclass_notation=None
)

Bases: DataClassJsonMixin

A TEP (Thésaurus des Expositions Professionnelles) exposition entry.

class_label class-attribute instance-attribute 
class_label = None
class_notation class-attribute instance-attribute 
class_notation = None
created_at instance-attribute 
created_at
dc_type instance-attribute 
dc_type
label instance-attribute 
label
line_number instance-attribute 
line_number
notation instance-attribute 
notation
parent_notation instance-attribute 
parent_notation
risk_category instance-attribute 
risk_category
status instance-attribute 
status
subclass_label class-attribute instance-attribute 
subclass_label = None
subclass_notation class-attribute instance-attribute 
subclass_notation = None
version instance-attribute 
version

visit

PlannedVisit dataclass 

PlannedVisit(profile_id, timestamp_planned)

Bases: DataClassJsonMixin

Planned visit information

profile_id instance-attribute 
profile_id
timestamp_planned instance-attribute 
timestamp_planned

VisitInfo dataclass 

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

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
occupational_health_profile_id instance-attribute 
occupational_health_profile_id
visit_date instance-attribute 
visit_date
visit_id instance-attribute 
visit_id
visit_status instance-attribute 
visit_status
visit_type instance-attribute 
visit_type

workspace_action

WorkspaceAction dataclass 

WorkspaceAction(
    id,
    title,
    target_type,
    status,
    company_name=None,
    members_count=None,
    member_full_name=None,
    job_title=None,
    health_professional_fullname=None,
    profile_id=None,
    account_id=None,
    eta_date=None,
    prevention_type=None,
    note=None,
)

Bases: DataClassJsonMixin

Represents a WorkspaceAction (an AMT in France) in the context of occupational health.

account_id class-attribute instance-attribute 
account_id = None
company_name class-attribute instance-attribute 
company_name = None
eta_date class-attribute instance-attribute 
eta_date = None
health_professional_fullname class-attribute instance-attribute 
health_professional_fullname = None
id instance-attribute 
id
job_title class-attribute instance-attribute 
job_title = None
member_full_name class-attribute instance-attribute 
member_full_name = None
members_count class-attribute instance-attribute 
members_count = None
note class-attribute instance-attribute 
note = None
prevention_type class-attribute instance-attribute 
prevention_type = None
profile_id class-attribute instance-attribute 
profile_id = None
status instance-attribute 
status
target_type instance-attribute 
target_type
title instance-attribute 
title

WorkspaceActionPreventionType

Bases: AlanBaseEnum

Represents the type of prevention of the WorkspaceAction

primary class-attribute instance-attribute 
primary = 'primary'
secondary class-attribute instance-attribute 
secondary = 'secondary'
tertiary class-attribute instance-attribute 
tertiary = 'tertiary'

WorkspaceActionStatus

Bases: AlanBaseEnum

Represents a current status of the WorkspaceAction

in_progress class-attribute instance-attribute 
in_progress = 'in_progress'
todo class-attribute instance-attribute 
todo = 'todo'
validated class-attribute instance-attribute 
validated = 'validated'

WorkspaceActionType

Bases: AlanBaseEnum

Represents the target of the WorkspaceAction. For now it's only a member or a company

company class-attribute instance-attribute 
company = 'company'
member class-attribute instance-attribute 
member = 'member'

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'

RiskCategory

Bases: AlanBaseEnum

Risk categories / "type de suivi" for occupational health visits.

These determine the frequency and type of required medical visits.

SI class-attribute instance-attribute 

SI = 'SI'

SIA class-attribute instance-attribute 

SIA = 'SIA'

SIR class-attribute instance-attribute 

SIR = 'SIR'

WorkerHealthStatus

Bases: AlanBaseEnum

Corresponds to "type de suivi" in the DMST administrative profile

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

INV1 class-attribute instance-attribute 

INV1 = 'INV1'

INV2 class-attribute instance-attribute 

INV2 = 'INV2'

INV3 class-attribute instance-attribute 

INV3 = 'INV3'

Inaptitude class-attribute instance-attribute 

Inaptitude = 'Inaptitude'

PDP class-attribute instance-attribute 

PDP = 'PDP'

RQTH class-attribute instance-attribute 

RQTH = 'RQTH'

SI class-attribute instance-attribute 

SI = 'SI'

SIA class-attribute instance-attribute 

SIA = 'SIA'

SIR class-attribute instance-attribute 

SIR = 'SIR'

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 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def subscribe_to_events() -> None:
    """
    All event subscriptions for the occupational health component should be done here.
    """
    from components.occupational_health.internal.profile.actions.update_occupational_health_profile_merge import (
        update_occupational_health_profile_merge,
    )
    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.exceptions

CompanyIdNotFound 

CompanyIdNotFound(company_id)

Bases: ConsumerError

The company ID mentioned in the employment change could not be found.

Should only happen within unit tests, or if a blocked movement has its company deleted (very rare)

Source code in components/occupational_health/public/exceptions.py 
36
37
38
39
40
def __init__(self, company_id: CompanyId) -> None:
    super().__init__(
        message=f"Company ID '{company_id}' not found",
        consumer=occupational_health_employment_change_consumer.__name__,
    )

error_code class-attribute instance-attribute 

error_code = 'company_id_not_found'

UserIdNotFound 

UserIdNotFound(message)

Bases: UserNotFoundError

A user ID cannot be found.

This happens generally after a user merge event, when the employment change still refers the old user ID.

Source code in components/occupational_health/public/exceptions.py 
20
21
22
23
24
def __init__(self, message: str) -> None:
    super().__init__(
        message=message,
        consumer=occupational_health_employment_change_consumer.__name__,
    )

components.occupational_health.public.marmot

actions

Actions for use in Marmot controllers.

AffiliationRequest dataclass 

AffiliationRequest(
    user_id,
    start_date,
    risk_category,
    extra_data,
    last_visit=None,
    personal_email=None,
    professional_email=None,
)

Bases: DataClassJsonMixin

extra_data instance-attribute 
extra_data
last_visit class-attribute instance-attribute 
last_visit = None
personal_email class-attribute instance-attribute 
personal_email = None
professional_email class-attribute instance-attribute 
professional_email = None
risk_category instance-attribute 
risk_category
start_date instance-attribute 
start_date
user_id instance-attribute 
user_id

DuplicateAdminError

Bases: Exception

Raised when trying to add a duplicate occupational health dashboard admin.

RuleAction

Bases: AlanBaseEnum

Which action to take when affiliating employees matching the rule.

AFFILIATE class-attribute instance-attribute 
AFFILIATE = 'affiliate'

Affiliate the employee automatically

IGNORE class-attribute instance-attribute 
IGNORE = 'ignore'

Do not affiliate the employee

REQUIRE_MANUAL_REVIEW class-attribute instance-attribute 
REQUIRE_MANUAL_REVIEW = 'require_manual_review'

Ask the admin to decide whether to affiliate the employee or not

add_occupational_health_dashboard_admin_to_account 

add_occupational_health_dashboard_admin_to_account(
    account_id, user_id
)

Add an occupational health dashboard admin to an account.

Raises:

Type Description
DuplicateAdminError

If the user is already an admin for this account.
Source code in components/occupational_health/internal/business_logic/actions/dashboard_admin.py 
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 add_occupational_health_dashboard_admin_to_account(
    account_id: AccountId, user_id: UserId
) -> UUID:
    """
    Add an occupational health dashboard admin to an account.

    Raises:
        DuplicateAdminError: If the user is already an admin for this account.
    """

    # Check if admin already exists for this user and account
    existing_admin = current_session.scalar(
        select(OccupationalHealthDashboardAdmin).where(
            OccupationalHealthDashboardAdmin.user_id == user_id,
            OccupationalHealthDashboardAdmin.account_id == account_id,
        )
    )

    if existing_admin:
        raise DuplicateAdminError(
            f"User {user_id} is already an admin for account {account_id}. "
            "Cannot add duplicate admin."
        )

    current_logger.info(
        f"Adding occupational health dashboard admin user {user_id} to account {account_id}"
    )
    occupational_health_dashboard_admin = OccupationalHealthDashboardAdmin(
        account_id=account_id,
        user_id=user_id,
        scope=OccupationalHealthDashboardAdminScope.ACCOUNT,
    )
    current_session.add(occupational_health_dashboard_admin)
    current_session.commit()

    return occupational_health_dashboard_admin.id

affiliate_multiple_members 

affiliate_multiple_members(account_id, affiliations)

Affiliate multiple members to Prévenir in a single transaction.

Parameters:

Name Type Description Default
account_id AccountId

The account ID to affiliate members to

 required
affiliations list[AffiliationRequest]

List of affiliation requests with user_id, start_date, optional last_visit, and optional risk_category

 required

Returns:

Type Description
list[UUID]

List of affiliation IDs
Source code in components/occupational_health/internal/business_logic/actions/affiliation_tool.py 
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
def affiliate_multiple_members(
    account_id: AccountId,
    affiliations: list[AffiliationRequest],
) -> list[UUID]:
    """
    Affiliate multiple members to Prévenir in a single transaction.

    Args:
        account_id: The account ID to affiliate members to
        affiliations: List of affiliation requests with user_id, start_date, optional last_visit, and optional risk_category

    Returns:
        List of affiliation IDs
    """
    current_logger.info(
        f"Affiliating {len(affiliations)} members to account {account_id}..."
    )
    affiliation_ids: list[UUID] = []

    for affiliation_request in affiliations:
        # Get or create occupational health profile ID from user_id
        profile_id = get_or_create_profile_id(affiliation_request.user_id, commit=False)
        affiliation_id = affiliate_member(
            occupational_health_profile_id=profile_id,
            start_date=affiliation_request.start_date,
            end_date=None,
            account_id=account_id,
            commit=False,
        )
        affiliation_ids.append(affiliation_id)

        # Update personal email if provided
        if affiliation_request.personal_email:
            update_personal_email(
                profile_id=profile_id,
                personal_email=affiliation_request.personal_email,
                commit=False,
            )

        # Update professional email if provided
        if affiliation_request.professional_email:
            update_professional_email(
                profile_id=profile_id,
                account_id=account_id,
                professional_email=affiliation_request.professional_email,
                commit=False,
            )

        # Store risk category in administrative profile if provided
        if affiliation_request.risk_category:
            update_administrative_profile(
                occupational_health_profile_id=profile_id,
                risk_category=affiliation_request.risk_category,
                commit=False,
            )

        affiliation_log = OccupationalHealthMassAffiliationLog(
            profile_id=profile_id,
            account_id=account_id,
            start_date=affiliation_request.start_date,
            extra_data=affiliation_request.extra_data,
            affiliation_id=affiliation_id,
        )
        current_session.add(affiliation_log)

        # Create visit record if last_visit data is provided
        if affiliation_request.last_visit:
            visit_type = affiliation_request.last_visit.visit_type
            visit_date = affiliation_request.last_visit.date

            create_backfilled_visit_if_not_exists(
                profile_id=profile_id,
                visit_date=visit_date,
                visit_type=visit_type,
                account_id=account_id,
            )

    current_session.commit()

    return affiliation_ids

affiliate_new_member_for_marmot 

affiliate_new_member_for_marmot(
    *,
    user_id,
    account_id,
    start_date,
    end_date=None,
    personal_email=None,
    professional_email=None,
    last_visit=None,
    risk_category=None,
    actor_user_id=None
)

Affiliate a new member to occupational health.

Source code in components/occupational_health/public/marmot/actions.py 
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
377
378
379
380
381
382
383
384
385
386
387
388
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
def affiliate_new_member_for_marmot(
    *,
    user_id: UserId,
    account_id: AccountId,
    start_date: date,
    end_date: date | None = None,
    personal_email: str | None = None,
    professional_email: str | None = None,
    last_visit: "LastVisit | None" = None,
    risk_category: "RiskCategory | None" = None,
    actor_user_id: UserId | str | None = None,
) -> None:
    """
    Affiliate a new member to occupational health.
    """
    from components.occupational_health.internal.business_logic.actions.members.update_administrative_profile import (
        update_administrative_profile,
    )

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

    # Update personal email if provided
    if personal_email:
        update_personal_email(
            profile_id=profile_id,
            personal_email=personal_email,
        )

    # Update professional email if provided
    if professional_email:
        update_professional_email(
            profile_id=profile_id,
            account_id=account_id,
            professional_email=professional_email,
        )

    # Store risk category in administrative profile if provided
    if risk_category:
        update_administrative_profile(
            occupational_health_profile_id=profile_id,
            risk_category=risk_category,
        )

    # Create visit record if last_visit data is provided
    if last_visit:
        create_backfilled_visit_if_not_exists(
            profile_id=profile_id,
            visit_date=last_visit.date,
            visit_type=last_visit.visit_type,
            account_id=account_id,
            commit=True,
        )

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 
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
def cancel_affiliation_for_marmot(
    affiliation_id: UUID,
    actor_user_id: UserId | 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)

create_affiliation_strategy_rule_for_marmot 

create_affiliation_strategy_rule_for_marmot(
    *, account_id, company_id=None, siret=None, name, action
)

Create a new affiliation strategy rule for a company or SIRET.

Either company_id or siret must be provided, but not both.

Source code in components/occupational_health/public/marmot/actions.py 
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
485
486
487
488
489
490
491
492
493
494
495
def create_affiliation_strategy_rule_for_marmot(
    *,
    account_id: AccountId,
    company_id: CompanyId | None = None,
    siret: str | None = None,
    name: str | None,
    action: RuleAction,
) -> UUID:
    """
    Create a new affiliation strategy rule for a company or SIRET.

    Either company_id or siret must be provided, but not both.
    """
    from components.occupational_health.internal.models.affiliation_strategy_rule import (
        AffiliationStrategyRule,
    )

    if not company_id and not siret:
        raise BaseErrorCode.invalid_arguments(
            message="Either company_id or siret must be provided"
        )
    if company_id and siret:
        raise BaseErrorCode.invalid_arguments(
            message="Cannot provide both company_id and siret"
        )

    current_logger.info(
        "Creating affiliation strategy rule from Marmot",
        extra={
            "account_id": account_id,
            "company_id": company_id,
            "siret": siret,
            "name": name,
            "action": action,
        },
    )

    # Check if rule already exists
    query = select(AffiliationStrategyRule).where(
        AffiliationStrategyRule.account_id == account_id,
    )
    if company_id:
        query = query.where(AffiliationStrategyRule.company_id == company_id)
    if siret:
        query = query.where(AffiliationStrategyRule.siret == siret)

    existing_rule = current_session.scalars(query).first()

    if existing_rule:
        rule_type = f"company {company_id}" if company_id else f"SIRET {siret}"
        raise BaseErrorCode.invalid_arguments(
            message=f"Rule already exists for {rule_type} in account {account_id}"
        )

    # Create new rule
    new_rule = AffiliationStrategyRule(
        account_id=account_id,
        company_id=company_id,
        siret=siret,
        name=name,
        action=action,
        is_missing_siret=False,
    )

    current_session.add(new_rule)
    current_session.commit()

    current_logger.info(
        f"Created new affiliation strategy rule: {new_rule.id}",
        extra={
            "rule_id": new_rule.id,
            "account_id": account_id,
            "company_id": company_id,
            "siret": siret,
            "name": name,
            "action": action,
        },
    )

    return new_rule.id

create_contract 

create_contract(session, account_id, start_date)
Source code in components/occupational_health/internal/business_logic/contracting/actions.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
@transactional(propagation=Propagation.REQUIRES_NEW)
def create_contract(
    session: Session,  # noqa:ARG001
    account_id: AccountId,
    start_date: date,
) -> UUID:
    from components.contracting.subcomponents.subscription.public.actions import (
        initialize_subscription,
        record_subscription_updates,
    )
    from components.contracting.subcomponents.subscription.public.entities import (
        SubscriptionScope,
        SubscriptionUpdateRequest,
    )

    subscription = initialize_subscription(
        subscription_scope=SubscriptionScope.occupational_health,
        owner_type=SubscriptionOwnerType.ACCOUNT,
        owner_ref=str(account_id),
    )
    current_logger.info(
        f"Created subscription {subscription.id} for account {account_id}"
    )

    updates = [
        SubscriptionUpdateRequest(
            validity_period=ValidityPeriod(
                start_date=start_date,
                end_date=None,
            ),
            # mandatory but not used
            payload_ref=UUID("00000000-0000-0000-0000-000000000000"),
            operation_ref=None,
            is_deletion=False,
        ),
    ]

    record_subscription_updates(
        subscription_scope=SubscriptionScope.occupational_health,
        # NOTE: for now "subscription_ref" was used in the context of affiliations,
        # and was a string of the format "occhealth:account:<account_id>" (see
        # build_subscription_ref()). Here the subscription ref is just the
        # subscription UUID, which could be confusing: let's not expose that
        # "contracting" subscription_ref under that name at this point.
        subscription_ref=str(subscription.id),
        updates=updates,
        commit=False,
    )
    current_logger.info(
        f"Added subscription version for subscription {subscription.id} and start date {start_date}"
    )

    return subscription.id

update_employment_nic 

update_employment_nic(employment_id, nic)

Update the NIC of an employment.

Source code in components/occupational_health/public/marmot/actions.py 
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
def update_employment_nic(
    employment_id: UUID,
    nic: str | None,
) -> None:
    """
    Update the NIC of an employment.
    """
    from components.employment.public.business_logic.queries.core_employment_version import (
        get_latest_version_for_employment_or_raise,
    )
    from components.fr.public.employment.fr_extended_values import (
        FrExtendedValues,
    )
    from shared.temporal.validity_period import ValidityPeriod

    employment_version = get_latest_version_for_employment_or_raise(employment_id)
    validity_period = ValidityPeriod(
        start_date=employment_version.start_date,
        end_date=employment_version.end_date,
    )

    set_extended_employment_values(
        employment_id=employment_id,
        values=FrExtendedValues({"nic": nic}),
        source_type=SourceType.fr_marmot_occupational_health,
        source_information=SourceInformation(
            raw_data={},
            metadata={"endpoint": "occupational_health.update_employment"},
        ),
        validity_period=validity_period,
        commit=True,
    )

    current_logger.info(
        "Updated NIC of employment",
        employment_id=employment_id,
        nic=nic,
    )

entities

AccountForMarmot dataclass 

AccountForMarmot(id, name)

Bases: DataClassJsonMixin

Represent an account with an occupational health contract for display in Marmot.

id instance-attribute 
id
name instance-attribute 
name

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

CompanyWithAffiliatedCount dataclass 

CompanyWithAffiliatedCount(
    company_id,
    company_display_name,
    ever_affiliated_members_count,
    sirets,
)

Bases: DataClassJsonMixin

Represent a company with its count of affiliated members for display in Marmot.

company_display_name instance-attribute 
company_display_name
company_id instance-attribute 
company_id
ever_affiliated_members_count instance-attribute 
ever_affiliated_members_count
sirets instance-attribute 
sirets

ContractInfo dataclass 

ContractInfo(account_id, start_date, end_date)

Bases: DataClassJsonMixin

Information about a contract.

account_id instance-attribute 
account_id
end_date instance-attribute 
end_date
start_date instance-attribute 
start_date

DiscrepancyForMarmot dataclass 

DiscrepancyForMarmot(
    account_id,
    profile_id,
    user_id,
    category,
    message,
    employments_timeline=None,
    affiliations_timeline=None,
    *,
    first_name,
    last_name,
    birth_date,
    employments
)

Bases: Discrepancy

Represent a discrepancy found in an affiliation for display in Marmot.

birth_date instance-attribute 
birth_date
employments instance-attribute 
employments
first_name instance-attribute 
first_name
last_name instance-attribute 
last_name

LastVisit dataclass 

LastVisit(date, visit_type)

Bases: DataClassJsonMixin

Information about the last occupational health visit.

date instance-attribute 
date
visit_type instance-attribute 
visit_type

MemberToAffiliate dataclass 

MemberToAffiliate(
    row_index,
    first_name,
    last_name,
    birthdate,
    ssn,
    professional_email,
    personal_email,
    proposed_user_id,
    proposed_user_name,
    proposed_birthdate,
    matched_name,
    matched_birthdate,
    matching_result,
    affiliations,
    employments,
    target_start_date,
    last_visit,
    risk_category,
    extra_data,
)

Bases: DataClassJsonMixin

Result of matching a spreadsheet row.

affiliations instance-attribute 
affiliations
birthdate instance-attribute 
birthdate
employments instance-attribute 
employments
extra_data instance-attribute 
extra_data
first_name instance-attribute 
first_name
last_name instance-attribute 
last_name
last_visit instance-attribute 
last_visit
matched_birthdate instance-attribute 
matched_birthdate
matched_name instance-attribute 
matched_name
matching_result instance-attribute 
matching_result
personal_email instance-attribute 
personal_email
professional_email instance-attribute 
professional_email
proposed_birthdate instance-attribute 
proposed_birthdate
proposed_user_id instance-attribute 
proposed_user_id
proposed_user_name instance-attribute 
proposed_user_name
risk_category instance-attribute 
risk_category
row_index instance-attribute 
row_index
ssn instance-attribute 
ssn
target_start_date instance-attribute 
target_start_date

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

RiskCategory

Bases: AlanBaseEnum

Risk categories / "type de suivi" for occupational health visits.

These determine the frequency and type of required medical visits.

SI class-attribute instance-attribute 
SI = 'SI'
SIA class-attribute instance-attribute 
SIA = 'SIA'
SIR class-attribute instance-attribute 
SIR = 'SIR'

SiretWithMemberCount dataclass 

SiretWithMemberCount(siret, member_count, pretty_name)

Bases: DataClassJsonMixin

Represent a SIRET with its count of members.

member_count instance-attribute 
member_count
pretty_name instance-attribute 
pretty_name

Company name from DSN, if available. Format: 'Name (City)' or None if not found.

siret instance-attribute 
siret

VisitType

Bases: AlanBaseEnum

EXAMEN_MEDICAL_APTITUDE_EMBAUCHE class-attribute instance-attribute 
EXAMEN_MEDICAL_APTITUDE_EMBAUCHE = (
    "3.examen_medical_aptitude_embauche"
)
EXAMEN_MEDICAL_APTITUDE_PERIODIQUE class-attribute instance-attribute 
EXAMEN_MEDICAL_APTITUDE_PERIODIQUE = (
    "4.examen_medical_aptitude_periodique"
)
VISITE_CESSATION_EXPOSITION class-attribute instance-attribute 
VISITE_CESSATION_EXPOSITION = (
    "30.visite_cessation_exposition"
)
VISITE_CESSATION_EXPOSITION_COURS_CARRIERE class-attribute instance-attribute 
VISITE_CESSATION_EXPOSITION_COURS_CARRIERE = (
    "31.visite_cessation_exposition_cours_carriere"
)
VISITE_CESSATION_EXPOSITION_COURS_CARRIERE_INITIATIVE_EMPLOYEUR class-attribute instance-attribute 
VISITE_CESSATION_EXPOSITION_COURS_CARRIERE_INITIATIVE_EMPLOYEUR = "32.visite_cessation_exposition_cours_carriere_initiative_employeur"
VISITE_CESSATION_EXPOSITION_COURS_CARRIERE_INITIATIVE_TRAVAILLEUR class-attribute instance-attribute 
VISITE_CESSATION_EXPOSITION_COURS_CARRIERE_INITIATIVE_TRAVAILLEUR = "33.visite_cessation_exposition_cours_carriere_initiative_travailleur"
VISITE_CESSATION_EXPOSITION_FIN_CARRIERE class-attribute instance-attribute 
VISITE_CESSATION_EXPOSITION_FIN_CARRIERE = (
    "34.visite_cessation_exposition_fin_carriere"
)
VISITE_CESSATION_EXPOSITION_FIN_CARRIERE_INITIATIVE_EMPLOYEUR class-attribute instance-attribute 
VISITE_CESSATION_EXPOSITION_FIN_CARRIERE_INITIATIVE_EMPLOYEUR = "35.visite_cessation_exposition_fin_carriere_initiative_employeur"
VISITE_CESSATION_EXPOSITION_FIN_CARRIERE_INITIATIVE_TRAVAILLEUR class-attribute instance-attribute 
VISITE_CESSATION_EXPOSITION_FIN_CARRIERE_INITIATIVE_TRAVAILLEUR = "36.visite_cessation_exposition_fin_carriere_initiative_travailleur"
VISITE_DEMANDE class-attribute instance-attribute 
VISITE_DEMANDE = '11.visite_demande'
VISITE_DEMANDE_EMPLOYEUR class-attribute instance-attribute 
VISITE_DEMANDE_EMPLOYEUR = '13.visite_demande_employeur'
VISITE_DEMANDE_MEDECIN_TRAVAIL class-attribute instance-attribute 
VISITE_DEMANDE_MEDECIN_TRAVAIL = (
    "14.visite_demande_medecin_travail"
)
VISITE_DEMANDE_MEDECIN_TRAVAIL_ORIENTATION_INFIRMIER class-attribute instance-attribute 
VISITE_DEMANDE_MEDECIN_TRAVAIL_ORIENTATION_INFIRMIER = "15.visite_demande_medecin_travail_orientation_infirmier"
VISITE_DEMANDE_TRAVAILLEUR class-attribute instance-attribute 
VISITE_DEMANDE_TRAVAILLEUR = "12.visite_demande_travailleur"
VISITE_INFORMATION_PREVENTION_INITIALE class-attribute instance-attribute 
VISITE_INFORMATION_PREVENTION_INITIALE = (
    "1.visite_information_prevention_initiale"
)
VISITE_INFORMATION_PREVENTION_PERIODIQUE class-attribute instance-attribute 
VISITE_INFORMATION_PREVENTION_PERIODIQUE = (
    "2.visite_information_prevention_periodique"
)
VISITE_INTERMEDIAIRE_ENTRE_DEUX_EXAMENS class-attribute instance-attribute 
VISITE_INTERMEDIAIRE_ENTRE_DEUX_EXAMENS = (
    "5.visite_intermediaire_entre_deux_examens"
)
VISITE_MI_CARRIERE class-attribute instance-attribute 
VISITE_MI_CARRIERE = '29.visite_mi_carriere'
VISITE_PREREPRISE class-attribute instance-attribute 
VISITE_PREREPRISE = '6.visite_prereprise'
VISITE_PREREPRISE_INITIATIVE_MEDECIN_CONSEIL class-attribute instance-attribute 
VISITE_PREREPRISE_INITIATIVE_MEDECIN_CONSEIL = (
    "7.visite_prereprise_initiative_medecin_conseil"
)
VISITE_PREREPRISE_INITIATIVE_MEDECIN_TRAITANT class-attribute instance-attribute 
VISITE_PREREPRISE_INITIATIVE_MEDECIN_TRAITANT = (
    "8.visite_prereprise_initiative_medecin_traitant"
)
VISITE_PREREPRISE_INITIATIVE_MEDECIN_TRAVAIL class-attribute instance-attribute 
VISITE_PREREPRISE_INITIATIVE_MEDECIN_TRAVAIL = (
    "10.visite_prereprise_initiative_medecin_travail"
)
VISITE_PREREPRISE_INITIATIVE_TRAVAILLEUR class-attribute instance-attribute 
VISITE_PREREPRISE_INITIATIVE_TRAVAILLEUR = (
    "9.visite_prereprise_initiative_travailleur"
)
VISITE_REPRISE class-attribute instance-attribute 
VISITE_REPRISE = '16.visite_reprise'
VISITE_REPRISE_ABSENCE_30_JOURS_ACCIDENT_TRAVAIL class-attribute instance-attribute 
VISITE_REPRISE_ABSENCE_30_JOURS_ACCIDENT_TRAVAIL = (
    "20.visite_reprise_absence_30_jours_accident_travail"
)
VISITE_REPRISE_ABSENCE_30_JOURS_ACCIDENT_TRAVAIL_INITIATIVE_EMPLOYEUR class-attribute instance-attribute 
VISITE_REPRISE_ABSENCE_30_JOURS_ACCIDENT_TRAVAIL_INITIATIVE_EMPLOYEUR = "21.visite_reprise_absence_30_jours_accident_travail_initiative_employeur"
VISITE_REPRISE_ABSENCE_30_JOURS_ACCIDENT_TRAVAIL_INITIATIVE_TRAVAILLEUR class-attribute instance-attribute 
VISITE_REPRISE_ABSENCE_30_JOURS_ACCIDENT_TRAVAIL_INITIATIVE_TRAVAILLEUR = "22.visite_reprise_absence_30_jours_accident_travail_initiative_travailleur"
VISITE_REPRISE_ABSENCE_MALADIE_PROFESSIONNELLE class-attribute instance-attribute 
VISITE_REPRISE_ABSENCE_MALADIE_PROFESSIONNELLE = (
    "23.visite_reprise_absence_maladie_professionnelle"
)
VISITE_REPRISE_ABSENCE_MALADIE_PROFESSIONNELLE_INITIATIVE_EMPLOYEUR class-attribute instance-attribute 
VISITE_REPRISE_ABSENCE_MALADIE_PROFESSIONNELLE_INITIATIVE_EMPLOYEUR = "24.visite_reprise_absence_maladie_professionnelle_initiative_employeur"
VISITE_REPRISE_ABSENCE_MALADIE_PROFESSIONNELLE_INITIATIVE_TRAVAILLEUR class-attribute instance-attribute 
VISITE_REPRISE_ABSENCE_MALADIE_PROFESSIONNELLE_INITIATIVE_TRAVAILLEUR = "25.visite_reprise_absence_maladie_professionnelle_initiative_travailleur"
VISITE_REPRISE_CAUSE_MALADIE_ACCIDENT_NON_PRO class-attribute instance-attribute 
VISITE_REPRISE_CAUSE_MALADIE_ACCIDENT_NON_PRO = (
    "17.visite_reprise_cause_maladie_accident_non_pro"
)
VISITE_REPRISE_CAUSE_MALADIE_ACCIDENT_NON_PRO_INITIATIVE_EMPLOYEUR class-attribute instance-attribute 
VISITE_REPRISE_CAUSE_MALADIE_ACCIDENT_NON_PRO_INITIATIVE_EMPLOYEUR = "18.visite_reprise_cause_maladie_accident_non_pro_initiative_employeur"
VISITE_REPRISE_CAUSE_MALADIE_ACCIDENT_NON_PRO_INITIATIVE_TRAVAILLEUR class-attribute instance-attribute 
VISITE_REPRISE_CAUSE_MALADIE_ACCIDENT_NON_PRO_INITIATIVE_TRAVAILLEUR = "19.visite_reprise_cause_maladie_accident_non_pro_initiative_travailleur"
VISITE_REPRISE_CONGE_MATERNITE class-attribute instance-attribute 
VISITE_REPRISE_CONGE_MATERNITE = (
    "26.visite_reprise_conge_maternite"
)
VISITE_REPRISE_CONGE_MATERNITE_INITIATIVE_EMPLOYEUR class-attribute instance-attribute 
VISITE_REPRISE_CONGE_MATERNITE_INITIATIVE_EMPLOYEUR = (
    "27.visite_reprise_conge_maternite_initiative_employeur"
)
VISITE_REPRISE_CONGE_MATERNITE_INITIATIVE_TRAVAILLEUR class-attribute instance-attribute 
VISITE_REPRISE_CONGE_MATERNITE_INITIATIVE_TRAVAILLEUR = "28.visite_reprise_conge_maternite_initiative_travailleur"
detect_from_string classmethod 
detect_from_string(visit_type_str)

Detect the visit type from a string.

Source code in components/occupational_health/internal/enums/visit_type.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
@classmethod
def detect_from_string(cls, visit_type_str: str) -> "VisitType":
    """
    Detect the visit type from a string.
    """
    try:
        return VisitType(visit_type_str)
    except ValueError:
        pass

    mapping = {
        "Visite d'embauche": VisitType.VISITE_INFORMATION_PREVENTION_INITIALE,
        "VIP": VisitType.VISITE_INFORMATION_PREVENTION_INITIALE,
        "Visite périodique": VisitType.VISITE_INFORMATION_PREVENTION_PERIODIQUE,
        "Visite de reprise": VisitType.VISITE_REPRISE,
        "Visite à la demande": VisitType.VISITE_DEMANDE,
        # not 100% sure about the original meaning of this one, but limited number of cases
        "Visite de suivi": VisitType.VISITE_DEMANDE_MEDECIN_TRAVAIL,
    }

    if visit_type_str not in mapping:
        raise ValueError(f"Visit type not found in mapping: {visit_type_str}")

    return mapping[visit_type_str]
level_1_category property 
level_1_category

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.

InvalidInputDataError

Bases: ValueError

The input data is not valid.

get_accounts_list_for_marmot 

get_accounts_list_for_marmot()

Return the list of accounts with occupational health contracts for Marmot.

Source code in components/occupational_health/public/marmot/queries.py 
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
@tracer_wrap()
def get_accounts_list_for_marmot() -> list[AccountForMarmot]:
    """
    Return the list of accounts with occupational health contracts for Marmot.
    """
    from components.occupational_health.internal.business_logic.contracting.queries.contracting import (
        get_all_account_ids_with_contract,
    )
    from components.occupational_health.public.dependencies import (
        get_app_dependency,
    )

    dependency = get_app_dependency()

    accounts = [
        AccountForMarmot(
            id=account_id,
            name=dependency.get_account_name(account_id=account_id),
        )
        for account_id in get_all_account_ids_with_contract()
    ]

    # Sort by name for consistent ordering
    return sorted(accounts, key=lambda a: a.name)

get_admin_accounts_from_user_for_marmot 

get_admin_accounts_from_user_for_marmot(user_id)

Returns the list of accounts for the current admin user

Source code in components/occupational_health/public/marmot/queries.py 
678
679
680
681
682
683
684
def get_admin_accounts_from_user_for_marmot(
    user_id: str,
) -> list[OccupationalHealthDashboardAdmin]:
    """
    Returns the list of accounts for the current admin user
    """
    return get_accounts_for_admin(UserId(user_id))

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 
 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
@tracer_wrap()
def get_affiliated_members_for_marmot(
    filter_by_account_id: AccountId | 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(AccountId(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 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
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
@tracer_wrap()
@inject_profile_service
def get_affiliation_decisions_for_marmot(
    profile_service: ProfileService,
    account_id: AccountId,
    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_all_admins_for_account 

get_all_admins_for_account(account_id, profile_service)

Get all admins for a given account.

Source code in components/occupational_health/internal/business_logic/queries/admin_dashboard/admin_authorization.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
@inject_profile_service
def get_all_admins_for_account(
    account_id: AccountId,
    profile_service: ProfileService,
) -> list[DashboardAdmin]:
    """
    Get all admins for a given account.
    """
    stmt = select(OccupationalHealthDashboardAdmin).filter(
        OccupationalHealthDashboardAdmin.account_id == account_id,
        OccupationalHealthDashboardAdmin.scope
        == OccupationalHealthDashboardAdminScope.ACCOUNT,
    )
    result = current_session.execute(stmt).scalars().all()

    user_ids = {admin.user_id for admin in result}
    profiles = profile_service.user_compat.get_user_profile_mapping(user_ids)
    profile_by_user_id = {
        str(user_id): profile for user_id, profile in profiles.user_to_profile.items()
    }

    return [
        DashboardAdmin(
            admin_id=admin.id,
            user_id=admin.user_id,
            # Will break if a user_id was merged into another one and no longer exists
            first_name=profile_by_user_id[admin.user_id].first_name,
            last_name=profile_by_user_id[admin.user_id].last_name,
            account_id=admin.account_id,
        )
        for admin in result
    ]

get_companies_with_affiliated_count_for_marmot 

get_companies_with_affiliated_count_for_marmot(account_id)

Return the list of all companies in an account with their count of affiliated members.

Source code in components/occupational_health/public/marmot/queries.py 
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
@tracer_wrap()
def get_companies_with_affiliated_count_for_marmot(
    account_id: AccountId | UUID,
) -> list[CompanyWithAffiliatedCount]:
    """
    Return the list of all companies in an account with their count of affiliated members.
    """
    from components.occupational_health.internal.business_logic.queries.affiliation.affiliations import (
        get_all_affiliations_for_account,
    )
    from components.occupational_health.internal.business_logic.queries.employments import (
        get_employments_by_profile_id,
    )

    account_id = AccountId(account_id)

    # Get all companies in the account
    company_ids = get_company_ids_in_account(account_id)
    company_names = get_company_names_by_ids(company_ids=set(company_ids))

    # Get all affiliations for this account (including terminated, upcoming, but not cancelled ones).
    affiliations = get_all_affiliations_for_account(account_id=account_id)

    # Fetch employments
    profile_ids = {affiliation.profile_id for affiliation in affiliations}
    employments_by_profile_id = get_employments_by_profile_id(
        account_id=account_id,
        profile_ids=profile_ids,
        should_overlap_prevenir_contract=False,
    )

    # Count unique affiliated members per company and collect SIRETs
    company_affiliated_member_ids: dict[CompanyId, set[ProfileId]] = {}
    # Map: company_id -> siret -> set of profile_ids
    profile_ids_by_siret_by_company_id: dict[CompanyId, dict[str, set[ProfileId]]] = {}
    dependency = get_app_dependency()

    siren_per_company = {
        company_id: dependency.get_company_siren(company_id)
        for company_id in company_ids
    }
    siret_per_company = {
        company_id: dependency.get_company_siret(company_id)
        for company_id in company_ids
    }

    nic_per_employment_id = get_nic_for_employments(
        on_date_per_employment_id={
            employment.employment_id: clamp(
                utctoday(), employment.start_date, employment.end_date
            )
            for employments in employments_by_profile_id.values()
            for employment in employments
        }
    )

    for profile_id, employments in employments_by_profile_id.items():
        for employment in employments:
            company_id = CompanyId(employment.company_id)
            if company_id not in company_ids:
                raise RuntimeError(
                    f"Encountered employment from out-of-account company:"
                    f" {employment.employment_id} has company ID {company_id} which is not in account {company_ids}."
                )

            # Count members per company
            if company_id not in company_affiliated_member_ids:
                company_affiliated_member_ids[company_id] = set()
            company_affiliated_member_ids[company_id].add(profile_id)

            nic = nic_per_employment_id[employment.employment_id]
            if not nic:
                # Trust first the Employment Component, then the Company configuration
                siret = siret_per_company[company_id]
                if siret:
                    nic = siret[-5:]

            if nic:
                siren = siren_per_company[company_id]
                if siren:
                    siret = f"{siren}{nic}"

                    if company_id not in profile_ids_by_siret_by_company_id:
                        profile_ids_by_siret_by_company_id[company_id] = defaultdict(
                            set
                        )
                    profile_ids_by_siret_by_company_id[company_id][siret].add(
                        profile_id
                    )

    # Build the result list with all companies (even those with 0 affiliated members)
    result = []
    for company_id in company_ids:
        count = len(company_affiliated_member_ids.get(company_id, set()))
        profile_ids_by_siret = profile_ids_by_siret_by_company_id.get(company_id, {})

        company_data_per_siret = {
            siret: dependency.get_company_from_siret(siret=siret)
            for siret in profile_ids_by_siret.keys()
            if siret
        }

        # Fetch company names from DSN for each SIRET
        sirets = []
        for siret, profile_ids in profile_ids_by_siret.items():
            pretty_name = None
            company_data = company_data_per_siret.get(siret)
            if company_data and company_data.name:
                pretty_name = (
                    f"{company_data.name} ({company_data.city})"
                    if company_data.city
                    else company_data.name
                )

            sirets.append(
                SiretWithMemberCount(
                    siret=siret,
                    member_count=len(profile_ids),
                    pretty_name=pretty_name,
                )
            )

        # Sort SIRETs by member count descending, then by SIRET
        sirets.sort(key=lambda s: (-s.member_count, s.siret))

        result.append(
            CompanyWithAffiliatedCount(
                company_id=company_id,
                company_display_name=company_names.get(
                    company_id, f"Company {company_id}"
                ),
                ever_affiliated_members_count=count,
                sirets=sirets,
            )
        )

    # Sort by company display name for consistent ordering
    return sorted(result, key=lambda c: c.company_display_name)

get_contract_info_for_account 

get_contract_info_for_account(account_id)

Get contract information for an account.

Source code in components/occupational_health/internal/business_logic/queries/affiliation/affiliation_tool.py 
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
def get_contract_info_for_account(account_id: AccountId) -> ContractInfoResult:
    """Get contract information for an account."""
    try:
        account_name = get_account_name(account_id)
    except NoResultFound:
        raise BaseErrorCode.missing_resource(f"Account {account_id} not found")
    timeline = get_contract_timeline_for_account(account_id)

    contracts: list[ContractInfo] = []
    if timeline:
        for version in timeline:
            contracts.append(
                ContractInfo(
                    account_id=account_id,
                    start_date=version.validity_period.start_date,
                    end_date=version.validity_period.end_date,
                )
            )

    return ContractInfoResult(
        account_name=account_name,
        contracts=contracts,
    )

get_occupational_health_profile_data_for_marmot 

get_occupational_health_profile_data_for_marmot(
    profile_id, account_id=None
)

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

Source code in components/occupational_health/public/marmot/queries.py 
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
@tracer_wrap()
def get_occupational_health_profile_data_for_marmot(
    profile_id: ProfileId | UUID,
    account_id: AccountId | None = None,
) -> OccupationalHealthProfileData:
    """
    Return all occupational health data for a profile, for display in Marmot.
    """
    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.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

    profile_id = ProfileId(profile_id)

    affiliations = get_all_affiliations(profile_id, account_id)
    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 
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
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
324
325
326
327
328
329
330
@inject_profile_service
def get_recent_affiliation_movements(
    profile_service: ProfileService,
    account_id: AccountId | UUID,
    not_older_than: date,
) -> list[AffiliationMovementForMarmot]:
    """
    Return the list of affiliation decisions made since the given date for the given account.
    """
    account_id = AccountId(account_id)

    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

parse_and_match_spreadsheet 

parse_and_match_spreadsheet(
    profile_service,
    account_id,
    spreadsheet_data,
    raise_on_invalid_data=False,
)

Parse spreadsheet data and return matching results (members that will could affiliate)

Source code in components/occupational_health/internal/business_logic/queries/affiliation/affiliation_tool.py 
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
@inject_profile_service
@tracer_wrap()
def parse_and_match_spreadsheet(
    profile_service: ProfileService,
    account_id: AccountId,
    spreadsheet_data: str,
    raise_on_invalid_data: bool = False,
) -> ParsingAndMatchingResult:
    """
    Parse spreadsheet data and return matching results (members that will could affiliate)
    """
    from components.fr.internal.fr_employment_data_sources.business_logic.queries.user import (  # noqa: ALN043 - function should be made public (or existing global one made compatible)
        get_user_from_personal_informations,
    )
    from components.fr.internal.fr_employment_data_sources.entities.employee_personal_informations import (  # noqa: ALN043 - function should be made public (or existing global one made compatible)
        EmployeePersonalInformations,
    )

    parsing_warnings = []

    rows = parse_spreadsheet_data(spreadsheet_data)

    new_warnings = _sanitize_ssns(
        rows,
        raise_on_invalid_or_duplicate_ssn=raise_on_invalid_data,
    )
    parsing_warnings.extend(new_warnings)

    # Raise on duplicate proposed user IDs
    _sanitize_data(rows)

    # Get all affiliations for the account to check if users are already affiliated
    all_affiliations = get_all_affiliations_for_account(account_id)

    # Build a set of affiliated user IDs
    affiliations_by_user_ids: dict[UserId, Affiliation] = {}
    for affiliation in all_affiliations:
        # TODO: improve performance (N+1 query)
        user_id = get_user_id_from_occupational_health_profile_id(
            affiliation.profile_id
        )
        affiliations_by_user_ids[user_id] = affiliation

    # Find the contract start date
    contract_timeline = get_contract_timeline_for_account(account_id)
    if not contract_timeline:
        raise ValueError(f"Couldn't find any contract for account {account_id}")
    default_start_date = contract_timeline.start_date

    # Collect user IDs for employment lookup
    user_ids_for_employment: set[UserId] = set()

    # Detect the date format for the entire file
    all_birthdates: set[str] = set()
    all_visit_dates: set[str] = set()
    for row in rows:
        # Collect birthdates
        if row.get("date_naissance_gsheet"):
            all_birthdates.add(row["date_naissance_gsheet"])
        # Collect last visit dates
        if row.get("date_de_derniere_visite"):
            all_visit_dates.add(row["date_de_derniere_visite"].strip())

    birthdate_format = detect_date_format(all_birthdates)
    visit_date_format = detect_date_format(all_visit_dates)

    matches: list[MemberToAffiliate] = []

    for row_index, row in enumerate(rows):
        # TODO: support aliases for column names
        first_name = row.get("prenom_gsheet", "").strip()
        last_name = row.get("nom_gsheet", "").strip()
        birthdate_str = row.get("date_naissance_gsheet")
        birthdate = (
            datetime.strptime(birthdate_str, birthdate_format).date()
            if birthdate_str
            else None
        )
        ssn = row.get("ssn_original", "").strip()
        email_pro_gsheet_raw = row.get("email_pro_gsheet", "").strip()
        email_perso_gsheet_raw = row.get("email_perso_gsheet", "").strip()

        # Validate and normalize professional email
        email_pro_gsheet, email_pro_warning = _validate_email(
            row_index=row_index,
            email=email_pro_gsheet_raw,
            field_name="email_pro_gsheet",
        )
        if email_pro_warning is not None:
            parsing_warnings.append(email_pro_warning)

        # Validate and normalize personal email
        email_perso_gsheet, email_perso_warning = _validate_email(
            row_index=row_index,
            email=email_perso_gsheet_raw,
            field_name="email_perso_gsheet",
        )
        if email_perso_warning is not None:
            parsing_warnings.append(email_perso_warning)

        # Parse risk category from 'suivi_si_sia_sir' column
        risk_category, risk_category_warning = parse_risk_category(
            row_index=row_index,
            risk_category_str=row.get("suivi_si_sia_sir"),
        )
        if risk_category_warning is not None:
            parsing_warnings.append(risk_category_warning)

        # Parse last visit information
        last_visit, warning = _parse_last_visit(
            row_index=row_index,
            last_visit_date_str=row.get("date_de_derniere_visite", "").strip(),
            last_visit_type_str=row.get("type_de_visite", "").strip(),
            visit_date_format=visit_date_format,
        )
        if warning is not None:
            parsing_warnings.append(warning)

        # Create EmployeePersonalInformations for matching
        personal_informations = EmployeePersonalInformations(
            first_name=first_name,
            known_last_names={
                # we could inject the birth last name if available
                last_name
            },
            email=email_pro_gsheet,
            ssn=ssn.strip(),
            birth_date=birthdate,
            auto_compute_ssn_key=True,  # Auto-compute SSN key if needed
            # TODO: consider using siren / company external IDs when available
        )

        # Match user (company is optional)
        # NOTE: might need to move this back into the fr component / component injection until it gets globalized
        matching_result = get_user_from_personal_informations(
            personal_informations=personal_informations,
            company=None,
        )

        # Get matched user data if found
        if matching_result.matched_user_id is not None:
            matched_global_profile_id = (
                profile_service.user_compat.get_profile_id_by_user_id(
                    user_id=matching_result.matched_user_id
                )
            )
            if matched_global_profile_id is None:
                raise RuntimeError(
                    f"User ID {matching_result.matched_user_id} has no global profile"
                )

            matched_profile = profile_service.get_profile(
                profile_id=matched_global_profile_id
            )
        else:
            matched_profile = None

        proposed_user_id = row.get("alan_user_id", "").strip()
        proposed_user_name = None
        proposed_birthdate = None
        if proposed_user_id:
            if (
                matching_result.matched_user_id
                and matched_profile
                and str(matching_result.matched_user_id) == proposed_user_id
            ):
                # Everybody agrees, let's reuse the same name
                proposed_user_name = matched_profile.full_name
                proposed_birthdate = matched_profile.birth_date
            else:
                proposed_global_profile_id = (
                    profile_service.user_compat.get_profile_id_by_user_id(
                        user_id=UserId(proposed_user_id)
                    )
                )
                if proposed_global_profile_id is None:
                    raise RuntimeError(
                        f"User ID {proposed_user_id} has no global profile"
                    )

                proposed_profile = profile_service.get_profile(
                    profile_id=proposed_global_profile_id
                )
                if proposed_profile:
                    proposed_user_name = proposed_profile.full_name
                    proposed_birthdate = proposed_profile.birth_date

        relevant_user_ids = {
            UserId(str(matching_result.matched_user_id))
            if matching_result.matched_user_id
            else None,
            UserId(proposed_user_id),
        }
        relevant_user_ids -= {None}
        relevant_affiliations = []
        for relevant_user_id in relevant_user_ids:
            if relevant_user_id in affiliations_by_user_ids:
                affiliation = affiliations_by_user_ids[relevant_user_id]
                relevant_affiliations.append(
                    RelevantAffiliation(
                        user_id=relevant_user_id,
                        start_date=affiliation.start_date,
                        end_date=affiliation.end_date,
                    )
                )

        user_ids_for_employment.update(relevant_user_ids)  # type: ignore[arg-type]

        extra_data_from_row = row.copy()
        # No need to store some of the columns
        extra_data_from_row.pop("marmot_user_link", None)
        extra_data_from_row.pop("marmot_account_link", None)
        extra_data_from_row.pop("marmot_company_link", None)

        matches.append(
            MemberToAffiliate(
                row_index=row_index,
                first_name=first_name,
                last_name=last_name,
                birthdate=birthdate,
                ssn=ssn,
                professional_email=email_pro_gsheet or None,
                personal_email=email_perso_gsheet or None,
                # return None instead of an empty string
                proposed_user_id=proposed_user_id or None,
                proposed_user_name=proposed_user_name,
                proposed_birthdate=proposed_birthdate,
                matching_result=matching_result,
                matched_name=matched_profile.full_name if matched_profile else None,
                matched_birthdate=(
                    matched_profile.birth_date if matched_profile else None
                ),
                affiliations=relevant_affiliations,
                employments=[],  # Will be filled below
                # TODO: clamp with employment date
                target_start_date=default_start_date,
                last_visit=last_visit,
                risk_category=risk_category,
                extra_data=extra_data_from_row,
            )
        )

    # Get employments for all matched users and inject them into matches
    if user_ids_for_employment:
        employments_by_user_id = get_employments_by_user_id(
            account_ids={account_id},
            user_ids=user_ids_for_employment,
            should_overlap_prevenir_contract=False,
        )

        # Build company names
        company_ids = set()
        for employments in employments_by_user_id.values():
            for employment in employments:
                company_ids.add(CompanyId(employment.company_id))
        company_names = get_company_names_by_ids(company_ids)

        # Fill employments in matches
        for match in matches:
            match_user_id: str | int | None = (
                match.matching_result.matched_user_id or match.proposed_user_id
            )
            if match_user_id is None:
                continue
            user_employments = employments_by_user_id.get(
                UserId(str(match_user_id)), []
            )
            match.employments = [
                EmploymentInfo(
                    company_id=employment.company_id,
                    company_name=company_names.get(
                        CompanyId(employment.company_id),
                        "Unknown!",
                    ),
                    start_date=employment.start_date,
                    end_date=employment.end_date,
                )
                for employment in user_employments
            ]

    # Add warning if we suggest the same user ID for multiple users
    parsing_warnings.extend(_check_for_duplicate_matching_user_ids(matches))

    return ParsingAndMatchingResult(
        matches=matches,
        warnings=parsing_warnings,
    )

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:

Name Type Description
tuple tuple[list[InvoiceToGenerateData], dict[UUID, dict[str, int | str]]]

A tuple containing: - list[InvoiceToGenerateData]: The data about all invoices to generate for the given period. - dict[UUID, dict[str, int | str]]: Statistics per account with keys "account_name", "invoices_count" and "errors_count".
Source code in components/occupational_health/public/queries/billing.py 
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
def build_invoices_data_for_period(
    billing_period: OccupationalHealthBillingPeriod,
    billing_year: int,
) -> tuple[list[InvoiceToGenerateData], dict[UUID, dict[str, int | str]]]:
    """
    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:
        tuple: A tuple containing:
            - list[InvoiceToGenerateData]: The data about all invoices to generate for the given period.
            - dict[UUID, dict[str, int | str]]: Statistics per account with keys "account_name", "invoices_count" and "errors_count".
    """
    from components.occupational_health.internal.business_logic.contracting.queries.contracting import (
        get_all_account_ids_with_contract,
    )

    with read_only_db_context():
        # Get all customers with an active contract during the given billing period
        account_ids = get_all_account_ids_with_contract(
            active_during_period=billing_period.to_validity_period(year=billing_year),
        )

        dependency = get_app_dependency()
        invoice_to_generate_data_list: list[InvoiceToGenerateData] = []
        stats_per_account: dict[UUID, dict[str, int | str]] = {}

        for account_id in account_ids:
            invoices, errors = build_invoices_data_for_period_and_account(
                billing_period=billing_period,
                billing_year=billing_year,
                account_id=account_id,
            )
            invoice_to_generate_data_list.extend(invoices)
            stats_per_account[account_id] = {
                "account_name": dependency.get_account_name(account_id=account_id),
                "invoices_count": len(invoices),
                "errors_count": len(errors),
            }

        return invoice_to_generate_data_list, stats_per_account

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 | AccountId

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:

Name Type Description
tuple tuple[list[InvoiceToGenerateData], dict[UUID, str]]

A tuple containing: - list[InvoiceToGenerateData]: The data about all invoices to generate for the given period. - dict[UUID, str]: Mapping of affiliation IDs to rejection reasons for affiliations that couldn't be matched.
Source code in components/occupational_health/public/queries/billing.py 
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
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
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
324
325
def build_invoices_data_for_period_and_account(
    billing_period: OccupationalHealthBillingPeriod,
    billing_year: int,
    account_id: UUID | AccountId,
) -> tuple[list[InvoiceToGenerateData], dict[UUID, str]]:
    """
    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:
        tuple: A tuple containing:
            - list[InvoiceToGenerateData]: The data about all invoices to generate for the given period.
            - dict[UUID, str]: Mapping of affiliation IDs to rejection reasons for affiliations that couldn't be matched.
    """
    from components.occupational_health.internal.business_logic.billing.pricing import (
        YEARLY_PRICE_PER_EMPLOYEE_IN_CENTS,
    )
    from components.occupational_health.internal.business_logic.queries.affiliation.affiliations import (
        get_all_affiliations_for_account,
    )
    from components.occupational_health.internal.helpers.affiliation import (
        build_profile_id_from_affiliable_ref,
    )

    account_id = AccountId(account_id)

    # 1. Get the customer data, including the billable entities.
    customers = get_billed_entities_for_account(account_id=account_id)

    # 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
    )

    # Sort affiliations by start_date to process them in chronological order
    affiliations_sorted = sorted(
        affiliations_for_whole_account, key=lambda aff: aff.start_date
    )

    # Compute the start and end dates, between which we'll consider period to bill
    to_bill_period = billing_period.to_validity_period(year=billing_year)

    customer_by_contract_ref = {
        customer.contract_ref: customer for customer in customers
    }
    # Index customers by profile_id to detect duplicates
    customers_per_profile: defaultdict[ProfileId, list[BilledEntityData]] = defaultdict(
        list
    )
    errors = 0
    error_affiliation_ids: dict[UUID, str] = {}
    for affiliation in affiliations_sorted:
        if affiliation.is_ever_active_between(
            period_start=to_bill_period.start_date,
            period_end=to_bill_period.end_date,
        ):
            # Calculate the number of days the affiliation is active during the billing period
            affiliation_duration = _find_duration_of_affiliation_on_period(
                affiliation, to_bill_period, billing_period
            )
            # Skip affiliations that are active for 3 days or less
            if affiliation_duration and affiliation_duration <= 3:
                current_logger.info(
                    f"Skipping affiliation {affiliation.affiliation_id} - only active for {affiliation_duration} days"
                )
                continue
            # This affiliation is active at some point during the billing period:
            # take it into account for billing.
            try:
                profile_id = build_profile_id_from_affiliable_ref(
                    affiliation.affiliable_ref
                )
                # Find the right customer for this affiliation
                matching_customer = _find_customer_to_bill_for_affiliation(
                    account_id=account_id,
                    affiliation=affiliation,
                    profile_id=profile_id,
                    customers=customers,
                    to_bill_period=to_bill_period,
                )
            except UnableToDetermineCustomerToBill as e:
                # TODO: @mickael.bergem 2026-06-01 make it an logger.exception again, hopefully we fixed the root cause?
                current_logger.warning(
                    f"Unable to determine customer to bill for {affiliation}"
                )
                errors += 1
                error_affiliation_ids[affiliation.affiliation_id] = str(e)
                # Log and skip for now...
            else:
                # Collect customer for each profile_id
                customers_per_profile[profile_id].append(matching_customer)

    # 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. Process customers per profile and select the final customer to bill
    profile_ids_per_customer_ref: defaultdict[str, set[str]] = defaultdict(set)

    for profile_id, customer_list in customers_per_profile.items():
        uniq_contract_ref = {customer.contract_ref for customer in customer_list}
        if len(uniq_contract_ref) > 1:
            current_logger.info(
                f"Multiple customers found for profile<{profile_id}> : {', '.join(uniq_contract_ref)}, taking the first one"
            )

        # Take the first customer (from earliest affiliation due to sorting)
        selected_customer = customer_list[0]
        profile_ids_per_customer_ref[selected_customer.contract_ref].add(
            str(profile_id)
        )

    # 4. Package the data
    invoice_data = []
    for customer_contract_ref, profile_ids in profile_ids_per_customer_ref.items():
        customer = customer_by_contract_ref[customer_contract_ref]
        invoice_data.append(
            InvoiceToGenerateData(
                billing_period=billing_period,
                billing_year=billing_year,
                contract_type="occupational_health",
                contract_ref=customer_contract_ref,
                entity_name=customer.entity_name,
                siren=customer.siren,
                siret=customer.siret,
                contract_yearly_price_per_employee_in_cents=YEARLY_PRICE_PER_EMPLOYEE_IN_CENTS,
                contract_needs_installment_plan_for_employees_count=(
                    customer.number_of_employees_per_year_for_installment_plan or {}
                ).get(str(billing_year)),
                references_of_employees_affiliated_over_the_year=profile_ids,
            )
        )

    return invoice_data, error_affiliation_ids

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 
521
522
523
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
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
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_billed_entities_for_account 

get_billed_entities_for_account(account_id)

Get all billed entities for a given account.

Source code in components/occupational_health/public/queries/billing.py 
 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
def get_billed_entities_for_account(
    account_id: UUID,
) -> list["BilledEntityData"]:
    """
    Get all billed entities for a given account.
    """
    query = select(OccupationalHealthBilledEntity).where(
        OccupationalHealthBilledEntity.account_id == AccountId(account_id)
    )
    billed_entities: Sequence[OccupationalHealthBilledEntity] = (
        current_session.execute(query).scalars().all()
    )

    serialized_billed_entities: list[BilledEntityData] = []

    for billed_entity in billed_entities:
        serialized_billed_entities.append(
            BilledEntityData(
                id=billed_entity.id,
                entity_name=billed_entity.entity_name,
                siren=billed_entity.siren,
                siret=billed_entity.siret,
                postal_address=CustomerAddress(
                    street=billed_entity.postal_street,
                    postal_code=billed_entity.postal_code,
                    city=billed_entity.postal_city,
                ),
                account_id=billed_entity.account_id,
                number_of_employees_per_year_for_installment_plan=billed_entity.number_of_employees_per_year_for_installment_plan,
                contract_ref=billed_entity.contract_ref,
                subscription_ref=billed_entity.subscription_ref,
                emails_to_notify=billed_entity.emails_to_notify,
            )
        )

    return serialized_billed_entities

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 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_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
    """
    query = select(OccupationalHealthBilledEntity).where(
        OccupationalHealthBilledEntity.contract_ref == contract_ref  # type: ignore[comparison-overlap,arg-type]
    )

    with read_only_db_context():
        billed_entity = current_session.scalars(query).one_or_none()

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

    return CustomerData(
        entity_name=billed_entity.entity_name,
        siren=billed_entity.siren,
        siret=billed_entity.siret,
        postal_address=CustomerAddress(
            street=billed_entity.postal_street,
            postal_code=billed_entity.postal_code,
            city=billed_entity.postal_city,
        ),
    )

get_profile_list_from_invoice_data 

get_profile_list_from_invoice_data(invoices)

Build a list of profiles with customer names from invoice data.

Parameters:

Name Type Description Default
invoices list[InvoiceToGenerateData]

List of invoice objects, each with contract_ref and references_of_employees_affiliated_over_the_year attributes.

 required

Returns:

Type Description
list[dict[str, str | ProfileId]]

List of dicts with profile_id, first_name, last_name, and customer_name.
Source code in components/occupational_health/public/queries/billing.py 
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
def get_profile_list_from_invoice_data(
    invoices: list[InvoiceToGenerateData],
) -> list[dict[str, str | ProfileId]]:
    """
    Build a list of profiles with customer names from invoice data.

    Args:
        invoices: List of invoice objects, each with contract_ref and references_of_employees_affiliated_over_the_year attributes.

    Returns:
        List of dicts with profile_id, first_name, last_name, and customer_name.
    """
    from uuid import UUID

    from components.global_profile.public.api import ProfileService
    from components.occupational_health.external.global_profile import (
        get_global_profile_ids_by_occupational_health_profile_id_mapping,
    )

    # Build profile list with customer names
    profile_to_customer_name: dict[ProfileId, str] = {}
    all_profile_ids: set[ProfileId] = set()
    for invoice in invoices:
        customer_data = get_customer_data(invoice.contract_ref)
        for profile_id_str in invoice.references_of_employees_affiliated_over_the_year:
            profile_id = ProfileId(UUID(profile_id_str))
            all_profile_ids.add(profile_id)
            profile_to_customer_name[profile_id] = customer_data.entity_name

    # Get global_profile_ids from occupational health profile_ids
    profile_id_to_global_profile_id = (
        get_global_profile_ids_by_occupational_health_profile_id_mapping(
            profile_ids=all_profile_ids
        )
    )

    # Get first_name/last_name from global profiles using global_profile_ids directly
    global_profile_ids = list(profile_id_to_global_profile_id.values())
    profile_service = ProfileService.create()
    global_profiles = profile_service.get_profiles(profile_ids=global_profile_ids)
    # Build lookup by global_profile_id for easier access
    global_profile_by_id = {profile.id: profile for profile in global_profiles}

    profiles_with_customer_name: list[dict[str, str | ProfileId]] = []
    for profile_id in all_profile_ids:
        global_profile_id = profile_id_to_global_profile_id[profile_id]
        profile = global_profile_by_id.get(global_profile_id)
        first_name = (profile.first_name or "⁉️") if profile else "⁉️"
        last_name = (profile.last_name or "⁉️") if profile else "⁉️"
        profiles_with_customer_name.append(
            {
                "profile_id": profile_id,
                "first_name": first_name,
                "last_name": last_name,
                "customer_name": profile_to_customer_name[profile_id],
            }
        )
    return profiles_with_customer_name

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 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
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.contracting.queries.helpers import (
        has_currently_active_contract as _has_active_contract,
    )

    return _has_active_contract(AccountId(account_id))

has_active_or_upcoming_occupational_health_contract 

has_active_or_upcoming_occupational_health_contract(
    account_id,
)

Return whether the given account ID has an active or upcoming occupational health contract.

Parameters:

Name Type Description Default
account_id UUID

The ID of the account to check.

 required

Returns: bool: True if the account has an active or upcoming occupational health contract, False otherwise.

Source code in components/occupational_health/public/queries/has_active_occupational_health_contract.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def has_active_or_upcoming_occupational_health_contract(account_id: UUID) -> bool:
    """
    Return whether the given account ID has an active or upcoming occupational health contract.

    Args:
        account_id: The ID of the account to check.
    Returns:
        bool: True if the account has an active or upcoming occupational health contract, False otherwise.
    """
    from components.occupational_health.internal.business_logic.contracting.queries.helpers import (
        has_currently_active_or_upcoming_contract as _has_active_or_upcoming_contract,
    )

    return _has_active_or_upcoming_contract(AccountId(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 
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
def is_currently_affiliated_to_occupational_health(
    global_profile_id: GlobalProfileId | UUID,
    account_id: AccountId | 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.
    """
    global_profile_id = GlobalProfileId(global_profile_id)
    account_id = AccountId(account_id)

    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

profiles

get_profile_dmst 

get_profile_dmst(
    occupational_health_profile_id, profile_service
)

Returns the member information for a given occupational health profile ID.

Source code in components/occupational_health/internal/business_logic/queries/profiles/profiles.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
 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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@inject_profile_service
def get_profile_dmst(
    occupational_health_profile_id: UUID, profile_service: ProfileService
) -> Dmst:
    """
    Returns the member information for a given occupational health profile ID.
    """

    occupational_health_profile = current_session.get_one(
        OccupationalHealthProfile,
        occupational_health_profile_id,
        options=[joinedload(OccupationalHealthProfile.employment_data)],
    )
    profile = profile_service.get_or_raise_profile(
        occupational_health_profile.global_profile_id
    )
    jobs = OccupationalHealthJobBroker.get_existing_jobs(
        occupational_health_profile_id
    ).all()
    administrative_profile = (
        OccupationalHealthAdministrativeProfileBroker.get_by_profile(
            ProfileId(occupational_health_profile_id)
        ).one_or_none()
    )

    # Temporary getting the SSN while we implement the INS clinic side
    user_id = get_app_dependency().get_user_id_from_global_profile_id(
        GlobalProfileId(profile.id)
    )
    ssn, _ = get_app_dependency().get_ssn_and_ntt_for_user(user_id)

    personal_email = occupational_health_profile.personal_email
    phone_number = occupational_health_profile.phone_number

    latest_employment_data = (
        occupational_health_profile.employment_data[-1]
        if occupational_health_profile.employment_data
        else None
    )
    professional_email = (
        latest_employment_data.professional_email if latest_employment_data else None
    )

    return Dmst(
        occupational_health_profile_id=occupational_health_profile.id,
        first_name=profile.first_name or "⁉️",
        last_name=profile.last_name or "⁉️",
        gender=profile.gender,
        birthdate=profile.birth_date,
        medical_secrecy_worker_record_id=occupational_health_profile.medical_secrecy_worker_record_id,
        curriculum_laboris=CurriculumLaboris(
            occupational_medical_history=OccupationalMedicalHistory(
                types=[],  # In medical secrecy record (see get_medical_record)
                description=None,  # In medical secrecy record (see get_medical_record)
            ),
            jobs=(
                [
                    MemberJob(
                        id=job.id,
                        title=job.title,
                        start_date=job.start_date,
                        end_date=job.end_date,
                        employer=job.employer,
                        # The last job (oldest) is a generic "past job" job
                        is_past_job=job.is_past_job,
                        medical_secrecy_worker_job_id=job.medical_secrecy_worker_job_id,
                        description=None,  # In medical secrecy worker job (see get_medical_record)
                        working_hours=None,  # In medical secrecy worker job (see get_medical_record)
                        missions=None,  # In medical secrecy worker job (see get_medical_record)
                        physical_conditions=None,  # In medical secrecy worker job (see get_medical_record)
                        organizational_conditions=None,  # In medical secrecy worker job (see get_medical_record)
                        mental_conditions=None,  # In medical secrecy worker job (see get_medical_record)
                        tools=None,  # In medical secrecy worker job (see get_medical_record)
                        worn_equipment=None,  # In medical secrecy worker job (see get_medical_record)
                        risks_and_advice=None,  # In medical secrecy worker job (see get_medical_record)
                        contract_type=None,  # In medical secrecy worker job (see get_medical_record)
                        exposition_notations=[],  # In medical secrecy worker job (see get_medical_record)
                    )
                    for (index, job) in enumerate(jobs)
                ]
            ),
        ),
        administrative_profile=(
            AdministrativeProfile(
                ssn=ssn,
                health_statuses=administrative_profile.health_statuses,
                risk_category=administrative_profile.risk_category,
                notes=administrative_profile.notes,
                medical_record_sharing_consent=administrative_profile.medical_record_sharing_consent,
                medical_record_access_consent=administrative_profile.medical_record_access_consent,
                health_data_sharing_consent=administrative_profile.health_data_sharing_consent,
                video_consultation_consent=administrative_profile.video_consultation_consent,
                personal_email=personal_email,
                professional_email=professional_email,
                phone_number=phone_number,
            )
            if administrative_profile
            else AdministrativeProfile(
                ssn=ssn,
                health_statuses=[],
                risk_category=RiskCategory.SI,
                notes=None,
                medical_record_sharing_consent=False,
                medical_record_access_consent=False,
                health_data_sharing_consent=False,
                video_consultation_consent=False,
                personal_email=personal_email,
                professional_email=professional_email,
                phone_number=phone_number,
            )
        ),
        health_history=HealthHistory(),  # In Medical secrecy Medical record (see get_medical_record)
    )

search

get_search_results 

get_search_results(profile_service, search_input)

Search for members and accounts based on provided search input.

Note

The function currently only searches for members by full name

Source code in components/occupational_health/internal/business_logic/queries/search/search.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
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
@inject_profile_service
def get_search_results(
    profile_service: ProfileService, search_input: str
) -> SearchResult:
    """
    Search for members and accounts based on provided search input.

    Note:
        The function currently only searches for members by full name
    """
    from components.occupational_health.public.dependencies import get_app_dependency

    dependency = get_app_dependency()
    # Searching for the global profiles and associated occupational health profiles
    global_profiles = profile_service.search_profiles_by_fullname(search_input)
    global_profiles_by_id: dict[GlobalProfileId, list[Profile]] = group_by(
        global_profiles, lambda profile: GlobalProfileId(profile.id)
    )
    global_profiles_ids = list(global_profiles_by_id.keys())
    profile_with_jobs = (
        get_occupational_health_profile_by_global_profile_ids_with_latest_job(
            global_profiles_ids
        )
    )
    occupational_health_profiles = [profile for profile, jobs in profile_with_jobs]
    occupational_health_profiles_to_latest_health_job_mapping = {
        profile.id: job
        for (
            profile,
            job,
        ) in profile_with_jobs
    }
    occupational_health_profile_ids_by_administrative_profiles = {
        occupational_health_profile.id: occupational_health_profile.administrative_profile
        for occupational_health_profile in occupational_health_profiles
    }

    occupational_health_profile_ids_by_global_profile_id = (
        get_occupational_health_profile_ids_by_global_profile_id_mapping(
            global_profiles_ids
        )
    )
    company_names_by_occupational_health_profile_ids = (
        get_company_names_by_occupational_health_profile_id_mapping(
            occupational_health_profiles
        )
    )
    global_profile_ids_by_occupational_health_profile_ids = {
        value: key
        for key, value in occupational_health_profile_ids_by_global_profile_id.items()
    }

    affiliated_account_ids = get_all_account_ids_with_active_contract()
    matched_accounts = dependency.search_accounts_by_name(
        search_input, affiliated_account_ids
    )

    def get_latest_job_name_by_occupational_health_profile_id(
        occupational_health_profile_id: ProfileId,
    ) -> Optional[str]:
        latest_job = occupational_health_profiles_to_latest_health_job_mapping[
            occupational_health_profile_id
        ]
        return latest_job.title if latest_job else None

    def get_profile_by_occupational_health_profile_id(
        occupational_health_profile_id: ProfileId,
    ) -> Profile:
        global_profile_id = global_profile_ids_by_occupational_health_profile_ids[
            occupational_health_profile_id
        ]
        return one(global_profiles_by_id[global_profile_id])

    def get_risk_category_by_occupational_health_profile_id(
        occupational_health_profile_id: ProfileId,
    ) -> Optional[RiskCategory]:
        administrative_profile = (
            occupational_health_profile_ids_by_administrative_profiles[
                occupational_health_profile_id
            ]
        )
        return administrative_profile.risk_category if administrative_profile else None

    def get_company_names_by_occupational_health_profile_id(
        occupational_health_profile_id: ProfileId,
    ) -> Optional[str]:
        names = company_names_by_occupational_health_profile_ids.get(
            occupational_health_profile_id, None
        )
        return ",".join(names) if names else None

    return SearchResult(
        members=[
            Member(
                occupational_health_profile_id=str(occupational_health_profile_id),
                first_name=get_profile_by_occupational_health_profile_id(
                    occupational_health_profile_id
                ).first_name,
                last_name=get_profile_by_occupational_health_profile_id(
                    occupational_health_profile_id
                ).last_name,
                job_title=get_latest_job_name_by_occupational_health_profile_id(
                    occupational_health_profile_id
                ),
                company_name=get_company_names_by_occupational_health_profile_id(
                    occupational_health_profile_id
                ),
                risk_category=get_risk_category_by_occupational_health_profile_id(
                    occupational_health_profile_id
                ),
            )
            for occupational_health_profile_id in occupational_health_profile_ids_by_global_profile_id.values()
        ],
        accounts=[
            Account(id=str(account.id), name=account.name)
            for account in matched_accounts
        ],
        companies=[],  # TODO @david.barthelemy Remove this when frontend is deployed
    )

strategy_rules

AffiliationStrategyRuleForMarmot dataclass 

AffiliationStrategyRuleForMarmot(
    rule_id,
    account_id,
    name,
    company_id,
    company_name,
    siret,
    company_name_from_siret,
    is_missing_siret,
    action,
    created_at,
    updated_at,
)

Bases: DataClassJsonMixin

Affiliation strategy rules displayed in Marmot.

account_id instance-attribute 
account_id
action instance-attribute 
action
company_id instance-attribute 
company_id
company_name instance-attribute 
company_name
company_name_from_siret instance-attribute 
company_name_from_siret
created_at instance-attribute 
created_at
is_missing_siret instance-attribute 
is_missing_siret
name instance-attribute 
name
rule_id instance-attribute 
rule_id
siret instance-attribute 
siret
updated_at instance-attribute 
updated_at

get_affiliation_strategy_rules_for_account 

get_affiliation_strategy_rules_for_account(account_id)

Get all affiliation strategy rules for a given account.

Source code in components/occupational_health/public/queries/strategy_rules.py 
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
def get_affiliation_strategy_rules_for_account(
    account_id: AccountId | UUID,
) -> list[AffiliationStrategyRuleForMarmot]:
    """Get all affiliation strategy rules for a given account."""
    account_id = AccountId(account_id)

    rules = current_session.scalars(
        select(AffiliationStrategyRule)
        .where(AffiliationStrategyRule.account_id == account_id)
        # Maybe we should order them by priority instead
        .order_by(AffiliationStrategyRule.created_at.desc())
    ).all()

    all_sirets = set(rule.siret for rule in rules if rule.siret)
    company_name_by_siret = {}
    for siret in all_sirets:
        company_data = get_app_dependency().get_company_from_siret(siret)
        if company_data:
            company_name = f"{company_data.name} ({company_data.city})"
            company_name_by_siret[siret] = company_name

    all_company_ids = set(rule.company_id for rule in rules if rule.company_id)
    company_name_by_company_id = get_app_dependency().get_company_names_by_ids(
        all_company_ids
    )

    return [
        AffiliationStrategyRuleForMarmot(
            rule_id=rule.id,
            account_id=rule.account_id,
            name=rule.name,
            company_name=company_name_by_company_id.get(rule.company_id)
            if rule.company_id
            else None,
            company_id=rule.company_id,
            siret=rule.siret,
            company_name_from_siret=company_name_by_siret.get(rule.siret)
            if rule.siret
            else None,
            is_missing_siret=rule.is_missing_siret,
            action=rule.action,
            created_at=rule.created_at,
            updated_at=rule.updated_at,
        )
        for rule in rules
    ]

subscribers

get_affiliated_members_for_account 

get_affiliated_members_for_account(
    account_id, profile_service
)

Return the list of currently affiliated members for a subscriber profile.

Source code in components/occupational_health/internal/business_logic/queries/subscribers/members.py 
 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
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
@inject_profile_service
@tracer_wrap()
def get_affiliated_members_for_account(
    account_id: AccountId,
    profile_service: ProfileService,
) -> list[SubscriberMember]:
    """
    Return the list of currently affiliated members for a subscriber profile.
    """
    affiliations = get_all_affiliations_for_account(
        account_id=account_id,
        only_active_on=utctoday(),
    )

    if not affiliations:
        return []

    occupational_health_profile_ids = {
        build_profile_id_from_affiliable_ref(affiliation.affiliable_ref)
        for affiliation in affiliations
    }

    profiles = list(
        current_session.execute(
            select(OccupationalHealthProfile).where(
                OccupationalHealthProfile.id.in_(occupational_health_profile_ids)
            )
        ).scalars()
    )
    profile_by_id = {profile.id: profile for profile in profiles}
    global_profile_ids = [profile.global_profile_id for profile in profiles]

    profiles_with_jobs = (
        get_occupational_health_profile_by_global_profile_ids_with_latest_job(
            global_profile_ids=global_profile_ids
        )
    )
    job_by_profile_id = {profile.id: job for profile, job in profiles_with_jobs}

    global_profiles = profile_service.get_profiles(global_profile_ids)
    global_profiles_by_id = {profile.id: profile for profile in global_profiles}

    company_names_by_profile_id = (
        get_company_names_by_occupational_health_profile_id_mapping(profiles)
    )

    risk_category_by_profile_id = get_current_risk_category_for_profiles(
        profile_ids=occupational_health_profile_ids,
    )

    employments_by_global_profile_id = get_employments_by_global_profile_id(
        account_id=account_id,
        global_profile_ids=global_profile_ids,
    )

    user_id_to_global_profile_id = get_user_id_to_global_profile_id_mapping(
        global_profile_ids=global_profile_ids
    )
    global_profile_id_to_user_id = {
        gp_id: user_id for user_id, gp_id in user_id_to_global_profile_id.items()
    }
    user_id_by_profile_id: dict[UserId, ProfileId] = {
        global_profile_id_to_user_id[profile.global_profile_id]: profile.id
        for profile in profiles
        if profile.global_profile_id in global_profile_id_to_user_id
    }

    # Fetch visits
    last_visit_by_profile_id = get_last_visit_for_profiles(
        profile_ids=occupational_health_profile_ids,
        user_id_mapping=user_id_by_profile_id,
    )
    next_visit_by_profile_id = get_future_visit_for_profiles(
        profile_ids=occupational_health_profile_ids,
        user_id_mapping=user_id_by_profile_id,
        employments_by_global_profile_id=employments_by_global_profile_id,
        affiliations=affiliations,
        risk_category_by_profile_id=risk_category_by_profile_id,
    )

    members = []
    for affiliation in affiliations:
        profile_id = build_profile_id_from_affiliable_ref(affiliation.affiliable_ref)

        if profile_id not in profile_by_id:
            current_logger.error(
                f"Profile {profile_id} not found for affiliation {affiliation.affiliation_id}",
            )
            continue

        profile = profile_by_id[profile_id]
        global_profile = global_profiles_by_id.get(profile.global_profile_id)

        if not global_profile:
            current_logger.error(
                f"Global profile {profile.global_profile_id} not found",
            )
            continue

        next_visit = next_visit_by_profile_id[profile_id]

        job = job_by_profile_id.get(profile_id)
        job_title = job.title if job else None

        company_names = list(company_names_by_profile_id.get(profile_id, set()))

        members.append(
            SubscriberMember(
                occupational_health_profile_id=str(profile_id),
                first_name=global_profile.first_name or "",
                last_name=global_profile.last_name or "",
                job_title=job_title,
                company_names=company_names,
                risk_category=risk_category_by_profile_id.get(profile_id),
                last_visit=last_visit_by_profile_id.get(profile_id),
                next_visit=next_visit,
                status=get_member_status(
                    next_visit=next_visit,
                ),
            )
        )

    return members

get_subscriber_for_account 

get_subscriber_for_account(account_id)

Get a subscriber for an account.

Source code in components/occupational_health/internal/business_logic/queries/subscribers/subscribers.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
def get_subscriber_for_account(account_id: UUID) -> Subscriber | None:
    """Get a subscriber for an account."""
    dependency = get_app_dependency()

    try:
        account_name = dependency.get_account_name(AccountId(account_id))
    except NoResultFound:
        return None

    return Subscriber(
        id=account_id,
        name=account_name,
    )

visits

get_all_visits_for_member 

get_all_visits_for_member(
    profile_service, occupational_health_profile_id
)

Get all visits for a specific member

Parameters:

Name Type Description Default
profile_service ProfileService

The profile service for fetching member information

 required
occupational_health_profile_id ProfileId | UUID

Optional filter by occupational health profile ID

 required

Returns:

Type Description
list[VisitInfo]

List[VisitInfo]: List of visits scheduled for the specified date
Source code in components/occupational_health/internal/business_logic/queries/visits/medical_app.py 
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
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
@inject_profile_service
def get_all_visits_for_member(
    profile_service: ProfileService,
    occupational_health_profile_id: ProfileId | UUID,
) -> list[VisitInfo]:
    """
    Get all visits for a specific member

    Args:
        profile_service: The profile service for fetching member information
        occupational_health_profile_id: Optional filter by occupational health profile ID

    Returns:
        List[VisitInfo]: List of visits scheduled for the specified date
    """

    if isinstance(occupational_health_profile_id, UUID):
        occupational_health_profile_id = ProfileId(occupational_health_profile_id)

    occupational_health_profile = (
        current_session.query(OccupationalHealthProfile)  # noqa: ALN085
        .filter(OccupationalHealthProfile.id == occupational_health_profile_id)
        .one_or_none()
    )
    if not occupational_health_profile:
        raise Exception("Occupational Health Profile not found")
    global_profile_id = occupational_health_profile.global_profile_id
    subquery_backend = current_session.query(  # type: ignore[var-annotated] # noqa: ALN085
        OccupationalHealthVisit.id.label("visit_id"),
        OccupationalHealthVisit.profile_id.label("profile_id"),
        null().label("user_id"),
        OccupationalHealthVisit.visit_date.label("visit_date"),
        OccupationalHealthVisit.visit_type.label("visit_type"),
        literal_column(f"'{VisitStatus.HAPPENED}'").label("status"),
        OccupationalHealthVisit.health_professional_id.label("health_professional_id"),
    ).filter(OccupationalHealthVisit.profile_id == occupational_health_profile_id)

    subquery_predictable = current_session.query(  # noqa: ALN085
        TuringOccupationalHealthPredictableVisit.id.label("visit_id"),
        null().label("profile_id"),
        TuringOccupationalHealthPredictableVisit.user_id.label("user_id"),
        TuringOccupationalHealthPredictableVisit.date_planned.label("visit_date"),
        TuringOccupationalHealthPredictableVisit.visit_type.label("visit_type"),
        TuringOccupationalHealthPredictableVisit.status.label("status"),
        TuringOccupationalHealthPredictableVisit.hp_visit_owner_id.label(
            "health_professional_id"
        ),
    ).filter(TuringOccupationalHealthPredictableVisit.user_id == str(global_profile_id))
    subquery_on_demand = current_session.query(  # noqa: ALN085
        TuringOccupationalHealthOnDemandVisit.id.label("visit_id"),
        null().label("profile_id"),
        TuringOccupationalHealthOnDemandVisit.user_id.label("user_id"),
        TuringOccupationalHealthOnDemandVisit.date_planned.label("visit_date"),
        TuringOccupationalHealthOnDemandVisit.visit_type.label("visit_type"),
        TuringOccupationalHealthOnDemandVisit.status.label("status"),
        TuringOccupationalHealthOnDemandVisit.hp_visit_owner_id.label(
            "health_professional_id"
        ),
    ).filter(TuringOccupationalHealthOnDemandVisit.user_id == str(global_profile_id))

    subquery = subquery_backend.union_all(
        subquery_predictable,
        subquery_on_demand,
    ).subquery()

    all_visits = current_session.query(subquery).all()  # noqa: ALN085

    # Fetch HPs
    health_professional_mapping = {
        hp.id: hp
        for hp in current_session.query(  # noqa: ALN085
            OccupationalHealthHealthProfessional.id,
            OccupationalHealthHealthProfessional.first_name,
            OccupationalHealthHealthProfessional.last_name,
        )
    }

    # Convert to dataclass instances
    result = []
    profile = profile_service.get_or_raise_profile(
        occupational_health_profile.global_profile_id
    )
    for visit in all_visits:
        # Get health professional name
        health_professional = health_professional_mapping.get(
            visit.health_professional_id
        )
        health_professional_name = (
            f"{health_professional.first_name} {health_professional.last_name}".strip()
            if health_professional
            else None
        )

        result.append(
            VisitInfo(
                visit_id=visit.visit_id,
                member_first_name=profile.first_name or "⁉️",
                member_last_name=profile.last_name or "⁉️",
                member_gender=profile.gender,
                member_birthdate=profile.birth_date,
                visit_date=visit.visit_date,
                visit_type=visit.visit_type,
                health_professional_name=health_professional_name,
                health_professional_id=visit.health_professional_id,
                visit_status=visit.status,
                occupational_health_profile_id=occupational_health_profile_id,
            )
        )

    return result

get_most_recent_visit_by_health_professional_and_timestamp 

get_most_recent_visit_by_health_professional_and_timestamp(
    health_professional_id, on_datetime
)

Retrieves the most recent occupational health visit by a health professional within a specific time range.

Warning: only querying turing tables for now, could refactor with code above

Source code in components/occupational_health/internal/business_logic/queries/visits/visits.py 
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
def get_most_recent_visit_by_health_professional_and_timestamp(
    health_professional_id: UUID | None,
    on_datetime: datetime | None,
) -> Optional[PlannedVisit]:
    """
    Retrieves the most recent occupational health visit by a health professional within a specific time range.

    Warning: only querying turing tables for now, could refactor with code above
    """
    from sqlalchemy import select

    from components.occupational_health.external.global_profile import (
        get_occupational_health_profile_id,
    )
    from components.occupational_health.internal.enums.visit_status import VisitStatus
    from components.occupational_health.internal.models.turing.turing_occupational_health_on_demand_visit import (
        TuringOccupationalHealthOnDemandVisit,
    )
    from components.occupational_health.internal.models.turing.turing_occupational_health_predictable_visit import (
        TuringOccupationalHealthPredictableVisit,
    )
    from components.occupational_health.internal.types import UserId
    from shared.helpers.db import current_session

    if not health_professional_id or not on_datetime:
        raise RuntimeError(
            "Either or both health_professional_id / on_datetime must be provided."
        )

    on_demand_stmt = select(
        TuringOccupationalHealthOnDemandVisit.user_id,
        TuringOccupationalHealthOnDemandVisit.timestamp_planned,
    ).filter(
        TuringOccupationalHealthOnDemandVisit.hp_visit_owner_id
        == health_professional_id,
        TuringOccupationalHealthOnDemandVisit.date_planned == on_datetime.date(),
        # Add a 15 minutes buffer for the edge case when the visit starts a bit earlier than planned.
        TuringOccupationalHealthOnDemandVisit.timestamp_planned.between(
            on_datetime - timedelta(hours=1), on_datetime + timedelta(minutes=15)
        ),
        TuringOccupationalHealthOnDemandVisit.status == VisitStatus.HAPPENED,
    )

    predictable_stmt = select(
        TuringOccupationalHealthPredictableVisit.user_id,
        TuringOccupationalHealthPredictableVisit.timestamp_planned,
    ).filter(
        TuringOccupationalHealthPredictableVisit.hp_visit_owner_id
        == health_professional_id,
        TuringOccupationalHealthPredictableVisit.date_planned == on_datetime.date(),
        TuringOccupationalHealthPredictableVisit.timestamp_planned.between(
            on_datetime - timedelta(hours=1), on_datetime + timedelta(minutes=15)
        ),
        TuringOccupationalHealthPredictableVisit.status == VisitStatus.HAPPENED,
    )

    unioned = on_demand_stmt.union_all(predictable_stmt).subquery()
    stmt = (
        select(
            unioned.c.user_id.label("user_id"),
            unioned.c.timestamp_planned.label("timestamp_planned"),
        )
        .order_by(unioned.c.timestamp_planned.desc())
        .limit(1)
    )
    visit = current_session.execute(stmt).first()

    if not visit or not visit.user_id:
        current_logger.info(
            f"No visit found for HP {health_professional_id} starting within 1 hour before {on_datetime}."
        )
        return None

    user_id = UserId(visit.user_id)
    profile_id = get_occupational_health_profile_id(user_id)

    return PlannedVisit(
        profile_id=profile_id,
        timestamp_planned=visit.timestamp_planned,
    )

get_visits 

get_visits(profile_service, on_date)

Get all visits scheduled for a specific date. If no date is provided, return visits for today.

Parameters:

Name Type Description Default
profile_service ProfileService

The profile service for fetching member information

 required
on_date date | None

The date to fetch visits for. If None, uses today's date.

 required

Returns:

Type Description
list[VisitInfo]

List[VisitInfo]: List of visits scheduled for the specified date
Source code in components/occupational_health/internal/business_logic/queries/visits/medical_app.py 
 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
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
232
233
234
235
236
237
238
239
240
241
242
243
@inject_profile_service
def get_visits(
    profile_service: ProfileService,
    on_date: date | None,
) -> list[VisitInfo]:
    """
    Get all visits scheduled for a specific date.
    If no date is provided, return visits for today.

    Args:
        profile_service: The profile service for fetching member information
        on_date: The date to fetch visits for. If None, uses today's date.

    Returns:
        List[VisitInfo]: List of visits scheduled for the specified date
    """
    if not on_date:
        on_date = utctoday()

    # Query visits scheduled for the specified date

    subquery_backend = current_session.query(  # type: ignore[var-annotated] # noqa: ALN085
        OccupationalHealthVisit.profile_id.label("profile_id"),
        null().label("user_id"),
        OccupationalHealthVisit.visit_date.label("visit_date"),
        OccupationalHealthVisit.visit_type.label("visit_type"),
        literal_column(f"'{VisitStatus.HAPPENED}'").label("status"),
        OccupationalHealthVisit.health_professional_id.label("health_professional_id"),
    ).filter(
        # CAVEAT: the data is stored using the local date, not UTC
        OccupationalHealthVisit.visit_date == on_date,
        OccupationalHealthVisit.profile_id.is_not(None),
    )

    subquery_predictable = current_session.query(  # noqa: ALN085
        null().label("profile_id"),
        TuringOccupationalHealthPredictableVisit.user_id.label("user_id"),
        TuringOccupationalHealthPredictableVisit.date_planned.label("visit_date"),
        TuringOccupationalHealthPredictableVisit.visit_type.label("visit_type"),
        TuringOccupationalHealthPredictableVisit.status.label("status"),
        TuringOccupationalHealthPredictableVisit.hp_visit_owner_id.label(
            "health_professional_id"
        ),
    ).filter(
        TuringOccupationalHealthPredictableVisit.date_planned == on_date,
    )
    subquery_on_demand = current_session.query(  # noqa: ALN085
        null().label("profile_id"),
        TuringOccupationalHealthOnDemandVisit.user_id.label("user_id"),
        TuringOccupationalHealthOnDemandVisit.date_planned.label("visit_date"),
        TuringOccupationalHealthOnDemandVisit.visit_type.label("visit_type"),
        TuringOccupationalHealthOnDemandVisit.status.label("status"),
        TuringOccupationalHealthOnDemandVisit.hp_visit_owner_id.label(
            "health_professional_id"
        ),
    ).filter(
        TuringOccupationalHealthOnDemandVisit.date_planned == on_date,
    )

    subquery = subquery_backend.union_all(
        subquery_predictable,
        subquery_on_demand,
    ).subquery()

    all_visits = current_session.query(subquery).all()  # noqa: ALN085

    # Now, the mapping mess: we either have the user ID, or the (Occupational Health) profile ID

    # Get the global profile ID for the user IDs
    user_ids_found = {
        visit.user_id for visit in all_visits if visit.user_id is not None
    }
    user_id_mapping = (
        get_app_dependency().get_user_id_by_global_profile_id_mapping_from_user_ids(
            user_ids_found
        )
    )
    global_profile_id_by_user_id = {
        user_id: global_profile_id
        for global_profile_id, user_id in user_id_mapping.items()
    }

    # Get the list of found profile IDs
    profile_ids_found = {
        visit.profile_id for visit in all_visits if visit.profile_id is not None
    }

    # Based on those two lists, get the final mapping of (Occupational Health) profile IDs indexed by the Global Profile ID
    profile_id_by_global_profile_id: dict[GlobalProfileId, ProfileId] = {
        profile.global_profile_id: profile.id
        for profile in (
            current_session.query(  # noqa: ALN085
                OccupationalHealthProfile.id,
                OccupationalHealthProfile.global_profile_id,
            )
            .filter(
                or_(
                    OccupationalHealthProfile.global_profile_id.in_(
                        user_id_mapping.keys()
                    ),
                    OccupationalHealthProfile.id.in_(profile_ids_found),
                )
            )
            .all()
        )
    }

    # Get profile information from profile service
    profiles = profile_service.get_profiles(
        profile_ids=list(profile_id_by_global_profile_id.keys())
    )
    profiles_by_occupational_health_profile_id = {
        profile_id_by_global_profile_id[
            GlobalProfileId(global_profile.id)
        ]: global_profile
        for global_profile in profiles
    }

    # Fetch HPs
    health_professional_mapping = {
        hp.id: hp
        for hp in current_session.query(  # noqa: ALN085
            OccupationalHealthHealthProfessional.id,
            OccupationalHealthHealthProfessional.first_name,
            OccupationalHealthHealthProfessional.last_name,
        )
    }

    # Convert to dataclass instances
    result = []
    for visit in all_visits:
        # Get health professional name
        health_professional = health_professional_mapping.get(
            visit.health_professional_id
        )
        health_professional_name = (
            f"{health_professional.first_name} {health_professional.last_name}".strip()
            if health_professional
            else None
        )

        # Get member information from global profile
        occupational_health_profile_id = visit.profile_id
        if not occupational_health_profile_id:
            # Visit imported from GSheets, we only know the User ID
            global_profile_id = global_profile_id_by_user_id.get(visit.user_id)
            if not global_profile_id:
                current_logger.error(
                    f"Found spreadsheet visit with unknown user ID {visit.user_id} ({visit.visit_type} on date {visit.visit_date})."
                    " The user ID from the GSheet probably doesn't exist.",
                    stack_info=False,
                )
                # Do not crash but return "empty" info about member. HPs will not be
                # able to open the DMST.
                result.append(_make_empty_visit_info(visit, health_professional_name))
                continue

            if global_profile_id not in profile_id_by_global_profile_id:
                current_logger.error(
                    f"Global Profile ID {global_profile_id} ({visit.user_id=}) not found for visit ({visit.visit_type} on date {visit.visit_date})."
                    " The user is probably not affiliated to Prévenir.",
                    stack_info=False,
                )
                # Do not crash but return "empty" info about member. HPs will not be
                # able to open the DMST.
                result.append(_make_empty_visit_info(visit, health_professional_name))
                continue

            occupational_health_profile_id = profile_id_by_global_profile_id[
                global_profile_id
            ]

        if (
            occupational_health_profile_id
            not in profiles_by_occupational_health_profile_id
        ):
            current_logger.error(
                f"Occupational Health Profile ID {occupational_health_profile_id} ({visit.user_id=}) not found for visit ({visit.visit_type} on date {visit.visit_date})."
                "The profile is probably not affiliated to Prévenir.",
                stack_info=False,
            )
            # Do not crash but return "empty" info about member. HPs will not be
            # able to open the DMST.
            result.append(_make_empty_visit_info(visit, health_professional_name))
            continue

        global_profile = profiles_by_occupational_health_profile_id[
            occupational_health_profile_id
        ]

        result.append(
            VisitInfo(
                member_first_name=global_profile.first_name or "⁉️",
                member_last_name=global_profile.last_name or "⁉️",
                member_gender=global_profile.gender,
                member_birthdate=global_profile.birth_date,
                visit_date=visit.visit_date,
                visit_type=visit.visit_type,
                health_professional_name=health_professional_name,
                health_professional_id=visit.health_professional_id,
                visit_status=visit.status,
                visit_id=None,  # Not used for now in this method.
                occupational_health_profile_id=profile_id_by_global_profile_id[
                    GlobalProfileId(global_profile.id)
                ],
            )
        )

    return result

workspace_actions

get_all_workspace_actions 

get_all_workspace_actions(
    occupational_health_profile_id=None,
)

Get all workspace actions, optionally filtered by occupational health profile ID.

Parameters:

Name Type Description Default
occupational_health_profile_id Optional[UUID]

Optional profile ID to filter actions

 None

Returns:

Type Description
list[WorkspaceAction]

List of WorkspaceAction entities
Source code in components/occupational_health/public/queries/workspace_actions.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def get_all_workspace_actions(
    occupational_health_profile_id: Optional[UUID] = None,
) -> list[WorkspaceAction]:
    """
    Get all workspace actions, optionally filtered by occupational health profile ID.

    Args:
        occupational_health_profile_id: Optional profile ID to filter actions

    Returns:
        List of WorkspaceAction entities
    """
    return _get_all_workspace_actions(
        occupational_health_profile_id=occupational_health_profile_id
    )

components.occupational_health.public.thesaurus

expositions

queries

get_expositions 
get_expositions(exposition_notations)

Return the exposition data matching the given notations

Source code in components/occupational_health/internal/thesaurus/expositions/queries.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@tracer_wrap()
def get_expositions(
    exposition_notations: list[str],
) -> list[ThesaurusOccupationalExposure]:
    """
    Return the exposition data matching the given notations
    """
    expositions = ThesaurusOccupationalExposureBroker.get_by_notations(
        exposition_notations
    ).all()
    class_and_subclass_labels_by_notation = get_class_subclass_labels(list(expositions))
    return [
        ThesaurusOccupationalExposureMapper.to_entity(
            model, class_and_subclass_labels_by_notation
        )
        for model in expositions
    ]
search_expositions 
search_expositions(search_input)

Return the exposition data matching the given search input (case-insensitive and accent-insensitive)

Note: adds the class and subclass labels to the exposition objects

Source code in components/occupational_health/internal/thesaurus/expositions/queries.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@tracer_wrap()
def search_expositions(search_input: str) -> list[ThesaurusOccupationalExposure]:
    """
    Return the exposition data matching the given search input (case-insensitive and accent-insensitive)

    Note: adds the class and subclass labels to the exposition objects
    """
    expositions = ThesaurusOccupationalExposureBroker.search_in_label(
        search_input
    ).all()
    class_and_subclass_labels_by_notation = get_class_subclass_labels(list(expositions))
    return [
        ThesaurusOccupationalExposureMapper.to_entity(
            model, class_and_subclass_labels_by_notation
        )
        for model in expositions
    ]

means

queries

get_all_thesaurus_means 
get_all_thesaurus_means()
Source code in components/occupational_health/internal/thesaurus/means/queries.py 
11
12
13
14
@tracer_wrap()
def get_all_thesaurus_means() -> list[ThesaurusMean]:
    means = ThesaurusMeanBroker.get_all().all()
    return [ThesaurusMeanMapper.to_entity(model) for model in means]

terminologies

queries

search_terminologies 
search_terminologies(search_input)

Return the terminology data matching the given search input (case-insensitive and accent-insensitive)

Source code in components/occupational_health/internal/thesaurus/terminologies/queries.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@tracer_wrap()
def search_terminologies(search_input: str) -> list[Terminology]:
    """
    Return the terminology data matching the given search input (case-insensitive and accent-insensitive)
    """
    matching_data: list[Terminology] = []
    terminologies_data = _get_terminologies_data()

    # Normalize search input once
    normalized_search = remove_accents(search_input.lower())

    for terminology in terminologies_data:
        # Normalize the fields being searched
        normalized_label = remove_accents(terminology.label.lower())

        if normalized_search in normalized_label:
            matching_data.append(terminology)
    return matching_data

components.occupational_health.public.types

AccountId module-attribute 

AccountId = NewType('AccountId', UUID)

The Account ID type.

CompanyId module-attribute 

CompanyId = NewType('CompanyId', str)

The Company ID type. We use 'str' to not make assumption on the country's representation of the ID.

GlobalProfileId module-attribute 

GlobalProfileId = NewType('GlobalProfileId', UUID)

The Global Profile ID type.

ProfileId module-attribute 

ProfileId = NewType('ProfileId', UUID)

The Occupational Health Profile ID type.

UserId module-attribute 

UserId = NewType('UserId', str)

The country-specific User ID type. We use 'str' to not make assumption on the country's representation of the ID.