Api reference

components.clinic.public.business_logic

gdpr_compliance

CLINIC_COUNTRY_APP_NAMES module-attribute

CLINIC_COUNTRY_APP_NAMES = [ALAN_FR]

CLINIC_RETENTION_PERIOD_IN_YEARS module-attribute

CLINIC_RETENTION_PERIOD_IN_YEARS = 2

delete_member_data

delete_member_data(global_profile_id)

Delete all clinic data related to a specific member.

Source code in components/clinic/public/business_logic/gdpr_compliance.py

def delete_member_data(global_profile_id: UUID) -> None:
    """Delete all clinic data related to a specific member."""
    from components.clinic.internal.business_logic.clinic_user import (
        delete_all_clinic_data_for_user,
    )

    is_deletion_enabled = bool_feature_flag(
        feature_flag_key="clinic-gdpr-member-data-deletion-killswitch",
        default_value=False,
    )

    current_logger.info(
        "Starting clinic member data deletion",
        global_profile_id=global_profile_id,
        is_deletion_enabled=is_deletion_enabled,
    )

    for app_name in CLINIC_COUNTRY_APP_NAMES:
        feature_user_to_delete = _get_app_feature_user_from_profile_id(
            global_profile_id, app_name
        )

        if not feature_user_to_delete:
            current_logger.info(
                "No feature user found, skipping",
                global_profile_id=global_profile_id,
                app_name=app_name.value,
            )
            continue

        try:
            clinic_user = get_existing_clinic_user_for_feature_user(
                app_user=feature_user_to_delete
            )
        except BaseErrorCode:
            current_logger.info(
                "No clinic user found, skipping",
                global_profile_id=global_profile_id,
                app_id=feature_user_to_delete.app_id,
                app_user_id=feature_user_to_delete.app_user_id,
            )
            continue

        action = "Deleting" if is_deletion_enabled else "Would have deleted"
        current_logger.info(
            f"{action} clinic data for user",
            global_profile_id=global_profile_id,
            clinic_user_id=clinic_user.id,
            is_deletion_enabled=is_deletion_enabled,
        )

        if is_deletion_enabled:
            delete_all_clinic_data_for_user(
                clinic_user_id=clinic_user.id, dry_run=False
            )

    current_logger.info(
        "Completed clinic member data deletion",
        global_profile_id=global_profile_id,
        is_deletion_enabled=is_deletion_enabled,
    )

get_profiles_ready_for_deletion

get_profiles_ready_for_deletion()

Return a list of inactive global profile IDs that are ready for deletion. We use global profile IDs, because user IDs are not unique between apps (FR/ES/BE)

in the clinic, we define a member as "ready for deletion" in https://github.com/alan-eu/Topics/discussions/31845?sort=old#discussioncomment-15755603 ⧉

The summary of the properties we are interested in: +----------------------------+-----------------------------------------------+-----------------------------------------------+ | Entity Name | Properties Checked | Rationale for Properties Checked | +============================+===============================================+===============================================+ | Medical Conversations | created_at (time of last medical conversation | The user sending a message clearly denotes an | | | part sent by user) | intentional interaction with our health | | | | services. | +----------------------------+-----------------------------------------------+-----------------------------------------------+ | Therapist Booking Sessions | ends_at (time of most recent therapist booking | A session is a clear intention of interaction | | | session where the user was the attending member | with our health services; the time it ends is | | | (clinic_user field)) | when that interaction ended. Sometimes followed| | | | by a medical conversation (covered above). | +----------------------------+-----------------------------------------------+-----------------------------------------------+ | Health Programs | last_updated_at (time of most recent health | This property is updated with any interaction | | | program progress created by the user; denoted by| with a health program. Some interactions may | | | app_id and user_id) | not truly be “program” interactions (e.g., a | | | | rating), but we don’t have a better way to | | | | differentiate between actions, and this catches| | | | them all. | +----------------------------+-----------------------------------------------+-----------------------------------------------+

Returns:

Type	Description
list[UUID]	List[uuid.UUID]: List of global profile IDs ready for deletion

Source code in components/clinic/public/business_logic/gdpr_compliance.py

@tracer.wrap(service="clinic")
def get_profiles_ready_for_deletion() -> list[UUID]:
    """
    Return a list of inactive global profile IDs that are ready for deletion.
    We use global profile IDs, because user IDs are not unique between apps (FR/ES/BE)

    in the clinic, we define a member as "ready for deletion" in https://github.com/alan-eu/Topics/discussions/31845?sort=old#discussioncomment-15755603

    The summary of the properties we are interested in:
    +----------------------------+-----------------------------------------------+-----------------------------------------------+
    | Entity Name                | Properties Checked                              | Rationale for Properties Checked               |
    +============================+===============================================+===============================================+
    | Medical Conversations      | created_at (time of last medical conversation   | The user sending a message clearly denotes an  |
    |                            | part sent by user)                              | intentional interaction with our health        |
    |                            |                                               | services.                                     |
    +----------------------------+-----------------------------------------------+-----------------------------------------------+
    | Therapist Booking Sessions | ends_at (time of most recent therapist booking  | A session is a clear intention of interaction  |
    |                            | session where the user was the attending member | with our health services; the time it ends is  |
    |                            | (clinic_user field))                            | when that interaction ended. Sometimes followed|
    |                            |                                               | by a medical conversation (covered above).     |
    +----------------------------+-----------------------------------------------+-----------------------------------------------+
    | Health Programs            | last_updated_at (time of most recent health     | This property is updated with any interaction  |
    |                            | program progress created by the user; denoted by| with a health program. Some interactions may   |
    |                            | app_id and user_id)                             | not truly be “program” interactions (e.g., a   |
    |                            |                                               | rating), but we don’t have a better way to     |
    |                            |                                               | differentiate between actions, and this catches|
    |                            |                                               | them all.                                      |
    +----------------------------+-----------------------------------------------+-----------------------------------------------+

    Returns:
        List[uuid.UUID]: List of global profile IDs ready for deletion
    """
    from components.clinic.public.dependencies import get_app_dependency

    all_candidate_profile_ids: list[UUID] = []
    profile_to_feature_user: dict[UUID, FeatureUser] = {}
    candidate_inactive_feature_users: list[FeatureUser] = []

    for app_name in CLINIC_COUNTRY_APP_NAMES:
        # collect candidate profiles and their feature user per country
        # each profile has at most 1 feature user per country
        adapter = get_app_dependency(app_name.value)
        current_logger.info("fetching inactive profile IDs", app_name=app_name.value)
        inactive_profile_ids = adapter.get_inactive_profile_ids_since(
            period_in_years=CLINIC_RETENTION_PERIOD_IN_YEARS
        )
        current_logger.info(
            "fetched inactive profile IDs",
            app_name=app_name.value,
            count=len(inactive_profile_ids),
        )
        all_candidate_profile_ids.extend(inactive_profile_ids)

        current_logger.info(
            "resolving profile IDs to feature users",
            app_name=app_name.value,
            count=len(inactive_profile_ids),
        )
        for inactive_profile_id in inactive_profile_ids:
            inactive_feature_user = _get_app_feature_user_from_profile_id(
                inactive_profile_id, app_name
            )

            if inactive_feature_user:
                profile_to_feature_user[inactive_profile_id] = inactive_feature_user
                candidate_inactive_feature_users.append(inactive_feature_user)

        current_logger.info(
            "resolved feature users",
            app_name=app_name.value,
            resolved=len(profile_to_feature_user),
        )

    current_logger.info(
        "total candidate profiles", count=len(all_candidate_profile_ids)
    )

    # batch get existing clinic users (no creation)
    current_logger.info("fetching existing clinic users")
    existing_clinic_users = get_existing_clinic_users_for_feature_users(
        candidate_inactive_feature_users
    )
    current_logger.info(
        "fetched existing clinic users", count=len(existing_clinic_users)
    )

    # resolve profile_id -> clinic_user_id directly
    feature_user_key_to_clinic_user_id = {
        (clinic_user.app_id, clinic_user.app_user_id): clinic_user.id
        for clinic_user in existing_clinic_users
    }
    inactive_profile_to_clinic_user_id = {
        profile_id: feature_user_key_to_clinic_user_id.get(
            (feature_user.app_id, feature_user.app_user_id)
        )
        for profile_id, feature_user in profile_to_feature_user.items()
    }

    candidates_without_clinic_user = len(all_candidate_profile_ids) - len(
        feature_user_key_to_clinic_user_id
    )
    current_logger.info(
        "candidates without clinic user", count=candidates_without_clinic_user
    )

    # batch check for recent interactions
    potentially_inactive_clinic_user_ids = [
        clinic_user_id
        for clinic_user_id in inactive_profile_to_clinic_user_id.values()
        if clinic_user_id is not None
    ]

    current_logger.info(
        "checking recent interactions",
        clinic_user_count=len(potentially_inactive_clinic_user_ids),
    )

    # union of all recent interaction sources
    active_via_conversations = (
        get_subset_of_clinic_user_ids_with_recent_medical_conversation_interactions(
            potentially_inactive_clinic_user_ids
        )
    )
    current_logger.info("active via conversations", count=len(active_via_conversations))

    active_via_sessions = (
        get_subset_of_clinic_user_ids_with_recent_session_interactions(
            potentially_inactive_clinic_user_ids
        )
    )
    current_logger.info("active via sessions", count=len(active_via_sessions))

    active_via_health_programs = (
        get_subset_of_clinic_user_ids_with_recent_health_program_interactions(
            candidate_inactive_feature_users, feature_user_key_to_clinic_user_id
        )
    )
    current_logger.info(
        "active via health programs", count=len(active_via_health_programs)
    )

    active_clinic_user_ids = (
        active_via_conversations | active_via_sessions | active_via_health_programs
    )
    current_logger.info(
        "total active clinic users (union)", count=len(active_clinic_user_ids)
    )

    # disqualify profiles with any recent interaction
    result = [
        profile_id
        for profile_id in all_candidate_profile_ids
        if inactive_profile_to_clinic_user_id.get(profile_id)
        not in active_clinic_user_ids
    ]

    current_logger.info("profiles ready for deletion", count=len(result))
    return result

get_subset_of_clinic_user_ids_with_recent_health_program_interactions

get_subset_of_clinic_user_ids_with_recent_health_program_interactions(
    feature_users, feature_user_key_to_clinic_user_id
)

Return the subset of clinic user IDs whose corresponding feature users have had health program interactions in the last CLINIC_RETENTION_PERIOD_IN_YEARS years.

Checks the last_updated_at time of the most recent health program progress for each feature user, then maps active feature users back to clinic user IDs.

Parameters:

Name	Type	Description	Default
feature_users	list[FeatureUser]	List of feature users to check	required
feature_user_key_to_clinic_user_id	dict[tuple[str, str], UUID]	Mapping of (app_id, app_user_id) to clinic user ID	required

Source code in components/clinic/public/business_logic/gdpr_compliance.py

