Skip to content

Api reference

components.occupational_health.public.actions

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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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,
            worker_health_status=dmst.administrative_profile.worker_health_status,
            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,
        )

    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

OCCUPATIONAL_HEALTH_COMPONENT_NAME module-attribute 

OCCUPATIONAL_HEALTH_COMPONENT_NAME = 'occupational_health'

OccupationalHealthDependency

Bases: ABC

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

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

create_profile_with_user abstractmethod 

create_profile_with_user(profile_data)

Create user profile with user data.

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

get_account_admins_profile_ids abstractmethod 

get_account_admins_profile_ids(account_id)

Get account admin profile IDs.

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

get_account_id abstractmethod 

get_account_id(company_id)

Get account ID for a company.

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

get_account_id_and_siret abstractmethod 

get_account_id_and_siret(company_id)

Get account ID and SIRET for a company.

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

get_account_name abstractmethod 

get_account_name(account_id)

Get account name.

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

get_company_admins_profile_ids abstractmethod 

get_company_admins_profile_ids(company_id)

Get company admin profile IDs.

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

get_company_display_name_and_account_id abstractmethod 

get_company_display_name_and_account_id(company_id)

Get company display name and associated account ID.

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

get_company_from_siret abstractmethod 

get_company_from_siret(siret)

Get company data from SIRET via DSN.

Source code in components/occupational_health/public/dependencies.py 
 97
 98
 99
100
@abstractmethod
def get_company_from_siret(self, siret: str) -> 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 
47
48
49
50
@abstractmethod
def get_company_ids_in_account(self, account_id: UUID) -> list[str]:
    """Get all company IDs in an account."""
    raise NotImplementedError()

get_company_names_by_ids abstractmethod 

get_company_names_by_ids(company_ids)

Get company names by IDs.

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

get_company_siren abstractmethod 

get_company_siren(company_id)

Get company SIREN.

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

get_company_siret abstractmethod 

get_company_siret(company_id, force_nic=None)

Get company SIRET, optionally forcing a specific NIC.

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

get_global_profile_id abstractmethod 

get_global_profile_id(user_id)

Get global profile ID for a user.

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

get_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 
137
138
139
140
@abstractmethod
def get_ssn_and_ntt_for_user(self, user_id: str) -> tuple[str | None, str | None]:
    """Set SSN and NTT on user."""
    raise NotImplementedError()

get_user_id_by_global_profile_mapping abstractmethod 

get_user_id_by_global_profile_mapping(global_profile_ids)

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

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

get_user_id_from_global_profile_id abstractmethod 

get_user_id_from_global_profile_id(global_profile_id)

Get user ID from global profile ID.

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

get_user_id_mapping_from_user_ids abstractmethod 

get_user_id_mapping_from_user_ids(user_ids)

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

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

get_work_stoppages_for_users abstractmethod 

get_work_stoppages_for_users(
    user_ids, ever_active_during_period=None
)

Get work stoppages for users.

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

set_ssn_ntt_on_user abstractmethod 

set_ssn_ntt_on_user(user_id, ssn, ntt)

Set SSN and NTT on user.

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

ProfileData dataclass 

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

The profile data used to create a user and profile.

birth_date class-attribute instance-attribute 

birth_date = None

email class-attribute instance-attribute 

email = None

first_name class-attribute instance-attribute 

first_name = None

gender class-attribute instance-attribute 

gender = None

language class-attribute instance-attribute 

language = None

last_name class-attribute instance-attribute 

last_name = None

phone_number class-attribute instance-attribute 

phone_number = None

get_app_dependency 

get_app_dependency()

Retrieves the occupational_health dependency

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

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

set_app_dependency 

set_app_dependency(dependency)

Sets the occupational_health dependency to the app

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

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

components.occupational_health.public.employment

employment_consumer

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

occupational_health_employment_change_consumer 

occupational_health_employment_change_consumer(
    employment_change, event_bus_orchestrator
)
Source code in components/occupational_health/public/employment/employment_consumer.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def occupational_health_employment_change_consumer(  # noqa: D103
    employment_change: "EmploymentChange[FrExtendedValues]",
    event_bus_orchestrator: "EventBusOrchestrator",
) -> None:
    try:
        from components.employment.public.enums import CountryCode

        if employment_change.country_code != CountryCode.fr:
            return

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

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

requires_siret_information

requires_siret_information 

requires_siret_information(account_id)

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

Source code in components/occupational_health/public/employment/requires_siret_information.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def requires_siret_information(account_id: UUID) -> bool:
    """
    Return whether the given account ID requires SIRET information to be provided when tracking employment movements.
    """
    if not has_active_occupational_health_contract(account_id):
        # No need to collect the SIRET if there is no active contract
        return False

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

    rules_count = rules.count()

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

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

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

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

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

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

components.occupational_health.public.entities

billing

BillingStrategy

Bases: AlanBaseEnum

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

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

CustomerAddress dataclass 

CustomerAddress(street, postal_code, city)

Postal address of a customer.

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

CustomerData dataclass 

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

Customer data needed for billing purposes.

billing_strategy property 
billing_strategy

Return the billing strategy for this customer.

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

InvoiceToGenerateData dataclass 

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

Data needed to generate an invoice for a customer.

billing_period instance-attribute 
billing_period
billing_year instance-attribute 
billing_year

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

contract_needs_installment_plan property 
contract_needs_installment_plan

Whether the contract needs an installment plan.

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

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

OccupationalHealthBillingPeriod

Bases: AlanBaseEnum

Billing period for occupational health services.

END_OF_YEAR_REGUL class-attribute instance-attribute 
END_OF_YEAR_REGUL = 'END_OF_YEAR_REGUL'
Q1 class-attribute instance-attribute 
Q1 = 'Q1'
Q2 class-attribute instance-attribute 
Q2 = 'Q2'
Q3 class-attribute instance-attribute 
Q3 = 'Q3'
Q4 class-attribute instance-attribute 
Q4 = 'Q4'
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 
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
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,
    worker_health_status,
    notes,
    medical_record_sharing_consent,
    medical_record_access_consent,
    health_data_sharing_consent,
    video_consultation_consent
)

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

medical_record_access_consent

medical_record_sharing_consent
notes instance-attribute 
notes
ssn instance-attribute 
ssn

video_consultation_consent
worker_health_status instance-attribute 
worker_health_status

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

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

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'

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.marmot

actions

Actions for use in Marmot controllers.

AffiliationRequest dataclass 

AffiliationRequest(user_id, start_date, extra_data)

Bases: DataClassJsonMixin

extra_data instance-attribute 
extra_data
start_date instance-attribute 
start_date
user_id instance-attribute 
user_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 UUID

The account ID to affiliate members to

 required
affiliations list[AffiliationRequest]

List of tuples (user_id, start_date)

 required

Returns:

Type Description
list[UUID]

List of affiliation IDs
Source code in components/occupational_health/internal/business_logic/actions/affiliation_tool.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
def affiliate_multiple_members(
    account_id: UUID,
    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 tuples (user_id, start_date)

    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)

        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)

    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,
    actor_user_id=None
)

Affiliate a new member to occupational health.

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

    profile_id = get_or_create_profile_id(user_id)

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

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

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

cancel_affiliation_for_marmot 

cancel_affiliation_for_marmot(
    affiliation_id, actor_user_id
)

Cancel an affiliation for a profile.

This is a wrapper around cancel_affiliation that adds logging.

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

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

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

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

    cancel_affiliation(affiliation_id=affiliation_id)

create_contract 