@tracer.wrap(service="clinic")
def get_subset_of_clinic_user_ids_with_recent_health_program_interactions(
    feature_users: list[FeatureUser],
    feature_user_key_to_clinic_user_id: dict[tuple[str, str], UUID],
) -> set[UUID]:
    """
    Return the subset of clinic user IDs whose corresponding feature users have had
    health program interactions in the last CLINIC_RETENTION_PERIOD_IN_YEARS years.

    Checks the last_updated_at time of the most recent health program progress for each feature user,
    then maps active feature users back to clinic user IDs.

    Args:
        feature_users: List of feature users to check
        feature_user_key_to_clinic_user_id: Mapping of (app_id, app_user_id) to clinic user ID
    """
    if not feature_users:
        return set()

    cutoff = datetime.now(tz=UTC) - relativedelta(
        years=CLINIC_RETENTION_PERIOD_IN_YEARS
    )
    active_health_programs_feature_users = (
        get_feature_users_with_health_program_interaction_after(feature_users, cutoff)
    )

    active_clinic_user_ids: set[UUID] = set()
    for feature_user in active_health_programs_feature_users:
        clinic_user_id = feature_user_key_to_clinic_user_id.get(
            (feature_user.app_id, feature_user.app_user_id)
        )
        if clinic_user_id is not None:
            active_clinic_user_ids.add(clinic_user_id)

    return active_clinic_user_ids

get_subset_of_clinic_user_ids_with_recent_medical_conversation_interactions

get_subset_of_clinic_user_ids_with_recent_medical_conversation_interactions(
    clinic_user_ids,
)

Return clinic user IDs that had medical conversation interactions within retention period.

Uses batched IN clauses to avoid overwhelming PostgreSQL with large queries.

Parameters:

Name	Type	Description	Default
clinic_user_ids	list[UUID]	List of clinic user IDs to check	required

Source code in components/clinic/public/business_logic/gdpr_compliance.py

@tracer.wrap(service="clinic")
def get_subset_of_clinic_user_ids_with_recent_medical_conversation_interactions(
    clinic_user_ids: list[UUID],
) -> set[UUID]:
    """Return clinic user IDs that had medical conversation interactions within retention period.

    Uses batched IN clauses to avoid overwhelming PostgreSQL with large queries.

    Args:
        clinic_user_ids: List of clinic user IDs to check
    """
    if not clinic_user_ids:
        return set()

    cutoff = datetime.now() - relativedelta(years=CLINIC_RETENTION_PERIOD_IN_YEARS)
    total_batches = math.ceil(len(clinic_user_ids) / _BATCH_SIZE)
    current_logger.info(
        "checking recent conversations in batches",
        total=len(clinic_user_ids),
        total_batches=total_batches,
    )

    result: set[UUID] = set()
    for batch_idx in range(total_batches):
        start = batch_idx * _BATCH_SIZE
        batch = clinic_user_ids[start : start + _BATCH_SIZE]

        rows = (
            current_session.execute(
                select(MedicalConversationPart.clinic_user_author_id)
                .where(
                    MedicalConversationPart.clinic_user_author_id.in_(batch),
                    MedicalConversationPart.created_at > cutoff,
                )
                .distinct()
            )
            .scalars()
            .all()
        )

        batch_result = {uid for uid in rows if uid is not None}
        result.update(batch_result)
        current_logger.info(
            "checked conversations batch",
            batch=batch_idx + 1,
            total_batches=total_batches,
            batch_results=len(batch_result),
        )

    return result

get_subset_of_clinic_user_ids_with_recent_session_interactions

get_subset_of_clinic_user_ids_with_recent_session_interactions(
    clinic_user_ids,
)

Return clinic user IDs that had therapist booking sessions within retention period.

Uses batched IN clauses to avoid overwhelming PostgreSQL with large queries.

Parameters:

Name	Type	Description	Default
clinic_user_ids	list[UUID]	List of clinic user IDs to check	required

Source code in components/clinic/public/business_logic/gdpr_compliance.py

@tracer.wrap(service="clinic")
def get_subset_of_clinic_user_ids_with_recent_session_interactions(
    clinic_user_ids: list[UUID],
) -> set[UUID]:
    """Return clinic user IDs that had therapist booking sessions within retention period.

    Uses batched IN clauses to avoid overwhelming PostgreSQL with large queries.

    Args:
        clinic_user_ids: List of clinic user IDs to check
    """
    if not clinic_user_ids:
        return set()

    cutoff = datetime.now() - relativedelta(years=CLINIC_RETENTION_PERIOD_IN_YEARS)
    total_batches = math.ceil(len(clinic_user_ids) / _BATCH_SIZE)
    current_logger.info(
        "checking recent sessions in batches",
        total=len(clinic_user_ids),
        total_batches=total_batches,
    )

    result: set[UUID] = set()
    for batch_idx in range(total_batches):
        start = batch_idx * _BATCH_SIZE
        batch = clinic_user_ids[start : start + _BATCH_SIZE]

        rows = (
            current_session.execute(
                select(TherapistBookingSession.clinic_user_id)
                .where(
                    TherapistBookingSession.clinic_user_id.isnot(None),
                    TherapistBookingSession.clinic_user_id.in_(batch),
                    TherapistBookingSession.ends_at > cutoff,
                )
                .distinct()
            )
            .scalars()
            .all()
        )

        result.update(rows)
        current_logger.info(
            "checked sessions batch",
            batch=batch_idx + 1,
            total_batches=total_batches,
            batch_results=len(rows),
        )

    return result

get_user_ids_ready_for_deletion

get_user_ids_ready_for_deletion()

Return inactive feature users ready for deletion, bypassing global profiles.

Same retention logic as get_profiles_ready_for_deletion but operates entirely with app user IDs, avoiding the N+1 profile→feature_user resolution.

Returns:

Type	Description
list[FeatureUser]	Feature users whose clinic data is ready for deletion.

Source code in components/clinic/public/business_logic/gdpr_compliance.py

@tracer.wrap(service="clinic")
def get_user_ids_ready_for_deletion() -> list[FeatureUser]:
    """Return inactive feature users ready for deletion, bypassing global profiles.

    Same retention logic as get_profiles_ready_for_deletion but operates entirely
    with app user IDs, avoiding the N+1 profile→feature_user resolution.

    Returns:
        Feature users whose clinic data is ready for deletion.
    """
    from components.clinic.public.dependencies import get_app_dependency

    all_candidates: list[FeatureUser] = []

    for app_name in CLINIC_COUNTRY_APP_NAMES:
        adapter = get_app_dependency(app_name.value)
        current_logger.info("fetching inactive user IDs", app_name=app_name.value)
        inactive_user_ids = adapter.get_inactive_user_ids_since(
            period_in_years=CLINIC_RETENTION_PERIOD_IN_YEARS
        )
        current_logger.info(
            "fetched inactive user IDs",
            app_name=app_name.value,
            count=len(inactive_user_ids),
        )
        all_candidates.extend(
            FeatureUser(app_id=app_name.value, app_user_id=uid)
            for uid in inactive_user_ids
        )

    current_logger.info("total candidate feature users", count=len(all_candidates))

    # batch get existing clinic users (no creation)
    current_logger.info("fetching existing clinic users")
    existing_clinic_users = get_existing_clinic_users_for_feature_users(all_candidates)
    current_logger.info(
        "fetched existing clinic users", count=len(existing_clinic_users)
    )

    feature_user_key_to_clinic_user_id: dict[tuple[str, str], UUID] = {
        (clinic_user.app_id, clinic_user.app_user_id): clinic_user.id
        for clinic_user in existing_clinic_users
    }

    candidates_without_clinic_user = len(all_candidates) - len(
        feature_user_key_to_clinic_user_id
    )
    current_logger.info(
        "candidates without clinic user", count=candidates_without_clinic_user
    )

    potentially_inactive_clinic_user_ids = [
        clinic_user_id
        for clinic_user_id in feature_user_key_to_clinic_user_id.values()
        if clinic_user_id is not None
    ]

    current_logger.info(
        "checking recent interactions",
        clinic_user_count=len(potentially_inactive_clinic_user_ids),
    )

    # union of all recent interaction sources
    active_via_conversations = (
        get_subset_of_clinic_user_ids_with_recent_medical_conversation_interactions(
            potentially_inactive_clinic_user_ids
        )
    )
    current_logger.info("active via conversations", count=len(active_via_conversations))

    active_via_sessions = (
        get_subset_of_clinic_user_ids_with_recent_session_interactions(
            potentially_inactive_clinic_user_ids
        )
    )
    current_logger.info("active via sessions", count=len(active_via_sessions))

    active_via_health_programs = (
        get_subset_of_clinic_user_ids_with_recent_health_program_interactions(
            all_candidates, feature_user_key_to_clinic_user_id
        )
    )
    current_logger.info(
        "active via health programs", count=len(active_via_health_programs)
    )

    active_clinic_user_ids = (
        active_via_conversations | active_via_sessions | active_via_health_programs
    )
    current_logger.info(
        "total active clinic users (union)", count=len(active_clinic_user_ids)
    )

    # disqualify feature users with any recent interaction
    result = [
        feature_user
        for feature_user in all_candidates
        if feature_user_key_to_clinic_user_id.get(
            (feature_user.app_id, feature_user.app_user_id)
        )
        not in active_clinic_user_ids
    ]

    current_logger.info("feature users ready for deletion", count=len(result))
    return result

insi_identity

is_identity_verified_for_user

is_identity_verified_for_user(app_user)

Check if the user's identity is verified.

Parameters:

Name	Type	Description	Default
app_user	FeatureUser	The application user whose identity is to be checked.	required

Returns: bool: True if the user's identity is verified, False otherwise.

Source code in components/clinic/public/business_logic/insi_identity.py

def is_identity_verified_for_user(app_user: FeatureUser) -> bool:
    """
    Check if the user's identity is verified.

    Args:
        app_user (FeatureUser): The application user whose identity is to be checked.
    Returns:
        bool: True if the user's identity is verified, False otherwise.
    """
    clinic_user = get_or_create_clinic_user(app_user=app_user)

    return is_identity_validated(clinic_user_id=clinic_user.id)

medical_conversation

get_first_medical_conversation_for_member

get_first_medical_conversation_for_member(app_user)

Retrieves the first medical conversation for a given member.

This function acts as a proxy to the internal business logic function get_first_medical_conversation_for_member to fetch the first medical conversation associated with the provided app_user.

Parameters:

Name	Type	Description	Default
app_user	FeatureUser	The user for whom the first medical conversation is to be retrieved.	required

Returns:

Type	Description
Optional[MedicalConversation]	Optional[MedicalConversation]: The first medical conversation for the given member, or None if no conversation is found.

Source code in components/clinic/public/business_logic/medical_conversation.py

def get_first_medical_conversation_for_member(
    app_user: FeatureUser,
) -> Optional[MedicalConversation]:
    """
    Retrieves the first medical conversation for a given member.

    This function acts as a proxy to the internal business logic function
    `get_first_medical_conversation_for_member` to fetch the first medical
    conversation associated with the provided `app_user`.

    Args:
        app_user (FeatureUser): The user for whom the first medical conversation
                                is to be retrieved.

    Returns:
        Optional[MedicalConversation]: The first medical conversation for the
                                       given member, or None if no conversation
                                       is found.
    """
    from components.clinic.internal.business_logic.medical_conversation import (
        get_first_medical_conversation_for_member,
    )

    return get_first_medical_conversation_for_member(app_user=app_user)

mark_conversation_as_deleted

mark_conversation_as_deleted(medical_conversation_id)

Source code in components/clinic/public/business_logic/medical_conversation.py

def mark_conversation_as_deleted(medical_conversation_id: UUID) -> None:  # noqa: D103
    from components.clinic.internal.business_logic.medical_chat_bot import (
        get_medical_chat_bot_clinic_user_id,
    )
    from components.clinic.internal.business_logic.medical_conversation import (
        mark_as_deleted,
    )

    mark_as_deleted(
        medical_conversation_id=medical_conversation_id,
        actor_clinic_user_id=get_medical_chat_bot_clinic_user_id(),
    )

therapist_booking_session

get_past_sessions_that_need_invoice_generation_count_for_user

get_past_sessions_that_need_invoice_generation_count_for_user(
    feature_user, session_type
)

Counts the number of past reimbursed therapist booking sessions waiting for invoicing for a specific user and session type. sessions that are waiting for invoicing are those that are not refunded, are confirmed, and have a valid credit. They ended between now and 2 days ago. (default_delay_in_hours is 48)

Arguments: feature_user (FeatureUser): The user for whom upcoming sessions need to be counted. session_type (TherapistBookingSessionType): The type of therapist session to filter by.

Returns: int: The count of done reimbursed therapist booking sessions matching the provided user and session type.

Source code in components/clinic/public/business_logic/therapist_booking_session.py

def get_past_sessions_that_need_invoice_generation_count_for_user(
    feature_user: FeatureUser, session_type: TherapistBookingSessionType
) -> int:
    """
    Counts the number of past reimbursed therapist booking sessions waiting for
    invoicing for a specific user and session type.
    sessions that are waiting for invoicing are those that are not refunded, are
    confirmed, and have a valid credit. They ended between now and 2 days ago. (default_delay_in_hours is 48)

    Arguments:
    feature_user (FeatureUser): The user for whom upcoming sessions need
    to be counted.
    session_type (TherapistBookingSessionType): The type of therapist session
    to filter by.

    Returns:
    int: The count of done reimbursed therapist booking sessions matching the
    provided user and session type.
    """
    clinic_user = get_or_create_clinic_user(app_user=feature_user)

    return int(
        get_past_sessions_that_need_invoices_query(
            clinic_user_id=clinic_user.id,
            session_type=session_type,
            min_date=(datetime.now() - timedelta(hours=DEFAULT_DELAY_IN_HOURS)),
            delay_in_hours=0,
        ).count()
    )

get_upcoming_reimbursed_therapist_booking_session_count_for_user

get_upcoming_reimbursed_therapist_booking_session_count_for_user(
    feature_user, session_type
)

Counts the number of upcoming reimbursed therapist booking sessions for a specific user and session type. Only counts sessions that are not refunded, are confirmed, have a valid credit, and are scheduled for a future date.

Arguments: feature_user (FeatureUser): The user for whom upcoming sessions need to be counted. session_type (TherapistBookingSessionType): The type of therapist session to filter by.

Returns: int: The count of upcoming reimbursed therapist booking sessions matching the provided user and session type.

Source code in components/clinic/public/business_logic/therapist_booking_session.py

def get_upcoming_reimbursed_therapist_booking_session_count_for_user(
    feature_user: FeatureUser, session_type: TherapistBookingSessionType
) -> int:
    """
    Counts the number of upcoming reimbursed therapist booking sessions for a
    specific user and session type. Only counts sessions that are not refunded,
    are confirmed, have a valid credit, and are scheduled for a future date.

    Arguments:
    feature_user (FeatureUser): The user for whom upcoming sessions need
    to be counted.
    session_type (TherapistBookingSessionType): The type of therapist session
    to filter by.

    Returns:
    int: The count of upcoming reimbursed therapist booking sessions matching
    the provided user and session type.
    """
    from components.clinic.internal.booking.models.therapist_booking_session import (
        TherapistBookingSession,
    )

    clinic_user = get_or_create_clinic_user(app_user=feature_user)

    return int(
        current_session.query(TherapistBookingSession)  # noqa: ALN085
        .filter(
            TherapistBookingSession.is_refunded == False,
            TherapistBookingSession.confirmed_at.isnot(None),
            TherapistBookingSession.clinic_user_id == clinic_user.id,
            TherapistBookingSession.session_type == session_type,
            TherapistBookingSession.starts_at > datetime.now(),
            TherapistBookingSession.therapist_booking_user_credit_id.isnot(None),
        )
        .count()
    )

user

app_user_has_upcoming_availability_for_alan_consultations

app_user_has_upcoming_availability_for_alan_consultations(
    member_app_user,
    locale=None,
    availability_threshold_in_hours=24,
    medical_admins_ids=None,
)

Returns true if the given member is able to book an Alan consultation within the following upcoming parameters: - next availability is within the next availability threshold in hours (24 hours by default)

Source code in components/clinic/public/business_logic/user.py

def app_user_has_upcoming_availability_for_alan_consultations(
    member_app_user: FeatureUser,
    locale: Optional[str] = None,
    availability_threshold_in_hours: Optional[int] = 24,
    medical_admins_ids: Optional[
        list[str]
    ] = None,  # if this argument is passed, we filter the medical admins to only choose those with an id in this list
) -> bool:
    """
    Returns true if the given member is able to book an Alan consultation within the following upcoming parameters:
    - next availability is within the next availability threshold in hours (24 hours by default)
    """
    from components.clinic.internal.business_logic.medical_admin import (
        get_medical_admins_for_member,
    )

    # general practitioners that are bookable online are the only ones that do Alan consultations
    available_medical_admins = get_medical_admins_for_member(
        app_user=member_app_user,
        is_bookable_online=True,
        locale=locale,
        medical_specialties=[MedicalAdminSpecialty.GENERAL_PRACTITIONER],
        medical_admins_ids=medical_admins_ids,
        availability_threshold_in_hours=availability_threshold_in_hours,
    )

    # since we applied the availability threshold filter, if there are no available medical admins, it means that the member is not able to book an Alan consultation within the given upcoming availability
    return len(available_medical_admins) > 0

get_clinic_user_for_french_user

get_clinic_user_for_french_user(user_id)

Source code in components/clinic/public/business_logic/user.py

def get_clinic_user_for_french_user(user_id: int) -> ClinicUser | None:  # noqa: D103
    from components.clinic.internal.business_logic.clinic_user import (
        get_clinic_user_for_french_user,
    )

    return get_clinic_user_for_french_user(user_id)

get_medical_admin_for_user

get_medical_admin_for_user(user_id, not_deleted=True)

Source code in components/clinic/public/business_logic/user.py

def get_medical_admin_for_user(  # noqa: D103
    user_id: int, not_deleted: Optional[bool] = True
) -> MedicalAdmin | None:
    from components.clinic.internal.business_logic.medical_admin import (
        get_medical_admin_by_app_user,
    )

    return get_medical_admin_by_app_user(
        medical_admin_app_user=FeatureUser(
            app_user_id=str(user_id),
            app_id=get_current_app_name(),
        ),
        not_deleted=not_deleted,
    )

has_app_user_access_to_orientation

has_app_user_access_to_orientation(app_user)

Source code in components/clinic/public/business_logic/user.py

def has_app_user_access_to_orientation(app_user: FeatureUser) -> bool:  # noqa: D103
    from components.clinic.internal.booking.business_logic.sessions_queries import (
        has_app_user_access_to_orientation,
    )

    return has_app_user_access_to_orientation(app_user=app_user)

update_clinic_consent_for_user

update_clinic_consent_for_user(
    actor_user, has_given_consent, delete_existing_data=True
)

Update the clinic user consent

Parameters:

Name	Type	Description	Default
actor_user	FeatureUser	The actor feature user	required
has_given_consent	bool	The user has given his consent	required
delete_existing_data	Optional[bool]	Has to delete existing data status	True

Source code in components/clinic/public/business_logic/user.py

def update_clinic_consent_for_user(
    actor_user: FeatureUser,
    has_given_consent: bool,
    delete_existing_data: Optional[bool] = True,
) -> None:
    """
    Update the clinic user consent

    Args:
        actor_user: The actor feature user
        has_given_consent: The user has given his consent
        delete_existing_data: Has to delete existing data status
    """
    from components.clinic.internal.business_logic.clinic_user import (
        update_clinic_consent,
    )

    update_clinic_consent(
        actor_app_user=actor_user,
        has_given_consent=has_given_consent,
        delete_existing_data=delete_existing_data,
    )

validate_child_dependent

validate_child_dependent_and_get_feature_user

validate_child_dependent_and_get_feature_user(
    parent_user, child_app_user_id
)

Validate that a child is a dependent of the current user and return a FeatureUser for the child.

Parameters:

Name	Type	Description	Default
parent_user	FeatureUser	The current user (parent)	required
child_app_user_id	str	The ID of the child to validate	required

Source code in components/clinic/public/business_logic/validate_child_dependent.py

def validate_child_dependent_and_get_feature_user(
    parent_user: FeatureUser, child_app_user_id: str
) -> FeatureUser:
    """
    Validate that a child is a dependent of the current user and return a FeatureUser for the child.

    Args:
        parent_user: The current user (parent)
        child_app_user_id: The ID of the child to validate

    Raises:
        BaseErrorCode.forbidden if user has no dependents or child is not a dependent of current user
    """
    # Get the app user data of the current user
    user_data = get_app_dependency(parent_user.app_id).get_app_user_data(
        app_user_id=parent_user.app_user_id
    )

    if not user_data.dependents:
        raise BaseErrorCode.forbidden(message="User has no dependents")

    # Get the list of dependent app_user_ids
    dependent_ids = [
        dependent.app_user_id
        for dependent in user_data.dependents
        if dependent.app_user_id
    ]

    if child_app_user_id not in dependent_ids:
        raise BaseErrorCode.forbidden(
            message="Child is not a dependent of the current user"
        )

    # Create and return feature user for the child
    return FeatureUser(
        app_user_id=child_app_user_id,
        app_id=parent_user.app_id,
    )

components.clinic.public.commands

app_group

clinic_commands module-attribute

clinic_commands = AppGroup('clinic')

register_clinic_commands

register_clinic_commands()

Source code in components/clinic/public/commands/app_group.py

def register_clinic_commands() -> None:  # noqa: D103
    from components.clinic.public.commands import (  # noqa: F401
        clinic_invoices,
        end_to_end_test,
        gdpr_compliance,
        medical_admins,
        medical_conversation,
        setup,
        unread_message_reminders,
        user,
    )

booking

app_group

booking_commands module-attribute

booking_commands = AppGroup('booking')

register_booking_commands

register_booking_commands()

Source code in components/clinic/public/commands/booking/app_group.py

def register_booking_commands() -> None:  # noqa: D103
    from components.clinic.public.commands.booking import (  # noqa: F401
        notifications,
        sessions,
        therapists,
    )

notifications

notify_all_scheduled_messages

notify_all_scheduled_messages()

Source code in components/clinic/public/commands/booking/notifications.py

@booking_commands.command()
@command_with_dry_run
def notify_all_scheduled_messages() -> None:  # noqa: D103
    from components.clinic.internal.booking.business_logic.notifications import (
        notify_all_scheduled_messages,
    )

    notify_all_scheduled_messages()

sessions

backfill_missing_session_user_lang

backfill_missing_session_user_lang(dry_run)

Backfill user_lang on sessions where it's missing, using each user's current app lang.

Source code in components/clinic/public/commands/booking/sessions.py

@booking_commands.command()
@command_with_dry_run
def backfill_missing_session_user_lang(dry_run: bool) -> None:
    """Backfill user_lang on sessions where it's missing, using each user's current app lang."""
    from components.clinic.internal.booking.business_logic.sessions_actions import (
        backfill_missing_session_user_lang,
    )

    backfill_missing_session_user_lang(dry_run=dry_run)