create_contract(session, account_id, start_date)
Source code in components/occupational_health/internal/business_logic/contracting/actions.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@transactional(propagation=Propagation.REQUIRES_NEW)
def create_contract(
    session: Session,  # noqa:ARG001
    account_id: UUID,
    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

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

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

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

MemberToAffiliate dataclass 

MemberToAffiliate(
    row_index,
    first_name,
    last_name,
    birthdate,
    ssn,
    proposed_user_id,
    proposed_user_name,
    proposed_birthdate,
    matched_name,
    matched_birthdate,
    matching_result,
    affiliations,
    employments,
    target_start_date,
    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
matched_birthdate instance-attribute 
matched_birthdate
matched_name instance-attribute 
matched_name
matching_result instance-attribute 
matching_result
proposed_birthdate instance-attribute 
proposed_birthdate
proposed_user_id instance-attribute 
proposed_user_id
proposed_user_name instance-attribute 
proposed_user_name
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

OccupationalHealthProfileData dataclass 

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

Bases: DataClassJsonMixin

All occupational health data for a profile

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

OccupationalHealthVisitForMarmot dataclass 

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

Bases: DataClassJsonMixin

Represent a visit for display in Marmot

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

enums

MovementType

Bases: AlanBaseEnum

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

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

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

queries

Queries for use in Marmot controllers.

get_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 
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
@cached_for(minutes=30)
@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_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 
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
@tracer_wrap()
def get_affiliated_members_for_marmot(
    filter_by_account_id: UUID | None = None,
) -> list[AffiliatedMemberForMarmot]:
    """
    Return the list of affiliated members (including past ones) for display in Marmot.
    """
    from components.affiliation_occ_health.public.api import (
        get_affiliation_list,
    )
    from components.occupational_health.internal.business_logic.queries.marmot import (
        get_affiliated_members_from_affiliations_for_marmot,
    )
    from components.occupational_health.internal.helpers.affiliation import (
        build_subscription_ref,
    )

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

    return get_affiliated_members_from_affiliations_for_marmot(
        affiliations=affiliations
    )

get_affiliation_decisions_for_marmot 

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

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

Source code in components/occupational_health/public/marmot/queries.py 
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
@tracer_wrap()
@inject_profile_service
def get_affiliation_decisions_for_marmot(
    profile_service: ProfileService,
    account_id: UUID,
    decision: AffiliationDecision | None = None,
    not_older_than: date | None = None,
) -> list[OccupationalHealthAffiliationDecisionWithProfileForMarmot]:
    """
    Return the list of affiliation decisions, optionally filtered by decision type and date.
    """
    company_ids = get_company_ids_in_account(account_id)
    query = (
        current_session.query(OccupationalHealthAffiliationDecision)  # noqa: ALN085
        .join(OccupationalHealthAffiliationDecision.rule, isouter=True)
        .join(OccupationalHealthAffiliationDecision.profile)
        .options(
            joinedload(OccupationalHealthAffiliationDecision.rule),
            joinedload(OccupationalHealthAffiliationDecision.profile),
            raiseload("*"),
        )
        .filter(
            cast(
                OccupationalHealthAffiliationDecision.employment_change_payload[
                    "core_employment_version"
                ]["company_id"].astext,
                Integer,
            ).in_(company_ids),
        )
        .order_by(OccupationalHealthAffiliationDecision.created_at.desc())
    )

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

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

    decisions = query.all()

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

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

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

get_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_tool.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def get_contract_info_for_account(account_id: UUID) -> 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)

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

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

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

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

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

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

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

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

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

get_recent_affiliation_movements 

get_recent_affiliation_movements(
    profile_service, account_id, not_older_than
)

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

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

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

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

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

    affiliation_movements_to_return: list[AffiliationMovementForMarmot] = []

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

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

    return affiliation_movements_to_return

parse_and_match_spreadsheet 

parse_and_match_spreadsheet(
    profile_service, account_id, spreadsheet_data
)

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

Source code in components/occupational_health/internal/business_logic/queries/affiliation_tool.py 
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
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
414
415
416
417
418
419
420
421
422
423
424
@inject_profile_service
@tracer_wrap()
def parse_and_match_spreadsheet(
    profile_service: ProfileService,
    account_id: UUID,
    spreadsheet_data: str,
) -> list[MemberToAffiliate]:
    """
    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,
    )

    rows = parse_spreadsheet_data(spreadsheet_data)

    _raise_if_invalid_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[str, 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[str] = set()

    # Detect the date format for the entire file
    all_birthdates: set[str] = set(
        row["date_naissance_gsheet"]
        for row in rows
        if "date_naissance_gsheet" in row and row.get("date_naissance_gsheet")
    )
    date_format = detect_date_format(all_birthdates)

    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, date_format).date()
            if birthdate_str
            else None
        )
        ssn = row.get("ssn_original", "").strip()
        email = row.get("email_pro_gsheet", "").strip()

        # 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.strip(),
            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=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 = {
            str(matching_result.matched_user_id)
            if matching_result.matched_user_id
            else None,
            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,
                proposed_user_id=proposed_user_id,
                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
                ),
                employments=[],  # Will be filled below
                # TODO: clamp with employment date
                target_start_date=default_start_date,
                affiliations=relevant_affiliations,
                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(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(str(match_user_id), [])
            match.employments = [
                EmploymentInfo(
                    company_id=employment.company_id,
                    company_name=company_names.get(employment.company_id, "Unknown!"),
                    start_date=employment.start_date,
                    end_date=employment.end_date,
                )
                for employment in user_employments
            ]

    return matches

components.occupational_health.public.queries

billing

Set of queries used by the Billing stack.

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

ContractRefNotFound

Bases: Exception

A contract reference was not found.

UnableToDetermineCustomerToBill

Bases: Exception

Unable to determine which customer to bill.

build_invoices_data_for_period 

build_invoices_data_for_period(
    billing_period, billing_year
)

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

Parameters:

Name Type Description Default
billing_period OccupationalHealthBillingPeriod

The billing period to build data for.

 required
billing_year int

The year of the billing period to build data for.

 required

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

Returns:

Type Description
list[InvoiceToGenerateData]

list[InvoiceToGenerateData]: The data about all invoices to generate for the given period.
Source code in components/occupational_health/public/queries/billing.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def build_invoices_data_for_period(
    billing_period: OccupationalHealthBillingPeriod,
    billing_year: int,
) -> list[InvoiceToGenerateData]:
    """
    Build the data required to generate all invoices for the given billing period and invoice year.

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

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

    Returns:
        list[InvoiceToGenerateData]: The data about all invoices to generate for the given period.
    """
    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),
        )

        invoice_to_generate_data_list: list[InvoiceToGenerateData] = []

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

        return invoice_to_generate_data_list

build_invoices_data_for_period_and_account 

build_invoices_data_for_period_and_account(
    billing_period, billing_year, account_id
)

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

Parameters:

Name Type Description Default
billing_period OccupationalHealthBillingPeriod

The billing period to build data for.

 required
billing_year int

The year of the billing period to build data for.

 required
account_id UUID

The account ID to build invoices for.

 required

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

Returns:

Type Description
list[InvoiceToGenerateData]

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

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

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

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

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

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

    # Compute the start and end dates, between which we'll consider affiliations
    affiliations_period = billing_period.to_validity_period(year=billing_year)

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

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

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

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

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

check_data 

check_data(invoices_data)

Temporary function to display the data produced by Occupational Health.

Source code in components/occupational_health/public/queries/billing.py 
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
420
421
422
423
424
425
426
427
428
429
430
431
def check_data(invoices_data: list[InvoiceToGenerateData]) -> None:
    """
    Temporary function to display the data produced by Occupational Health.
    """
    for invoice in invoices_data:
        customer_data = get_customer_data(invoice.contract_ref)
        click.secho(
            f"{invoice.billing_period} {invoice.billing_year} invoice for {customer_data.entity_name} ({customer_data.postal_address.city})",
            bold=True,
        )
        click.secho(invoice, dim=True)

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

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

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

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

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

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

get_customer_data 

get_customer_data(contract_ref)

Fetch customer data for the billing system.

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

Parameters:

Name Type Description Default
contract_ref str

The contract reference to look up customer data for

 required

Returns:

Name Type Description
CustomerData CustomerData

The customer data for the given contract reference

Raises:

Type Description
ContractRefNotFound

If the contract reference is not found in the customer data mapping
Source code in components/occupational_health/public/queries/billing.py 
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
def get_customer_data(contract_ref: str) -> CustomerData:
    """
    Fetch customer data for the billing system.

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

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

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

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

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

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

has_active_occupational_health_contract

has_active_occupational_health_contract 

has_active_occupational_health_contract(account_id)

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

Parameters:

Name Type Description Default
account_id UUID

The ID of the account to check.

 required

Returns:

Name Type Description
bool bool

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

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

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

    return _has_active_contract(account_id)

is_currently_affiliated_to_prevenir

is_currently_affiliated_to_occupational_health 

is_currently_affiliated_to_occupational_health(
    global_profile_id, account_id, commit=True
)

Check if the user is currently affiliated to Prevenir.

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

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

Source code in components/occupational_health/public/queries/is_currently_affiliated_to_prevenir.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def is_currently_affiliated_to_occupational_health(
    global_profile_id: UUID,
    account_id: UUID,
    commit: bool = True,
) -> bool:
    """
    Check if the user is currently affiliated to Prevenir.

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

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

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

    if active_affiliations:
        return True

    return False

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 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@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
    )
    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(
            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(profile.id)
    ssn, _ = get_app_dependency().get_ssn_and_ntt_for_user(user_id)

    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)
                        exposition_notations=[],  # In medical secrecy worker job (see get_medical_record)
                    )
                    for (index, job) in enumerate(jobs)
                ]
            ),
        ),
        administrative_profile=(
            AdministrativeProfile(
                ssn=ssn,
                worker_health_status=administrative_profile.worker_health_status,
                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,
            )
            if administrative_profile
            else AdministrativeProfile(
                ssn=ssn,
                worker_health_status=[],
                notes=None,
                medical_record_sharing_consent=False,
                medical_record_access_consent=False,
                health_data_sharing_consent=False,
                video_consultation_consent=False,
            )
        ),
        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 companies based on provided search input.

Note

The function currently only searches for members by full name. Company search and additional member fields (job_title, company, risk_category) are planned to be implemented.

Source code in components/occupational_health/internal/business_logic/queries/search/search.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 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
@inject_profile_service
def get_search_results(
    profile_service: ProfileService, search_input: str
) -> SearchResult:
    """
    Search for members and companies based on provided search input.

    Note:
        The function currently only searches for members by full name.
        Company search and additional member fields (job_title, company, risk_category)
        are planned to be implemented.
    """
    # 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[UUID, list[Profile]] = group_by(
        global_profiles, lambda profile: profile.id
    )
    global_profiles_ids = [profile.id for profile in global_profiles]
    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_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()
    }

    def get_latest_job_name_by_occupational_health_profile_id(
        occupational_health_profile_id: UUID,
    ) -> 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: UUID,
    ) -> 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_company_names_by_occupational_health_profile_id(
        occupational_health_profile_id: UUID,
    ) -> 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=None,  # todo alexandre.dubreuil - add
            )
            for occupational_health_profile_id in occupational_health_profile_ids_by_global_profile_id.values()
        ],
        companies=[],  # todo alexandre.dubreuil - add
    )

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 
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
def get_affiliation_strategy_rules_for_account(
    account_id: UUID,
) -> list[AffiliationStrategyRuleForMarmot]:
    """Get all affiliation strategy rules for a given account."""
    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
    ]

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 str

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 
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
@inject_profile_service
def get_all_visits_for_member(
    profile_service: ProfileService, occupational_health_profile_id: str
) -> 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
    """

    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 == str(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=UUID(occupational_health_profile_id),
            )
        )

    return result

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 
 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
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
@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_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 = {
        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[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 member information from global profile
        occupational_health_profile_id = (
            visit.profile_id
            or (
                profile_id_by_global_profile_id[
                    global_profile_id_by_user_id[visit.user_id]
                ]
            )
        )
        profile = profiles_by_occupational_health_profile_id[
            occupational_health_profile_id
        ]

        # 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(
                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,
                visit_id=None,  # Not used for now in this method.
                occupational_health_profile_id=profile_id_by_global_profile_id[
                    profile.id
                ],
            )
        )

    return result