cancel_all_future_sessions_for_medical_admin_as_medical_admin

cancel_all_future_sessions_for_medical_admin_as_medical_admin(
    dry_run, medical_admin_id
)

Cancels all future sessions for a given medical admin as if the same medical admin is cancelling them themselves. This is useful when a medical admin is departing Alan, and we want all the members with booked sessions to receive the appropriate notifications for the cancellation.

Source code in components/clinic/public/commands/booking/sessions.py

@booking_commands.command()
@click.option(
    "--medical-admin-id",
    type=str,
    required=True,
)
@command_with_dry_run
def cancel_all_future_sessions_for_medical_admin_as_medical_admin(
    dry_run: bool,
    medical_admin_id: str,
) -> None:
    """Cancels all future sessions for a given medical admin as if the same medical admin is cancelling them themselves.
    This is useful when a medical admin is departing Alan, and we want all the members with booked sessions to receive the appropriate notifications for the cancellation.
    """
    from components.clinic.internal.booking.business_logic.sessions_actions import (
        cancel_session_for_expert,
    )
    from components.clinic.internal.booking.business_logic.sessions_queries import (
        get_therapist_upcoming_sessions,
    )
    from components.clinic.internal.models.medical_admin import MedicalAdmin
    from shared.helpers.get_or_else import get_or_raise_missing_resource
    from shared.helpers.logging.logger import current_logger

    # get the feature user for the medical admin
    medical_admin = get_or_raise_missing_resource(MedicalAdmin, medical_admin_id)
    medical_admin_feature_user = medical_admin.clinic_user.feature_user

    # get all the sessions for the medical admin
    sessions = get_therapist_upcoming_sessions(
        actor_app_user=medical_admin_feature_user,
    )

    current_logger.info(
        f"Found {len(sessions)} upcoming sessions for medical admin {medical_admin_id}"
    )

    # cancel each session as the medical admin
    for num_cancelled, session in enumerate(sessions, 1):
        if dry_run:
            current_logger.info(f"DRY RUN: Would have cancelled session {session.id}")
        else:
            current_logger.info(f"Cancelling session {session.id}")

        cancel_session_for_expert(
            session_id=session.id,
            actor_app_user=medical_admin_feature_user,
            dry_run=dry_run,
        )
        if num_cancelled % 5 == 0:
            current_logger.info(f"Cancelled {num_cancelled}/{len(sessions)} sessions")

    current_logger.info(f"Completed. Cancelled {len(sessions)} sessions")

delete_test_sessions

delete_test_sessions()

Source code in components/clinic/public/commands/booking/sessions.py

@booking_commands.command()
@do_not_run_in_prod
def delete_test_sessions() -> None:  # noqa: D103
    from components.clinic.internal.booking.business_logic.end_to_end_test import (
        delete_test_sessions,
    )

    delete_test_sessions()

expire_booking_invites

expire_booking_invites(dry_run)

Source code in components/clinic/public/commands/booking/sessions.py

@booking_commands.command()
@command_with_dry_run
def expire_booking_invites(dry_run: bool) -> None:  # noqa: D103
    from components.clinic.internal.booking.business_logic.sessions_actions import (
        cancel_all_expired_invites,
    )

    cancel_all_expired_invites(dry_run=dry_run)

generate_past_sessions_invoice

generate_past_sessions_invoice(dry_run=False)

Source code in components/clinic/public/commands/booking/sessions.py

@booking_commands.command()
@command_with_dry_run
def generate_past_sessions_invoice(dry_run: bool = False) -> None:  # noqa: D103
    from components.clinic.internal.booking.business_logic.sessions_invoicing import (
        generate_past_sessions_invoice,
    )

    generate_past_sessions_invoice(dry_run=dry_run)

generate_sessions_invoices

generate_sessions_invoices(dry_run)

Source code in components/clinic/public/commands/booking/sessions.py

@booking_commands.command()
@command_with_dry_run
def generate_sessions_invoices(dry_run: bool) -> None:  # noqa: D103
    from components.clinic.internal.booking.business_logic.sessions_invoicing import (
        generate_sessions_invoices,
    )

    generate_sessions_invoices(dry_run=dry_run)

push_metadata_to_doctor_ai_for_recent_ended_sessions_booked_from_mo

push_metadata_to_doctor_ai_for_recent_ended_sessions_booked_from_mo(
    dry_run,
)

Push the metadata of recent ended sessions (<48 hours ago) booked from Mo to Doctor AI.

Source code in components/clinic/public/commands/booking/sessions.py

@booking_commands.command()
@command_with_dry_run
def push_metadata_to_doctor_ai_for_recent_ended_sessions_booked_from_mo(
    dry_run: bool,
) -> None:
    """Push the metadata of recent ended sessions (<48 hours ago) booked from Mo to Doctor AI."""
    from components.clinic.internal.booking.business_logic.events_subscribers import (
        track_lifecycle_event_for_session_from_mo_redirection,
    )
    from components.clinic.internal.booking.business_logic.sessions_queries import (
        get_recent_ended_sessions_booked_from_mo,
    )
    from components.clinic.internal.booking.entities.events import (
        SessionRecommendedByMoWasUpdated,
    )
    from shared.helpers.logging.logger import current_logger

    sessions = get_recent_ended_sessions_booked_from_mo()

    for session in sessions:
        if dry_run:
            current_logger.info(
                f"DRY RUN: Would have pushed metadata to Doctor AI for session {session.id} with prescription types {session.prescription_types}"
            )
            continue

        # We can afford running the event synchronously here as we are not a user facing process
        # and operate on a small batch (max 10^2)
        track_lifecycle_event_for_session_from_mo_redirection(
            event=SessionRecommendedByMoWasUpdated(
                session_id=session.id,
                metadata={
                    "status": "did_happen",
                    "prescription_types": session.prescription_types,
                },
            )
        )

refund_payment_not_linked_to_session

refund_payment_not_linked_to_session(dry_run)

Source code in components/clinic/public/commands/booking/sessions.py

@booking_commands.command()
@command_with_dry_run
def refund_payment_not_linked_to_session(dry_run: bool) -> None:  # noqa: D103
    from components.clinic.internal.booking.business_logic.credits import (
        refund_payment_not_linked_to_session,
    )

    refund_payment_not_linked_to_session(dry_run=dry_run)

therapists

update_cached_next_availability

update_cached_next_availability(dry_run)

Source code in components/clinic/public/commands/booking/therapists.py

@booking_commands.command()
@command_with_dry_run
def update_cached_next_availability(dry_run: bool) -> None:  # noqa: D103
    from components.clinic.internal.booking.business_logic.therapists import (
        update_cached_next_availability_of_all_therapists,
    )

    update_cached_next_availability_of_all_therapists(commit=not dry_run)

clinic_invoices

generate_fake_clinic_invoices

generate_fake_clinic_invoices(dry_run)

Source code in components/clinic/public/commands/clinic_invoices.py

@clinic_commands.command()
@command_with_dry_run
@do_not_run_in_prod
def generate_fake_clinic_invoices(dry_run: bool) -> None:  # noqa: D103
    from components.clinic.internal.business_logic.clinic_invoice import (
        generate_fake_clinic_invoices,
    )

    generate_fake_clinic_invoices(dry_run=dry_run)

sync_clinic_monthly_invoice_lines_from_turing

sync_clinic_monthly_invoice_lines_from_turing(
    dry_run, date, previous_month, mapped_ids
)

Source code in components/clinic/public/commands/clinic_invoices.py

@clinic_commands.command()
@command_with_dry_run
@click.option(
    "--date",
    type=click.DateTime(formats=["%Y-%m-%d"]),
    required=False,
    help="Date in YYYY-MM-DD format (e.g., 2021-09-01)",
)
@click.option(
    "--previous-month",
    is_flag=True,
    default=False,
    help="Run invoice generation for the previous month",
)
@click.option(
    "--mapped_ids",
    type=str,
    required=False,
    help="Optional mapping of origin IDs to target IDs in the format 'originid1:targetid1,originid2:targetid2'. If provided, only the specified origin IDs will be queried from Turing, and the corresponding target IDs will be used when inserting into the backend database.",
)
def sync_clinic_monthly_invoice_lines_from_turing(  # noqa: D103
    dry_run: bool,
    date: Optional[datetime.date],
    previous_month: bool,
    mapped_ids: Optional[str],
) -> None:
    from components.clinic.internal.business_logic.clinic_invoice import (
        sync_clinic_monthly_invoice_lines_from_turing,
    )

    if date and previous_month:
        raise click.BadParameter(
            "You can't specify both --date and --previous-month options"
        )

    if previous_month:
        date = (
            datetime.date.today().replace(day=1) - datetime.timedelta(days=1)
        ).replace(day=1)
        click.echo(f"Invoicing date: {date}")

    result = sync_clinic_monthly_invoice_lines_from_turing(
        dry_run=dry_run,
        date=date,
        mapped_ids=mapped_ids,
    )

    if not dry_run:
        current_app.slack_web_client.chat_postMessage(  # type: ignore[attr-defined]
            channel=(
                SlackChannel.alan_clinic_ops_channel
                if is_production_mode()
                else SlackChannel.test
            ),
            username="Clinic Bot",
            icon_emoji=":moneybag:",
            text=(
                f"👋 I just finished generating {result.created_invoices_count} invoices (for the {result.invoicing_date} date)."
                "\nHead over <https://alan.com/marmot-v2/alan-clinic/invoices|the billing page> in Marmot and send the invoices for validation 🍔"
                + (
                    f"\nnote: {result.removed_invoices_count} draft invoice(s) were first removed when generating today's invoices."
                    if result.removed_invoices_count
                    else ""
                )
            ),
        )

end_to_end_test

delete_test_conversations

delete_test_conversations()

Source code in components/clinic/public/commands/end_to_end_test.py

@clinic_commands.command()
@do_not_run_in_prod
def delete_test_conversations() -> None:  # noqa: D103
    from components.clinic.internal.business_logic.end_to_end_test import (
        delete_test_conversations,
    )

    delete_test_conversations()

keep_last_part_from_test_conversations

keep_last_part_from_test_conversations()

Source code in components/clinic/public/commands/end_to_end_test.py

@clinic_commands.command()
@do_not_run_in_prod
def keep_last_part_from_test_conversations() -> None:  # noqa: D103
    from components.clinic.internal.business_logic.end_to_end_test import (
        keep_last_part_from_test_conversations,
    )

    keep_last_part_from_test_conversations()

gdpr_compliance

get_user_ids_ready_for_deletion

get_user_ids_ready_for_deletion()

Return feature users whose clinic data is ready for GDPR deletion.

Source code in components/clinic/public/commands/gdpr_compliance.py

@clinic_commands.command()
def get_user_ids_ready_for_deletion() -> None:
    """Return feature users whose clinic data is ready for GDPR deletion."""
    from components.clinic.public.business_logic.gdpr_compliance import (
        get_user_ids_ready_for_deletion as _get_user_ids_ready_for_deletion,
    )

    feature_users = _get_user_ids_ready_for_deletion()

    current_logger.info(
        "user IDs ready for deletion",
        count=len(feature_users),
    )

medical_admins

backfill_medical_admin_synced_ans_data

backfill_medical_admin_synced_ans_data(dry_run)

This temporary command creates ANS data for medical admins that have a medical_identifier but don't have associated ANS data yet.

TODO: @sarah.louahem Remove this command once run in prod - we now create ANS data automatically.

Source code in components/clinic/public/commands/medical_admins.py

@clinic_commands.command()
@command_with_dry_run
def backfill_medical_admin_synced_ans_data(dry_run: bool) -> None:
    """
    This temporary command creates ANS data for medical admins that have
    a medical_identifier but don't have associated ANS data yet.

    TODO: @sarah.louahem Remove this command once run in prod - we now create ANS data automatically.
    """
    from components.clinic.internal.business_logic.medical_admin_ans import (
        backfill_medical_admin_synced_ans_data as backfill_ans_data,
    )

    click.echo(f"Starting backfill of medical admin ANS data with dry_run={dry_run}")

    try:
        backfill_ans_data(dry_run)
        click.echo(f"Backfill completed successfully with dry_run={dry_run}")
    except Exception as error:
        click.echo(
            f"Error during backfill: {error}",
            err=True,
        )
        raise

create_medical_admins

create_medical_admins(password, dry_run)

Source code in components/clinic/public/commands/medical_admins.py

@clinic_commands.command()
@click.option(
    "--password", help="Password for all the accounts", type=str, default="azerty"
)
@command_with_dry_run
def create_medical_admins(password: str, dry_run: bool) -> None:  # noqa: D103
    _create_medical_admins(password=password, dry_run=dry_run)

create_medical_chat_bot

create_medical_chat_bot()

Source code in components/clinic/public/commands/medical_admins.py

@clinic_commands.command()
def create_medical_chat_bot() -> None:  # noqa: D103
    from components.clinic.internal.business_logic.medical_chat_bot import (
        create_medical_chat_bot_clinic_user,
        medical_chat_bot_clinic_user_exists,
    )

    if not is_development_mode():
        click.echo("This command is only available in development mode!")
        return

    if medical_chat_bot_clinic_user_exists():
        click.echo("The Medical Chat Bot already exists.")
    else:
        create_medical_chat_bot_clinic_user()

set_offline_inactive_medical_admin

set_offline_inactive_medical_admin(minutes, dry_run)

Source code in components/clinic/public/commands/medical_admins.py

@clinic_commands.command()
@click.option(
    "--minutes",
    type=int,
    default=30,
    help="Number of minutes before medical admin is inactive",
)
@command_with_dry_run
def set_offline_inactive_medical_admin(minutes: int, dry_run: bool) -> None:  # noqa: D103
    from components.clinic.internal.business_logic.medical_admin import (
        set_offline_inactive_medical_admins,
    )

    set_offline_inactive_medical_admins(
        minutes=minutes,
        dry_run=dry_run,
    )

sync_medical_admin_ans_data

sync_medical_admin_ans_data(dry_run)

Sync medical admin ANS data with the ANS FHIR API.

This command checks all stored medical admin ANS data and updates records that have older versions than what's available in the ANS FHIR API.

Source code in components/clinic/public/commands/medical_admins.py

@clinic_commands.command()
@command_with_dry_run
def sync_medical_admin_ans_data(dry_run: bool) -> None:
    """
    Sync medical admin ANS data with the ANS FHIR API.

    This command checks all stored medical admin ANS data and updates records
    that have older versions than what's available in the ANS FHIR API.
    """
    from components.clinic.internal.business_logic.medical_admin_ans import (
        sync_outdated_medical_admin_ans_data,
    )

    click.echo(f"Starting sync of medical admin ANS data with dry_run={dry_run}")

    try:
        sync_outdated_medical_admin_ans_data(dry_run)
        click.echo(f"Sync completed successfully with dry_run={dry_run}")
    except Exception as error:
        click.echo(
            f"Error during sync: {error}",
            err=True,
        )
        raise

medical_admins_dataset

MedicalAdminData dataclass

MedicalAdminData(
    first_name,
    last_name,
    onboarding_status,
    accessible_conversation_specialties,
    apps_displayed_in,
    has_access_to_app_ids,
    access_types,
    specialty,
    country,
    languages,
    description=None,
    experiences=None,
    answers_to_proactive_conversation_topics=None,
    prod_id=None,
    dato_id=None,
    first_message_body=None,
    avatar=None,
    clinic_role=None,
)

access_types instance-attribute

access_types

accessible_conversation_specialties instance-attribute

accessible_conversation_specialties

answers_to_proactive_conversation_topics class-attribute instance-attribute

answers_to_proactive_conversation_topics = None

apps_displayed_in instance-attribute

apps_displayed_in

avatar class-attribute instance-attribute

avatar = None

clinic_role class-attribute instance-attribute

clinic_role = None

country instance-attribute

country

dato_id class-attribute instance-attribute

dato_id = None

description class-attribute instance-attribute

description = None

experiences class-attribute instance-attribute

experiences = None

first_message_body class-attribute instance-attribute

first_message_body = None

first_name instance-attribute

first_name

has_access_to_app_ids instance-attribute

has_access_to_app_ids

languages instance-attribute

languages

last_name instance-attribute

last_name

onboarding_status instance-attribute

onboarding_status

prod_id class-attribute instance-attribute

prod_id = None

specialty instance-attribute

specialty

MedicalAdminExperienceData dataclass

MedicalAdminExperienceData(title, subtitle, index)

index instance-attribute

index

subtitle instance-attribute

subtitle

title instance-attribute

title

RAW_MEDICAL_ADMINS module-attribute