workspace_actions

get_all_workspace_actions 

get_all_workspace_actions(profile_service)
Source code in components/occupational_health/internal/business_logic/queries/workspace_actions/workspace_actions.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 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
@inject_profile_service
def get_all_workspace_actions(profile_service: ProfileService) -> list[WorkspaceAction]:
    actions = list(OccupationalHealthWorkspaceActionBroker.get_all())
    global_profile_ids = list(
        {
            action.profile.global_profile_id
            for action in actions
            if action.profile is not None
        }
    )
    profile_with_jobs = (
        get_occupational_health_profile_by_global_profile_ids_with_latest_job(
            global_profile_ids
        )
    )
    occupational_health_profiles = [
        action.profile for action in actions if action.profile is not None
    ]
    global_profiles = profile_service.get_profiles(global_profile_ids)
    global_profiles_by_id: dict[UUID, list[Profile]] = group_by(
        global_profiles, lambda profile: profile.id
    )
    occupational_health_profiles_to_latest_health_job_mapping = {
        profile.id: job
        for (
            profile,
            job,
        ) in profile_with_jobs
    }
    occupational_health_profile_ids_by_global_profile_id = (
        get_occupational_health_profile_ids_by_global_profile_id_mapping(
            global_profile_ids
        )
    )
    global_profile_ids_by_occupational_health_profile_ids = {
        value: key
        for key, value in occupational_health_profile_ids_by_global_profile_id.items()
    }
    company_names_by_occupational_health_profile_ids = (
        get_company_names_by_occupational_health_profile_id_mapping(
            occupational_health_profiles
        )
    )

    # TODO @david.barthelemy: Challenge this and store the current company in the OccupationalHealthWorkspaceAction when the WorkspaceAction is created
    def get_latest_job_name_by_occupational_health_profile_id(
        occupational_health_profile_id: UUID,
    ) -> 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_full_name_by_occupational_health_profile_id(
        occupational_health_profile_id: UUID,
    ) -> str:
        global_profile_id = global_profile_ids_by_occupational_health_profile_ids[
            occupational_health_profile_id
        ]
        profile = one(global_profiles_by_id[global_profile_id])
        return f"{profile.first_name} {profile.last_name}"

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

    return [
        WorkspaceAction(
            id=action.id,
            title=action.title,
            target_type=WorkspaceActionType.member
            if action.profile_id
            else WorkspaceActionType.company,
            company_name=get_company_names_by_occupational_health_profile_id(
                action.profile_id
            )
            if action.profile_id
            else None,
            members_count=None,
            member_full_name=get_full_name_by_occupational_health_profile_id(
                action.profile_id
            )
            if action.profile_id
            else None,
            job_title=get_latest_job_name_by_occupational_health_profile_id(
                action.profile_id
            )
            if action.profile_id
            else None,
            health_professional_fullname=action.actor_full_name,
            profile_id=action.profile_id,
            account_id=action.account_id,
            eta_date=action.eta_date,
            status=str(action.status),
            prevention_type=str(action.prevention_type)
            if action.prevention_type
            else None,
            note=action.note,
        )
        for action in actions
    ]

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