RAW_MEDICAL_ADMINS = [
    MedicalAdminData(
        first_name="Augustin",
        last_name="Beaucote",
        specialty=GENERAL_PRACTITIONER,
        avatar="http://eu.alan.uploads.s3.eu-central-1.amazonaws.com/manual_upload/9d775ac221a84812a3cc18bae63b7766_doctor_beaucote_headshot.jpg",
        description="Je suis médecin généraliste et immunologue et j’ai rejoint Alan début 2020 pour développer les services de santé.\nJ’ai fait mes études en région parisienne à la faculté du Kremlin Bicêtre avant de rejoindre la région de Montpellier en 2015, où j’ai été interne pendant 3 ans.\nJ’ai choisi de me spécialiser en immunologie et j’ai fait des remplacements en cabinet de médecine générale et aux urgences adultes et pédiatriques.",
        experiences=[
            MedicalAdminExperienceData(
                title="Immunothérapies ciblées des maladies",
                subtitle="Faculté de médecine de Montpellier",
                index=3,
            ),
            MedicalAdminExperienceData(
                title="Master 2, Génome et différenciation cellulaire - Hématopoïèse",
                subtitle="Université Paris Diderot et IGMM",
                index=2,
            ),
            MedicalAdminExperienceData(
                title="DESC 2 d'immunologie clinique",
                subtitle="Université de Montpellier",
                index=1,
            ),
            MedicalAdminExperienceData(
                title="Doctorat en Médecine Générale",
                subtitle="Faculté de médecine de Montpellier",
                index=0,
            ),
        ],
        apps_displayed_in=[alan_insurance],
        accessible_conversation_specialties=[
            GENERAL_MEDICINE,
            PEDIATRICS,
            CHILDCARE,
            DIETETICS,
            DERMATOLOGY,
            PHYSIOTHERAPY,
            GYNECOLOGY,
        ],
        clinic_role=CLINIC_ADMIN,
        has_access_to_app_ids=[ALAN_FR],
        access_types=[CHAT],
        answers_to_proactive_conversation_topics=[
            SLEEP,
            STRESS,
        ],
        country=france,
        languages=[french, english],
        onboarding_status=COMPLETED,
    ),
    MedicalAdminData(
        first_name="Clara",
        last_name="Poncelet",
        specialty=MIDWIFE,
        first_message_body="Je suis Clara, je suis sage-femme, j’exerce en Suisse dans une structure hospitalière et en France comme sage-femme libérale. Je suis aussi maman d’une petite fille depuis fin 2020 !",
        avatar="http://eu.alan.uploads.s3.eu-central-1.amazonaws.com/manual_upload/3fec273f3a484072b50029a7ee7a0440_clara_poncelet.jpg",
        description="Je suis sage-femme depuis 2015 et heureuse nouvelle arrivée chez Alan.\nCes cinq dernières années sont marquées par mes voyages régionaux car j’ai choisi d’exercer dans différents grands centres hospitaliers afin d’enrichir ma formation.\nDepuis 2019 je fais des remplacements en libéral pour un accompagnement semi-global et j’exerce en Suisse dans une structure hospitalière. Je me forme dans les domaines ouverts aux sages-femmes dès que j’en ai l’occasion pour adapter mes pratiques.\nJe suis maman d’une petite fille depuis fin 2020.",
        experiences=[
            MedicalAdminExperienceData(
                title="Diplôme de sage-femme en 2015",
                subtitle="Université de Bourgogne à Dijon",
                index=0,
            )
        ],
        accessible_conversation_specialties=[GYNECOLOGY],
        apps_displayed_in=[alan_insurance],
        has_access_to_app_ids=[ALAN_FR],
        access_types=[CHAT],
        country=france,
        languages=[french, english],
        onboarding_status=COMPLETED,
    ),
    MedicalAdminData(
        first_name="Barbara",
        last_name="Dezileaux",
        specialty=GENERAL_PRACTITIONER,
        avatar="http://eu.alan.uploads.s3.eu-central-1.amazonaws.com/manual_upload/8b28ef4ffbea476c93565ad2d3cd4eb7_doctor_dezileaux_headshot.jpg",
        description="Médecin généraliste, j’ai choisi de rejoindre les équipes d’Alan car j’apprécie leur démarche d’aide et de conseil envers leurs utilisateurs.\nJe suis diplômée depuis 2015, après avoir réalisé mes études à Bordeaux puis mon internat à Lille.\nJ’exerce depuis en cabinet libéral et en clinique dans différentes régions de France.\nJ’ai à cœur de me former en continu, en particulier sur les sujets de la maternité et de la périnatalité qui me passionnent.",
        experiences=[
            MedicalAdminExperienceData(
                title="Doctorat en Médecine Générale",
                subtitle="Faculté de médecine de Lille",
                index=0,
            )
        ],
        accessible_conversation_specialties=[
            GENERAL_MEDICINE,
            PEDIATRICS,
            CHILDCARE,
            DIETETICS,
            DERMATOLOGY,
            PHYSIOTHERAPY,
            GYNECOLOGY,
        ],
        apps_displayed_in=[alan_insurance],
        access_types=[CHAT],
        has_access_to_app_ids=[ALAN_FR],
        country=france,
        languages=[french, english],
        onboarding_status=COMPLETED,
    ),
    MedicalAdminData(
        first_name="Marion",
        last_name="Cosson",
        specialty=GENERAL_PRACTITIONER,
        clinic_role=CLINIC_ADMIN,
        apps_displayed_in=[],
        has_access_to_app_ids=[ALAN_FR],
        accessible_conversation_specialties=[
            GENERAL_MEDICINE,
            PEDIATRICS,
            CHILDCARE,
            DIETETICS,
            DERMATOLOGY,
            PHYSIOTHERAPY,
            GYNECOLOGY,
        ],
        access_types=[CHAT],
        country=france,
        languages=[french, english, spanish],
        onboarding_status=COMPLETED,
    ),
    MedicalAdminData(
        first_name="Pauline",
        last_name="Lotte",
        specialty=CHILDCARE_NURSE,
        avatar="http://eu.alan.uploads.s3.eu-central-1.amazonaws.com/manual_upload/9a0297eaf92f4f88a763c9fe3c954983_pauline-lotte.jpg",
        description="Je suis infirmière puéricultrice depuis 2007 et installée en libéral depuis 2018. J'interviens chez Alan en tant que spécialiste du sommeil.\nAprès 10 ans de pratique en hôpital pédiatrique et 1 an en direction de crèche, j'ai choisi de travailler en indépendante pour accompagner les parents au plus près de leur quotidien avec bébé.\nJe me suis spécialisée dans plusieurs domaines, et plus spécialement le sommeil du tout-petit.\nJe suis maman d'une petite fille depuis 2019.",
        experiences=[
            MedicalAdminExperienceData(
                title="2006 - Diplôme d'Etat d'Infirmière",
                subtitle="IFSI de Reims",
                index=10,
            ),
            MedicalAdminExperienceData(
                title="2007 - Diplôme d'Etat de Puéricultrice",
                subtitle="Ecole de puéricultrice de Reims",
                index=9,
            ),
            MedicalAdminExperienceData(
                title="2013 - Diplôme Universitaire de prise en charge de la douleur de l'enfant",
                subtitle=None,
                index=8,
            ),
            MedicalAdminExperienceData(
                title="2019 - Instructrice Dunstan Baby Langage",
                subtitle="Dunstan Baby Langage France",
                index=7,
            ),
            MedicalAdminExperienceData(
                title="2019 - Educateur Montessori",
                subtitle="Enfance Positive",
                index=6,
            ),
            MedicalAdminExperienceData(
                title="2020 - Animateur de Signes Associés à la Parole",
                subtitle="Eveil et Signes",
                index=5,
            ),
            MedicalAdminExperienceData(
                title="2020 - Formation Nutrition pédiatrique",
                subtitle="EPM nutrition",
                index=4,
            ),
            MedicalAdminExperienceData(
                title="2020 - Formation au sommeil du bébé de 0 à 5 ans",
                subtitle="M. Bilodeau",
                index=3,
            ),
            MedicalAdminExperienceData(
                title="2020 - Comprendre et accompagner le sommeil de l'enfant",
                subtitle="Mandy Roman",
                index=2,
            ),
            MedicalAdminExperienceData(
                title="2021 - Formation consultation du sommeil, alimentation et rythmes de 0 à 6 ans en avril 2021",
                subtitle="Prosom",
                index=1,
            ),
            MedicalAdminExperienceData(
                title="2021 - Formation sommeil du tout petit et accompagnement parental",
                subtitle="Ingrid Bayot",
                index=0,
            ),
        ],
        apps_displayed_in=[alan_insurance],
        accessible_conversation_specialties=[CHILDCARE],
        has_access_to_app_ids=[ALAN_FR],
        access_types=[CHAT],
        country=france,
        languages=[dutch, english],
        onboarding_status=COMPLETED,
    ),
    MedicalAdminData(
        first_name="Margaux",
        last_name="Degen",
        specialty=PSYCHOLOGIST,
        accessible_conversation_specialties=[
            PSYCHOLOGY,
            THERAPY,
        ],
        apps_displayed_in=[alan_insurance],
        has_access_to_app_ids=[ALAN_FR],
        access_types=[VIDEO],
        country=france,
        languages=[french, english],
        onboarding_status=COMPLETED,
        dato_id="148187763",
    ),
    MedicalAdminData(
        first_name="Emile",
        last_name="Montrois",
        specialty=PSYCHOLOGIST,
        accessible_conversation_specialties=[
            PSYCHOLOGY,
            THERAPY,
        ],
        access_types=[VIDEO],
        apps_displayed_in=[alan_insurance],
        has_access_to_app_ids=[ALAN_FR],
        country=france,
        languages=[french, english],
        onboarding_status=COMPLETED,
        dato_id="148184344",
    ),
    MedicalAdminData(
        first_name="Florian",
        last_name="Ghiazza",
        specialty=PHYSIOTHERAPIST,
        accessible_conversation_specialties=[
            PHYSIOTHERAPY,
            BACK_PAIN,
            EMERGENCY_PHYSIOTHERAPY,
        ],
        access_types=[CHAT, VIDEO],
        apps_displayed_in=[alan_insurance],
        has_access_to_app_ids=[ALAN_FR],
        country=france,
        languages=[french, english],
        onboarding_status=COMPLETED,
        dato_id="OTiPilFASx-0yBrP2HtLrQ",
        prod_id="b65af61b-c455-4c5f-a678-164d165d1f45",
    ),
    MedicalAdminData(
        first_name="Diego",
        last_name="Ferral-Toro",
        specialty=PHYSIOTHERAPIST,
        accessible_conversation_specialties=[
            PHYSIOTHERAPY,
            BACK_PAIN,
            EMERGENCY_PHYSIOTHERAPY,
        ],
        access_types=[CHAT, VIDEO],
        apps_displayed_in=[alan_insurance],
        has_access_to_app_ids=[ALAN_ES],
        country=spain,
        languages=[spanish, english],
        onboarding_status=COMPLETED,
        dato_id="XSdL4ehETM-owb_20xEmhg",
        prod_id="18eb92d9-3271-4286-91dc-a4e1a8ac5de2",
    ),
    MedicalAdminData(
        first_name="Diane",
        last_name="Coomans",
        specialty=PHYSIOTHERAPIST,
        accessible_conversation_specialties=[
            PHYSIOTHERAPY,
            BACK_PAIN,
            EMERGENCY_PHYSIOTHERAPY,
        ],
        access_types=[CHAT, VIDEO],
        apps_displayed_in=[alan_insurance],
        has_access_to_app_ids=[ALAN_BE],
        country=belgium,
        languages=[dutch, english],
        onboarding_status=COMPLETED,
        dato_id="OXi_GJK7TziPCWlkkFWQCw",
        prod_id="59ce8029-2aa1-4ff5-81b1-25156ebcf5f1",
    ),
    MedicalAdminData(
        first_name="Robin",
        last_name="Vervaeke",
        specialty=PHYSIOTHERAPIST,
        accessible_conversation_specialties=[
            PHYSIOTHERAPY,
            BACK_PAIN,
            EMERGENCY_PHYSIOTHERAPY,
        ],
        access_types=[CHAT, VIDEO],
        apps_displayed_in=[alan_insurance],
        has_access_to_app_ids=[ALAN_FR, ALAN_ES, ALAN_BE],
        country=france,
        languages=[french, english],
        onboarding_status=COMPLETED,
        dato_id="R03pRAdAR1-Hq7_B4A0pAw",
        prod_id="59584887-47a7-464d-86c7-4f1a15299eaa",
    ),
    MedicalAdminData(
        first_name="Olivier",
        last_name="Delarras",
        specialty=GENERAL_PRACTITIONER,
        accessible_conversation_specialties=[
            GENERAL_MEDICINE,
            CONSULTATION,
            PEDIATRICS,
            CHILDCARE,
            DIETETICS,
            DERMATOLOGY,
            PHYSIOTHERAPY,
            GYNECOLOGY,
        ],
        access_types=[CHAT, VIDEO],
        apps_displayed_in=[alan_insurance],
        has_access_to_app_ids=[ALAN_FR, ALAN_ES, ALAN_BE],
        country=france,
        languages=[french, english],
        onboarding_status=COMPLETED,
        dato_id="RunHY9MqRXOl4ti7fuhoTw",
        prod_id="3c7c88e9-0794-4d23-9a32-002e75ca1f60",
    ),
]

medical_conversation

close_old_answered_medical_conversations

close_old_answered_medical_conversations(
    age_for_gp, age_for_non_gp, dry_run
)

Source code in components/clinic/public/commands/medical_conversation.py

@clinic_commands.command()
@command_with_dry_run
@click.option(
    "--age-for-gp", type=int, help="Minimum age in days to close gp conversations"
)
@click.option(
    "--age-for-non-gp",
    type=int,
    help="Minimum age in days to close non gp conversations",
)
def close_old_answered_medical_conversations(  # noqa: D103
    age_for_gp: int,
    age_for_non_gp: int,
    dry_run: bool,
) -> None:
    from components.clinic.internal.business_logic.medical_conversation import (
        close_old_answered_medical_conversations,
    )

    close_old_answered_medical_conversations(
        age_for_gp=age_for_gp,
        age_for_non_gp=age_for_non_gp,
        dry_run=dry_run,
    )

create_fake_conversations

create_fake_conversations(count, specialty, app, is_closed)

Create fake conversations for testing purposes only - count: number of fake conversations to create - specialty: medical specialty of the fake conversations to create - app: app of the fake conversations to create - is_closed: whether the fake conversations should be closed or not

Source code in components/clinic/public/commands/medical_conversation.py

@clinic_commands.command()
@click.option(
    "-c", "--count", default=50, help="Number of fake conversations to create"
)
@click.option(
    "-s",
    "--specialty",
    default=None,
    help="Medical specialty of fake conversations to create",
)
@click.option(
    "-a",
    "--app",
    default=None,
    help="App of fake conversations to create",
)
@click.option(
    "--is-closed",
    type=bool,
    default=False,
    help="App of fake conversations to create",
)
@do_not_run_in_prod
def create_fake_conversations(
    count: int, specialty: str, app: str, is_closed: bool
) -> None:
    """
    Create fake conversations for testing purposes only
    - count: number of fake conversations to create
    - specialty: medical specialty of the fake conversations to create
    - app: app of the fake conversations to create
    - is_closed: whether the fake conversations should be closed or not
    """
    _create_fake_conversations(
        count=count, specialty=specialty, app=app, is_closed=is_closed
    )

erase_all_conversation_data

erase_all_conversation_data()

Source code in components/clinic/public/commands/medical_conversation.py

@clinic_commands.command()
@do_not_run_in_prod
def erase_all_conversation_data() -> None:  # noqa: D103
    from components.clinic.internal.models.medical_conversation import (
        MedicalConversation,
    )
    from components.clinic.internal.models.medical_conversation_ai_suggestions_result import (
        MedicalConversationAiSuggestionsResult,
    )
    from components.clinic.internal.models.medical_conversation_attachment import (
        MedicalConversationAttachment,
    )
    from components.clinic.internal.models.medical_conversation_bookmark import (
        MedicalConversationBookmark,
    )
    from components.clinic.internal.models.medical_conversation_follow_up import (
        MedicalConversationFollowUp,
    )
    from components.clinic.internal.models.medical_conversation_part import (
        MedicalConversationPart,
    )
    from components.clinic.internal.models.medical_conversation_rating_v2 import (
        MedicalConversationRatingV2,
    )
    from components.clinic.internal.models.medical_conversation_redirection import (
        MedicalConversationRedirection,
    )
    from components.clinic.internal.models.medical_conversation_redirection_medical_conversation_part import (
        MedicalConversationRedirectionMedicalConversationPart,
    )
    from components.clinic.internal.models.medical_conversation_review_from_medical_admin import (
        MedicalConversationReviewFromMedicalAdmin,
    )
    from components.clinic.internal.models.medical_conversation_tag_mapping import (
        MedicalConversationTagMapping,
    )

    if not click.confirm("This will erase all conversation data, are you sure?"):
        return

    current_session.query(MedicalConversationAiSuggestionsResult).delete()  # noqa: ALN085
    current_session.query(MedicalConversationReviewFromMedicalAdmin).delete()  # noqa: ALN085
    current_session.query(  # noqa: ALN085
        MedicalConversationRedirectionMedicalConversationPart
    ).delete()
    current_session.query(MedicalConversationRedirection).delete()  # noqa: ALN085
    current_session.query(MedicalConversationBookmark).delete()  # noqa: ALN085
    current_session.query(MedicalConversationFollowUp).delete()  # noqa: ALN085
    current_session.query(MedicalConversationTagMapping).delete()  # noqa: ALN085
    current_session.query(MedicalConversationRatingV2).delete()  # noqa: ALN085
    current_session.query(MedicalConversationAttachment).delete()  # noqa: ALN085
    current_session.query(MedicalConversationPart).delete()  # noqa: ALN085
    current_session.query(MedicalConversation).delete()  # noqa: ALN085

    current_session.commit()

faker module-attribute

faker = Faker('fr_FR')

notify_stale_consultation_conversations

notify_stale_consultation_conversations(dry_run)

Notify members in stale consultation conversations (36h without HP reply post-TLC).

Source code in components/clinic/public/commands/medical_conversation.py

@clinic_commands.command()
@command_with_dry_run
def notify_stale_consultation_conversations(dry_run: bool) -> None:
    """Notify members in stale consultation conversations (36h without HP reply post-TLC)."""
    from components.clinic.internal.business_logic.medical_conversation import (
        notify_stale_consultation_conversations as _notify,
    )

    _notify(dry_run=dry_run)

resolve_closable_conversations

resolve_closable_conversations()

Source code in components/clinic/public/commands/medical_conversation.py

@clinic_commands.command()
def resolve_closable_conversations() -> None:  # noqa: D103
    from components.clinic.internal.business_logic.medical_conversation import (
        resolve_closable_conversations as _resolve_closable_conversations,
    )

    _resolve_closable_conversations()

send_pending_proactive_conversations

send_pending_proactive_conversations(dry_run)

Source code in components/clinic/public/commands/medical_conversation.py

@clinic_commands.command()
@command_with_dry_run
def send_pending_proactive_conversations(dry_run: bool) -> None:  # noqa: D103
    from components.clinic.internal.business_logic.medical_conversation_proactive import (
        send_pending_proactive_conversations,
    )

    send_pending_proactive_conversations(commit=not dry_run)

update_active_after_for_conversations

update_active_after_for_conversations(
    dry_run, recently_modified_only
)

This command will update the active_after field on conversations. It's helpful for: - computing it the first time - resetting it from time to time, if we have a bug and we want to fix the values

Source code in components/clinic/public/commands/medical_conversation.py

@clinic_commands.command()
@command_with_dry_run
@click.option(
    "--recently-modified-only",
    type=bool,
    required=False,
    default=False,
    is_flag=True,
    help="Only look at recently modified conversations",
)
def update_active_after_for_conversations(
    dry_run: bool, recently_modified_only: bool
) -> None:
    """
    This command will update the active_after field on conversations.
    It's helpful for:
    - computing it the first time
    - resetting it from time to time, if we have a bug and we want to fix the values
    """
    from components.clinic.internal.business_logic.medical_conversation import (
        update_active_after_for_conversations as _update_active_after_for_conversations,
    )

    _update_active_after_for_conversations(
        commit=not dry_run, recently_modified_only=recently_modified_only
    )
    if dry_run:
        # Just being safe
        current_session.rollback()

setup

setup

setup(
    password,
    with_fake_conversations,
    fake_conversations_count,
    yes,
)

Bootstrap a local Alan Clinic environment end-to-end.

Wipes and re-creates encryption keys, medical admins, the medical chat bot, and the booking next-availability cache. Optionally generates fake conversations.

Prints a checklist of remaining manual steps (worker, frontend, Marmot HP setup).

Source code in components/clinic/public/commands/setup.py

@clinic_commands.command()
@click.option(
    "--password",
    type=str,
    default="azerty",
    help="Password for the generated medical admin accounts.",
)
@click.option(
    "--with-fake-conversations",
    is_flag=True,
    default=False,
    help="Generate fake medical conversations after the core setup.",
)
@click.option(
    "--fake-conversations-count",
    type=int,
    default=50,
    help="How many fake conversations to generate when --with-fake-conversations is set.",
)
@click.option(
    "--yes",
    is_flag=True,
    default=False,
    help="Skip the destructive-actions confirmation prompt.",
)
def setup(
    password: str,
    with_fake_conversations: bool,
    fake_conversations_count: int,
    yes: bool,
) -> None:
    """
    Bootstrap a local Alan Clinic environment end-to-end.

    Wipes and re-creates encryption keys, medical admins, the medical chat bot,
    and the booking next-availability cache. Optionally generates fake conversations.

    Prints a checklist of remaining manual steps (worker, frontend, Marmot HP setup).
    """
    if not is_development_mode():
        click.secho(
            "`flask clinic setup` is only available in development mode.",
            fg="red",
            err=True,
        )
        return

    if not yes:
        click.secho(
            "This will TRUNCATE encryption keys, medical_admin*, doctor_picture and "
            "(if --with-fake-conversations) reset medical_conversation data.",
            fg="yellow",
        )
        click.confirm("Continue?", abort=True)

    _run_step(
        "Clinic: create medical admins",
        lambda: _step_create_medical_admins(password=password),
    )
    _run_step("Clinic: create medical chat bot", _step_create_medical_chat_bot)
    _run_step(
        "Encryption: regenerate keys + doctor group (admins enrolled)",
        _step_encryption,
    )
    _run_step("Booking: refresh next-availability cache", _step_update_booking_cache)
    if with_fake_conversations:
        _run_step(
            f"Fake data: {fake_conversations_count} medical conversations",
            lambda: _step_create_fake_conversations(count=fake_conversations_count),
        )

    _print_manual_checklist(password=password)

unread_message_reminders

send_medical_conversations_unread_message_reminders

send_medical_conversations_unread_message_reminders(
    dry_run,
)

Source code in components/clinic/public/commands/unread_message_reminders.py

@clinic_commands.command()
@command_with_dry_run
def send_medical_conversations_unread_message_reminders(dry_run: bool) -> None:  # noqa: D103
    from components.clinic.internal.business_logic.medical_conversation import (
        send_medical_conversations_unread_message_reminders,
    )

    send_medical_conversations_unread_message_reminders(dry_run=dry_run)

user

archive_clinic_user_data

archive_clinic_user_data(clinic_user_id, dry_run)

Archive a clinic user's data to S3 without deleting them.

Runs collect → encrypt → upload → mapping in dry-run mode by default. Pass --execute to perform the actual archive.

Usage

flask clinic archive_clinic_user_data flask clinic archive_clinic_user_data --execute

Source code in components/clinic/public/commands/user.py

@clinic_commands.command()
@click.argument("clinic_user_id", type=UUID)
@command_with_dry_run
def archive_clinic_user_data(clinic_user_id: UUID, dry_run: bool) -> None:
    """Archive a clinic user's data to S3 without deleting them.

    Runs collect → encrypt → upload → mapping in dry-run mode by default.
    Pass --execute to perform the actual archive.

    Usage:
        flask clinic archive_clinic_user_data <clinic_user_id>
        flask clinic archive_clinic_user_data <clinic_user_id> --execute
    """
    from components.clinic.internal.archiving.archiver import archive_clinic_user

    current_logger.info(
        "Archiving clinic user data (no deletion)",
        clinic_user_id=clinic_user_id,
        dry_run=dry_run,
    )
    archive_clinic_user(
        clinic_user_id=clinic_user_id,
        contract_termination_date=None,
        dry_run=dry_run,
    )
    current_logger.info(
        "Done",
        clinic_user_id=clinic_user_id,
        dry_run=dry_run,
    )

clear_stripe_customer_ids

clear_stripe_customer_ids(dry_run)

Clears all stripe_customer_id values from clinic_user table. Only runs in non-production environments. Usage: flask clinic clear_stripe_customer_ids flask clinic clear_stripe_customer_ids --execute

Source code in components/clinic/public/commands/user.py

@clinic_commands.command()
@command_with_dry_run
def clear_stripe_customer_ids(dry_run: bool) -> None:
    # TODO frederic.bonnet 2026-03-23 encapsulate Stripe ID access
    """Clears all stripe_customer_id values from clinic_user table.
    Only runs in non-production environments.
    Usage:
        flask clinic clear_stripe_customer_ids
        flask clinic clear_stripe_customer_ids --execute
    """
    from sqlalchemy import func, select, update

    from components.clinic.internal.models.clinic_user import ClinicUser
    from shared.helpers.db import current_session
    from shared.helpers.env import is_production_mode

    if is_production_mode():
        current_logger.error("Cannot run in production")
        return

    if dry_run:
        count = current_session.execute(
            select(func.count())
            .select_from(ClinicUser)
            # Workspace key: eu_stripe_therapist_booking | ca_stripe_health_services
            .where(ClinicUser.stripe_customer_id.is_not(None))
        ).scalar_one()
        current_logger.info(
            f"DRY RUN - would clear stripe_customer_id for {count} clinic users"
        )
    else:
        result = current_session.execute(
            update(ClinicUser)
            # Workspace key: eu_stripe_therapist_booking | ca_stripe_health_services
            .where(ClinicUser.stripe_customer_id.is_not(None))
            .values(stripe_customer_id=None)
        )
        current_session.commit()
        current_logger.info(
            f"Cleared stripe_customer_id for {result.rowcount} clinic users"
        )

delete_clinic_data_for_user

delete_clinic_data_for_user(clinic_user_id, dry_run)

Delete a clinic user and all associated data for non-medical admin users.

Source code in components/clinic/public/commands/user.py

@clinic_commands.command()
@click.argument("clinic_user_id", type=str)
@command_with_dry_run
def delete_clinic_data_for_user(clinic_user_id: UUID, dry_run: bool) -> None:
    """Delete a clinic user and all associated data for non-medical admin users."""
    from components.clinic.internal.business_logic.clinic_user import (
        delete_all_clinic_data_for_user,
    )

    current_logger.warn(
        "⚠️ This command will delete all clinic user data (conversations, messages, ratings, etc.)"
    )
    delete_all_clinic_data_for_user(clinic_user_id=clinic_user_id, dry_run=dry_run)

inspect_archive

inspect_archive(mapping_filename)

Fetch, decrypt, and print a mapping file and its referenced archive. Dev only.

Usage

flask clinic inspect_archive

Example

flask clinic inspect_archive mapping_clinic-456_20260331_120000.csv.enc

Source code in components/clinic/public/commands/user.py

@clinic_commands.command()
@click.argument("mapping_filename")
def inspect_archive(mapping_filename: str) -> None:
    """Fetch, decrypt, and print a mapping file and its referenced archive. Dev only.

    Usage:
        flask clinic inspect_archive <mapping_filename>

    Example:
        flask clinic inspect_archive mapping_clinic-456_20260331_120000.csv.enc
    """
    import io
    import json

    import defusedcsv.csv as csv

    from components.clinic.internal.archiving.encryption import decrypt_blob
    from shared.helpers.config import current_config
    from shared.helpers.env import is_production_mode
    from shared.services.aws import RemoteFileClient

    if is_production_mode():
        click.echo("ERROR: This command only runs in non-production environments.")
        return

    bucket = current_config.get("CLINIC_ARCHIVAL_S3_BUCKET")
    mapping_key = f"mappings/{mapping_filename}"

    # --- Fetch and decrypt mapping file ---
    click.echo(f"\n=== Mapping file: {mapping_key} ===")
    mapping_buf = io.BytesIO()
    RemoteFileClient.download_from_key_into(mapping_key, mapping_buf, s3_bucket=bucket)
    mapping_csv = decrypt_blob(mapping_buf.read().decode())
    click.echo(mapping_csv.decode())

    # --- Parse the s3_key from the mapping ---
    reader = csv.DictReader(io.StringIO(mapping_csv.decode()))
    rows = list(reader)
    if not rows:
        click.echo("ERROR: Mapping file is empty.")
        return

    s3_key = rows[0]["s3_key"]

    # --- Fetch and decrypt archive file ---
    click.echo(f"=== Archive file: {s3_key} ===")
    archive_buf = io.BytesIO()
    RemoteFileClient.download_from_key_into(s3_key, archive_buf, s3_bucket=bucket)
    archive_data = json.loads(decrypt_blob(archive_buf.read().decode()))
    click.echo(json.dumps(archive_data, indent=2, default=str))

components.clinic.public.dependencies

TherapySessionConfiguration dataclass

TherapySessionConfiguration(
    duration,
    price_multiplier,
    credit_multiplier,
    force_same_length_slot=False,
    available_for_booking=True,
)

Bases: DataClassJsonMixin

Configuration for a therapy session duration variant.

Attributes:

Name	Type	Description
duration	int	Session duration in minutes
price_multiplier	int	Multiplier applied to the base session price
credit_multiplier	int	Multiplier applied to the session credit count
force_same_length_slot	bool	Force google calendar slots to match session duration
available_for_booking	bool	Whether this configuration is visible for booking

available_for_booking class-attribute instance-attribute

available_for_booking = True

credit_multiplier instance-attribute

credit_multiplier

duration instance-attribute

duration

force_same_length_slot class-attribute instance-attribute

force_same_length_slot = False

price_multiplier instance-attribute

price_multiplier

components.clinic.public.entities

available_health_service

AvailableHealthService dataclass

AvailableHealthService(
    name,
    has_access=None,
    is_recommended=None,
    has_upcoming_availability=None,
)

Bases: DataClassJsonMixin

Available health service available for a clinic user

has_access class-attribute instance-attribute

has_access = None

has_upcoming_availability class-attribute instance-attribute

has_upcoming_availability = None

is_recommended class-attribute instance-attribute

is_recommended = None

name instance-attribute

name

external_onboarding_user

ExternalOnboardingUserData dataclass

ExternalOnboardingUserData(
    email,
    first_name,
    last_name,
    phone_number,
    password,
    prehashed_password,
    gender_str,
    date_of_birth,
    social_security_number,
    place_of_birth_name,
    place_of_birth_postal_code,
    terms_accepted,
    refresh_token_type,
    client_id=None,
    birth_last_name=None,
)

External onboarding user data. Attributes: email: The email of the user. first_name: The first name of the user. last_name: The last name of the user. phone_number: The phone number of the user. password: The password of the user. prehashed_password: The prehashed password of the user. gender_str: The gender of the user. date_of_birth: The date of birth of the user. social_security_number: The social security number of the user. place_of_birth_name: The name of the place of birth of the user. place_of_birth_postal_code: The postal code of the place of birth of the user. terms_accepted: Whether the user has accepted the terms and conditions. refresh_token_type: The refresh token type of the user. client_id: The Keycloak client ID for email verification flow. birth_last_name: The birth last name of the user (maiden name). Defaults to last_name if not provided.

birth_last_name class-attribute instance-attribute

birth_last_name = None

client_id class-attribute instance-attribute

client_id = None

date_of_birth instance-attribute

date_of_birth

email instance-attribute

email

first_name instance-attribute

first_name

gender_str instance-attribute

gender_str

last_name instance-attribute

last_name

password instance-attribute

password

phone_number instance-attribute

phone_number

place_of_birth_name instance-attribute

place_of_birth_name

place_of_birth_postal_code instance-attribute

place_of_birth_postal_code

prehashed_password instance-attribute

prehashed_password

refresh_token_type instance-attribute

refresh_token_type

social_security_number instance-attribute

social_security_number

terms_accepted instance-attribute

terms_accepted

shop_product

ShopProduct dataclass

ShopProduct(id, name, brand, image_url=None)

Bases: DataClassJsonMixin

Shop product visible to health professionals in the catalog.

brand instance-attribute

brand

id instance-attribute

id

image_url class-attribute instance-attribute

image_url = None

name instance-attribute

name

components.clinic.public.enums

available_health_service_name

AvailableHealthServiceName

Bases: AlanBaseEnum

Health services available to members that can be shared to them in the clinic

DATO_CONTENT class-attribute instance-attribute

DATO_CONTENT = 'dato_content'

HEALTH_PROGRAM class-attribute instance-attribute

HEALTH_PROGRAM = 'health_program'

ORIENTATION_CALL class-attribute instance-attribute

ORIENTATION_CALL = 'orientation_call'

SHOP_PRODUCT class-attribute instance-attribute

SHOP_PRODUCT = 'shop_product'

THERAPIST class-attribute instance-attribute

THERAPIST = 'therapist'

THERAPY_SESSION class-attribute instance-attribute

THERAPY_SESSION = 'therapy_session'

VIDEO_CONSULTATION class-attribute instance-attribute

VIDEO_CONSULTATION = 'video_consultation'

VIDEO_CONSULTATION_LIVI class-attribute instance-attribute

VIDEO_CONSULTATION_LIVI = 'video_consultation_livi'

available_shop_product_types

AvailableShopProductTypes

Bases: AlanBaseEnum

Shop products available to HPs that can be shared to members

CONTACT_LENSES class-attribute instance-attribute

CONTACT_LENSES = 'contact_lenses'

pricing

ClinicInvoiceAmountCurrency

Bases: AlanBaseEnum

CAD class-attribute instance-attribute

CAD = 'CAD'

EUR class-attribute instance-attribute

EUR = 'EUR'

GBP class-attribute instance-attribute

GBP = 'GBP'

format_amount

format_amount(amount_in_cents)

Source code in components/clinic/internal/enums/clinic_invoice_amount_currency.py

def format_amount(self, amount_in_cents: int) -> str:
    amount = Decimal(amount_in_cents) / 100
    if self == self.EUR:
        return format_currency(amount, "EUR", locale="fr_FR")
    elif self == self.GBP:
        return format_currency(amount, "GBP", locale="en_GB")
    elif self == self.CAD:
        return format_currency(amount, "CAD", locale="en_CA")
    else:
        raise ValueError(f"Unsupported currency: {self}")

regions

CaAdministrativeAreas

Bases: AlanBaseEnum

Canadian provinces and territories.

Used to specify which administrative regions a medical admin is licensed to operate in. Values use ISO 3166-2:CA codes (CA- prefix + 2-letter region code). -

alberta class-attribute instance-attribute

alberta = 'CA-AB'

british_columbia class-attribute instance-attribute

british_columbia = 'CA-BC'

manitoba class-attribute instance-attribute

manitoba = 'CA-MB'

new_brunswick class-attribute instance-attribute

new_brunswick = 'CA-NB'

newfoundland_and_labrador class-attribute instance-attribute

newfoundland_and_labrador = 'CA-NL'

northwest_territories class-attribute instance-attribute

northwest_territories = 'CA-NT'

nova_scotia class-attribute instance-attribute

nova_scotia = 'CA-NS'

nunavut class-attribute instance-attribute

nunavut = 'CA-NU'

ontario class-attribute instance-attribute

ontario = 'CA-ON'

prince_edward_island class-attribute instance-attribute

prince_edward_island = 'CA-PE'

quebec class-attribute instance-attribute

quebec = 'CA-QC'

saskatchewan class-attribute instance-attribute

saskatchewan = 'CA-SK'

yukon class-attribute instance-attribute

yukon = 'CA-YT'

therapist_booking_session_type

TherapistBookingSessionType

Bases: AlanBaseEnum

Type of booking session supported by the clinic booking experience

consultation class-attribute instance-attribute

consultation = 'consultation'

dermatology class-attribute instance-attribute

dermatology = 'dermatology'

emergency_physiotherapy class-attribute instance-attribute

emergency_physiotherapy = 'emergency_physiotherapy'

orientation class-attribute instance-attribute

orientation = 'orientation'

physiotherapy class-attribute instance-attribute

physiotherapy = 'physiotherapy'

therapy class-attribute instance-attribute

therapy = 'therapy'

components.clinic.public.events

events

ClinicUserDataDeleted dataclass

ClinicUserDataDeleted(user_id, app_id)

Bases: Message

Published after a clinic user and all associated data have been deleted.

Other components should subscribe to clean up their own data linked to this user.

app_id instance-attribute

app_id

user_id instance-attribute

user_id

subscription

subscribe_to_events

subscribe_to_events()

All event subscriptions for the Clinic should be done here.

Source code in components/clinic/public/events/subscription.py

def subscribe_to_events() -> None:
    """
    All event subscriptions for the Clinic should be done here.
    """
    from components.clinic.internal.booking.business_logic.events_subscribers import (
        configure_medical_conversation_when_booking_sessions,
        setup_conversation_for_therapy_session,
        setup_lifecycle_tracking_for_booked_session_from_mo_redirection,
        track_lifecycle_event_for_session_from_mo_redirection,
        trigger_insi_create_or_update_from_id_verification_request,
    )
    from components.clinic.internal.booking.entities.events import (
        SessionCreatedForInternalTeleconsultation,
        SessionCreatedForTherapy,
        SessionRecommendedByMoWasBooked,
        SessionRecommendedByMoWasUpdated,
        TriggerINSiUpdate,
    )
    from components.clinic.internal.business_logic.identity.events_subscribers import (
        update_insi_identity_status_following_id_check,
    )
    from components.id_verification.public.events import (
        IdVerificationValidatedForClinic,
        IdVerificationValidatedForExternalOnboarding,
    )

    message_broker = get_message_broker()
    message_broker.subscribe_async(
        IdVerificationValidatedForClinic,
        update_insi_identity_status_following_id_check,
        queue_name=LOW_PRIORITY_QUEUE,
    )
    message_broker.subscribe_async(
        IdVerificationValidatedForExternalOnboarding,
        update_insi_identity_status_following_id_check,
        queue_name=LOW_PRIORITY_QUEUE,
    )
    message_broker.subscribe_async(
        SessionCreatedForInternalTeleconsultation,
        configure_medical_conversation_when_booking_sessions,
        queue_name=LOW_PRIORITY_QUEUE,
    )
    message_broker.subscribe_async(
        SessionCreatedForTherapy,
        setup_conversation_for_therapy_session,
        queue_name=LOW_PRIORITY_QUEUE,
    )
    message_broker.subscribe_async(
        TriggerINSiUpdate,
        trigger_insi_create_or_update_from_id_verification_request,
        queue_name=LOW_PRIORITY_QUEUE,
    )
    message_broker.subscribe_async(
        SessionRecommendedByMoWasBooked,
        setup_lifecycle_tracking_for_booked_session_from_mo_redirection,
        queue_name=LOW_PRIORITY_QUEUE,
    )
    message_broker.subscribe_async(
        SessionRecommendedByMoWasUpdated,
        track_lifecycle_event_for_session_from_mo_redirection,
        queue_name=LOW_PRIORITY_QUEUE,
    )