Api reference

components.es.public.account

get_account_admins

get_account_admins(account_id)

Source code in components/es/public/account.py

def get_account_admins(account_id: UUID) -> list[EsUser]:  # noqa: D103
    # TODO: don't return a native model, but a dataclass. Do this once we know the expected data from consumer.
    account_admins = (
        current_session.query(EsUser)  # noqa: ALN085
        .join(EsAccountAdmin)
        .join(EsAccount)
        .filter(
            EsAccount.id == account_id,
            EsAccountAdmin.ended_at.is_(None),
            EsUser.email.isnot(None),  # type: ignore[attr-defined]
        )
    )
    return account_admins.all()

get_account_name

get_account_name(*, account_id)

Source code in components/es/public/account.py

def get_account_name(*, account_id: UUID) -> str:  # noqa: D103
    from components.es.internal.models.es_account import EsAccount
    from shared.helpers.get_or_else import get_or_raise_missing_resource

    account = get_or_raise_missing_resource(EsAccount, account_id)
    return account.name

user_is_account_admin

user_is_account_admin(*, user_id, account_id)

Source code in components/es/public/account.py

def user_is_account_admin(*, user_id: UUID, account_id: UUID) -> bool:  # noqa: D103
    from components.es.internal.business_logic.account.account import is_admin

    if not user_id or not account_id:
        return False

    return is_admin(user_id=user_id, account_id=account_id)

components.es.public.auth

authorization

AuthorizationStrategies

alaner_admin class-attribute instance-attribute

alaner_admin = EsAlanerAdminStrategy

authenticated class-attribute instance-attribute

authenticated = EsAuthenticatedStrategy

open class-attribute instance-attribute

open = EsOpenStrategy

owner_only class-attribute instance-attribute

owner_only = EsOwnerOnlyStrategy

EsAlanerAdminStrategy

EsAlanerAdminStrategy(permitted_for=None)

Bases: AlanerAdminStrategy

Source code in components/es/public/auth/authorization.py

def __init__(self, permitted_for: set[EmployeePermission] | None = None) -> None:
    super().__init__(permitted_for=permitted_for)

EsAuthenticatedStrategy

EsAuthenticatedStrategy(allow_deep_link=False)

Bases: AuthenticatedStrategy

Source code in components/es/public/auth/authorization.py

def __init__(self, allow_deep_link: bool = False) -> None:
    super().__init__(allow_deep_link=allow_deep_link)

EsOpenStrategy

EsOpenStrategy(allow_deep_link=False)

Bases: OpenStrategy

Source code in components/es/public/auth/authorization.py

def __init__(self, allow_deep_link: bool = False) -> None:
    super().__init__(allow_deep_link=allow_deep_link)

EsOwnerOnlyStrategy

EsOwnerOnlyStrategy(
    owner_bypass_permitted_for=None, allow_deep_link=False
)

Bases: OwnerOnlyStrategy

Source code in components/es/public/auth/authorization.py

def __init__(
    self,
    owner_bypass_permitted_for: set[EmployeePermission] | None = None,
    allow_deep_link: bool = False,
) -> None:
    super().__init__(
        owner_bypass_permitted_for=owner_bypass_permitted_for,
        allow_deep_link=allow_deep_link,
    )

business_logic

mfa_push_notification

send_mfa_operation_pending_push_notification

send_mfa_operation_pending_push_notification(
    user_id, description, pending_operation_id, type
)

Sends a request to validate an MFA-protected operation

Source code in components/es/public/auth/business_logic/mfa_push_notification.py

@push_notification_sender_sync  # type: ignore[arg-type]
def send_mfa_operation_pending_push_notification(
    user_id: UserId,
    description: str,
    pending_operation_id: UUID,
    type: str | None,  # noqa: A002
) -> UserPushNotificationParams:
    """
    Sends a request to validate an MFA-protected operation
    """
    user = user_service.get_user(
        cast("UUID", user_id)  # safe to cast as we know it's a UUID
    )
    lang = user.profile.lang_compat if user is not None else Lang.spanish
    title = translate(language=lang, key_string="mfa_operation_pending.title")
    description = translate(
        language=lang, key_string="mfa_operation_pending.description"
    )

    current_logger.debug(
        f"Sending MFA push notification: {pending_operation_id}",
        pending_operation_id=pending_operation_id,
        user_id=user_id,
    )
    return UserPushNotificationParams(
        user_id=cast("UUID", user_id),  # safe to cast as we know it's a UUID
        tokens=get_push_notification_tokens_for_user(
            user_app_id=str(user_id), app_name=AppName.ALAN_ES
        ),
        name=PushNotificationName.mfa_operation_pending_push_notification,
        title=title,
        body=description,
        extra_data={
            "operation_id": str(pending_operation_id),
            "description": description,
            "type": type,  # type: ignore[dict-item]
            # Repeated so that it's also available for the screen displayed when the notification is opened
        },
    )

components.es.public.blueprints

admin_api_blueprint

admin_api_blueprint module-attribute

admin_api_blueprint = CustomBlueprint("admin_api", __name__)

admin_tools

CARD_SHIPMENT_PAGE_SIZE module-attribute

CARD_SHIPMENT_PAGE_SIZE = 25

customize_email

customize_email(template_name)

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route(
    "/mails/customize/<path:template_name>", methods=["GET", "POST"]
)
@obs.api_call()
def customize_email(template_name: str):  # type: ignore[no-untyped-def]  # noqa: D103
    from shared.blueprints.admin_tools.blueprint import render_customize_email

    return render_customize_email(template_name)

dkv_authorization

dkv_authorization(authorization_id)

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route(
    "/dkv/authorization/<uuid:authorization_id>",
    methods=["GET"],
    permitted_for={EmployeePermission.es_review_authorization},
)
@obs.api_call()
def dkv_authorization(authorization_id):  # type: ignore[no-untyped-def]  # noqa: D103
    authorization = get_or_raise_missing_resource(EsAuthorization, authorization_id)
    if authorization.dkv_authorization_id is None:
        html = flask.render_template_string(
            _DKV_AUTHORIZATION_TEMPLATE,
            dkv_authorization="",
            status="No DKV authorization ID for this authorization.",
        )
        return html

    status = ""
    response = None
    try:
        response = dkv_authorization_logic.update_authorization_status(authorization_id)
    except DkvApiCallError as exc:
        status = _format_dkv_error("Getting authorization", exc)
        current_logger.error(
            "DKV admin: failed to get authorization", authorization_id=authorization_id
        )
    if response is None:
        dkv_authorization_str = "Authorization doesn't exist."
    else:
        dkv_authorization_str = json.dumps(
            response.raw,
            sort_keys=True,
            indent=2,
            ensure_ascii=False,
        )
    html = flask.render_template_string(
        _DKV_AUTHORIZATION_TEMPLATE,
        dkv_authorization=dkv_authorization_str,
        status=status,
    )
    return html

dkv_enrollment

dkv_enrollment(enrollment_id)

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route(
    "/dkv/enrollment/<uuid:enrollment_id>", methods=["GET", "POST"]
)
@obs.api_call()
def dkv_enrollment(enrollment_id: UUID) -> str:  # noqa: D103
    enrollment = get_or_raise_missing_resource(EsEnrollment, enrollment_id)
    status = []

    if flask.request.method == "POST":
        action = flask.request.form.get("action")
        if action == "sync":
            try:
                dkv_enrollment_logic.enroll_in_dkv(enrollment_id, commit=True)
                status.append("Synced to DKV")
                current_logger.info(
                    "DKV admin: synced enrollment", enrollment_id=enrollment_id
                )
            except DkvApiCallError as exc:
                status.append(_format_dkv_error("Syncing", exc))
                current_logger.error(
                    "DKV admin: failed to sync enrollment", enrollment_id=enrollment_id
                )
        elif action == "cancel":
            try:
                dkv_enrollment_logic.cancel_dkv_enrollment(enrollment_id, commit=True)
                status.append("DKV enrollment cancelled")
                current_logger.info(
                    "DKV admin: cancelled DKV enrollement", enrollment_id=enrollment_id
                )
            except DkvApiCallError as exc:
                status.append(_format_dkv_error("Cancellation", exc))
                current_logger.error(
                    "DKV admin: failed to cancel enrollment",
                    enrollment_id=enrollment_id,
                )
        else:
            status.append(f"unknown action '{action}': ignoring")
            current_logger.error("DKV admin: unknown action, ignoring", action=action)

    dkv_openid = DkvOpenid()
    token = dkv_openid.generate_token(scope=DkvTokenScope.POLICY)
    client = DkvPolicyApi(access_token=token.response.access_token)

    dkv_policy_ref = dkv_enrollment_logic.enrollment_to_dkv_api_policy_ref(enrollment)

    dkv_enrollment = None
    try:
        result = client.get_dkv_policy(dkv_policy_ref)
        dkv_enrollment = result.response.result.data
    except DkvApiCallError as exc:
        status.append(_format_dkv_error("Fetching DKV data", exc))
        current_logger.error(
            "DKV admin: failed to get DKV enrollment", enrollment_id=enrollment_id
        )

    dkv_enrollment_str = json.dumps(
        dkv_enrollment, sort_keys=True, indent=2, ensure_ascii=False
    )

    enrollment_in_dkv_format = dkv_enrollment_logic.enrollment_to_dkv_api_policy(
        enrollment
    )
    enrollment_in_dkv_format_str = json.dumps(
        marshmallow_dataclass.class_schema(DkvApiPolicy)().dump(
            enrollment_in_dkv_format
        ),
        sort_keys=True,
        indent=2,
        ensure_ascii=False,
    )

    html = flask.render_template_string(
        _DKV_ENROLLMENT_TEMPLATE,
        enrollment_id=enrollment_id,
        dkv_enrollment=dkv_enrollment_str,
        enrollment_in_dkv_format=enrollment_in_dkv_format_str,
        status="\n".join(status),
    )
    return html

enrollment_address

enrollment_address(enrollment)

Source code in components/es/public/blueprints/admin_tools.py

def enrollment_address(enrollment: EsEnrollment):  # type: ignore[no-untyped-def]  # noqa: D103
    user_profile = user_service.get_or_raise_user_profile(
        enrollment.insurance_profile.user_id
    )
    if user_profile.address:
        return user_profile.address
    elif enrollment.type != EnrollmentType.primary:
        user_profiles = user_service.get_or_raise_user_profiles(
            e.insurance_profile.user_id for e in enrollment.policy.enrollments
        )
        for e in enrollment.policy.enrollments:
            if (
                e.type == EnrollmentType.primary
                and user_profiles[e.insurance_profile.user_id].address
            ):
                return user_profiles[e.insurance_profile.user_id].address

    return None

export_batch

export_batch(batch_id)

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route(
    "/cards/batch/<uuid:batch_id>.csv",
    permitted_for={EmployeePermission.tp_cards_management},
)
@obs.api_call()
def export_batch(batch_id: UUID):  # type: ignore[no-untyped-def]  # noqa: D103
    batch = get_or_raise_missing_resource(EsCardShipmentBatch, batch_id)

    file = io.BytesIO()
    w = writer(file)
    w.writerow(  # type: ignore[no-untyped-call]
        [
            "member_name",
            "policy_number",
            "street",
            "postal_code",
            "city",
            "province",
            "tp1cof",
            "tp1tsa",
            "tp1tcc",
            "tp1nta",
        ]
    )
    for card in batch.card_shipments:
        if card.status == CardShipmentStatus.CANCELLED:
            continue
        if not card.enrollment:
            continue
        if card.enrollment.is_cancelled:
            continue
        user = card.enrollment.insurance_profile.user
        dkv_magnetic_card = card.enrollment.dkv_magnetic_card
        address = enrollment_address(card.enrollment)
        w.writerow(  # type: ignore[no-untyped-call]
            [
                user.full_name,
                dkv_magnetic_card.number,
                address and address.street or "",
                address and address.postal_code or "",
                address and address.locality or "",
                address and province_name_for_postal_code(address.postal_code) or "",
                dkv_magnetic_card.tp1cof,
                dkv_magnetic_card.tp1tsa,
                dkv_magnetic_card.tp1tcc,
                dkv_magnetic_card.tp1nta,
            ]
        )

    file.seek(0)

    return send_file(file, f"batch-{batch_id}.csv", mimetype="text/csv")

index

index()

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route("/")
@obs.api_call()
def index():  # type: ignore[no-untyped-def]  # noqa: D103
    return render_template(
        "admin_tools/index.html",
        documents=admin_tools_blueprint.documents.values(),
        ipids_manzana=admin_tools_blueprint.ipids_manzana,
        ipids_fresa=admin_tools_blueprint.ipids_fresa,
        ipids_manzana_individuals=admin_tools_blueprint.ipids_manzana_individuals,
    )

init_blueprint

init_blueprint(state)

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.record_once
def init_blueprint(state: BlueprintSetupState) -> None:  # noqa: D103
    with state.app.app_context():

        def cb(admin_tool: AdminToolsBlueprint) -> None:
            admin_tool.ipids_manzana = get_ipids_manzana_preview()
            admin_tool.ipids_fresa = get_ipids_fresa_preview()
            admin_tool.ipids_manzana_individuals = (
                get_ipids_manzana_individuals_preview()
            )
            document_factories = [
                get_cg_preview,
                get_cg_individuals_preview,
                get_cg_fresa_preview,
                get_cp_preview,
                get_cp_individuals_preview,
                get_cp_fresa_preview,
                get_health_insurance_invoice_preview,
                get_healthy_benefits_invoice_preview,
                get_healthy_benefits_rectification_invoice_preview,
                get_sepa_preview,
                get_sepa_individuals_preview,
                get_sepa_fresa_preview,
                get_sepa_healthy_benefits_preview,
                get_certificate_preview,
                get_certificate_individuals_preview,
                get_certificate_fresa_preview,
            ]

            for document_factory in document_factories:
                try:
                    admin_tool.register_document(document_factory())
                except Exception:
                    # This can happen during tests if the health plans haven't been created
                    current_logger.warning(
                        "Unable to register document for admin blueprint.",
                        exc_info=True,
                    )

        admin_tools_blueprint.register_document_registration_callback(cb)

insurance_card

insurance_card()

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route("/csv/insurance_card")
@obs.api_call()
def insurance_card():  # type: ignore[no-untyped-def]  # noqa: D103
    start_date = today() - relativedelta(days=14)
    end_date = today()

    bl = InsuranceCardLogic()
    with NamedTemporaryFile(mode="w") as output_csv:
        bl.insurance_card_information_for_new_employees_between_as_csv(
            file=output_csv,  # type: ignore[arg-type]
            start_date=start_date,
            end_date=end_date,
        )
        output_csv.flush()
        return send_file(output_csv.name, "insurance_card.csv", mimetype="text/csv")

insurance_cards

insurance_cards()

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route(
    "/cards",
    methods=["GET", "POST"],
    permitted_for={EmployeePermission.tp_cards_management},
)
@obs.api_call()
def insurance_cards():  # type: ignore[no-untyped-def]  # noqa: D103
    message = None
    if flask.request.method == "POST":
        action = flask.request.form.get("action")
        if action == "create_card":
            dkv_id = flask.request.form.get("dkv_id")
            reason = flask.request.form.get("reason")

            try:
                user = (
                    current_session.query(EsUser)  # noqa: ALN085
                    .join(EsInsuranceProfile)
                    .filter(EsInsuranceProfile.dkv_id == dkv_id)
                    .one()
                )
            except NoResultFound:
                raise NotFound(f"DKV id '{dkv_id}' not found")

            enrollment = EsEnrollmentModelBroker.get_user_active_or_future_enrollment(
                user.id
            )

            if not enrollment:
                raise NotFound(f"No enrollment was found for user {user.id}")

            card = EsCardShipment(
                enrollment=enrollment,
                status=CardShipmentStatus.PENDING,
                reason=reason,
            )
            current_session.add(card)
            current_session.commit()
            message = "New card shipment created"
        elif action == "set_status":
            card_id = flask.request.form.get("card_id")
            status = flask.request.form.get("status")

            card = get_or_raise_missing_resource(EsCardShipment, card_id)
            card.status = CardShipmentStatus[status]  # type: ignore[misc]
            current_session.commit()
            message = "Updated status"
        elif action == "make_batch":
            cards = current_session.query(EsCardShipment).filter(  # noqa: ALN085
                EsCardShipment.batch_id.is_(None)
            )
            batch = EsCardShipmentBatch()
            for card in cards:
                card.batch = batch
                # CANCELLED cards can stay in that status, they won't be exported
                if card.status == CardShipmentStatus.PENDING:
                    card.status = CardShipmentStatus.SENT
            current_session.add(batch)
            current_session.commit()
            message = "New card batch"
        else:
            message = f"Unknown action {action}"

    page = flask.request.args.get("page", 1, type=int)
    query = (
        current_session.query(EsCardShipment)  # noqa: ALN085
        .join(EsEnrollment)
        .options(
            joinedload(EsCardShipment.enrollment)
            .joinedload(EsEnrollment.insurance_profile)
            .joinedload(EsInsuranceProfile.user)
            .joinedload(EsUser.address),
            joinedload(EsCardShipment.enrollment)
            .joinedload(EsEnrollment.policy)
            .joinedload(EsPolicy.enrollments)
            .joinedload(EsEnrollment.insurance_profile)
            .joinedload(EsInsuranceProfile.user)
            .joinedload(EsUser.address),
            joinedload(EsCardShipment.batch),
        )
        .order_by(EsCardShipment.created_at.desc())
    )

    pagination = paginate(
        query=query,
        page=page,
        per_page=CARD_SHIPMENT_PAGE_SIZE,
    )
    cards = pagination.items  # type: ignore[assignment]

    if cards and cards[0].batch_id is not None:
        unbatched_card_count = 0
        missing_addresses = False
    else:
        unbatched_cards = (
            current_session.query(EsCardShipment)  # noqa: ALN085
            .join(EsEnrollment)
            .options(
                joinedload(EsCardShipment.enrollment)
                .joinedload(EsEnrollment.insurance_profile)
                .joinedload(EsInsuranceProfile.user)
                .joinedload(EsUser.address),
                joinedload(EsCardShipment.enrollment)
                .joinedload(EsEnrollment.policy)
                .joinedload(EsPolicy.enrollments)
                .joinedload(EsEnrollment.insurance_profile)
                .joinedload(EsInsuranceProfile.user)
                .joinedload(EsUser.address),
                joinedload(EsCardShipment.batch),
            )
            .filter(EsCardShipment.batch_id.is_(None))
        )
        unbatched_card_count = unbatched_cards.count()
        missing_addresses = False
        for card in unbatched_cards:
            if not card.enrollment:
                continue
            address = enrollment_address(card.enrollment)
            if address is None and card.status != CardShipmentStatus.CANCELLED:
                missing_addresses = True
                break

    return flask.render_template_string(
        _CARD_LIST_TEMPLATE,
        cards=cards,
        message=message,
        pagination=pagination,
        url_for=flask.url_for,
        unbatched_card_count=unbatched_card_count,
        missing_addresses=missing_addresses,
        province_name_for_postal_code=province_name_for_postal_code,
        enrollment_address=enrollment_address,
        CardShipmentStatus=CardShipmentStatus,
    )

list_api_endpoints

list_api_endpoints()

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route("/api/documentation", methods=["GET"])
@obs.api_call()
def list_api_endpoints():  # type: ignore[no-untyped-def]  # noqa: D103
    from shared.blueprints.api_documentation import render_list_api_endpoints

    return render_list_api_endpoints()

list_commands

list_commands()

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route("/commands/list", methods=["GET"])
@obs.api_call()
def list_commands():  # type: ignore[no-untyped-def]  # noqa: D103
    from shared.blueprints.commands import render_list_commands

    return render_list_commands()

list_email

list_email()

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route("/mails/list", methods=["GET"])
@obs.api_call()
def list_email():  # type: ignore[no-untyped-def]  # noqa: D103
    from shared.blueprints.admin_tools.blueprint import render_list_email

    return render_list_email()

load_admin_tools_blueprint

load_admin_tools_blueprint()

Source code in components/es/public/blueprints/admin_tools.py

def load_admin_tools_blueprint() -> AdminToolsBlueprint:  # noqa: D103
    return admin_tools_blueprint

meeting_doctors_user_panel

meeting_doctors_user_panel()

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route("/meeting_doctors", methods=["GET", "POST"])
@obs.api_call()
def meeting_doctors_user_panel():  # type: ignore[no-untyped-def]  # noqa: D103
    from components.es.internal.business_logic.enrollment.meeting_doctors import (
        add_access_to_meeting_doctors,
        get_meeting_doctors_user,
        remove_access_to_meeting_doctors,
    )

    _user_id = unquote(
        flask.request.args.get("user_id", "")
    ).strip()  # remove spaces, tabs from user_id
    user_id = cast("UUID", _user_id)
    user_exists = None
    user_has_access_to_meeting_doctors = None
    user_has_insurance_profile = None
    meeting_doctors_user = None
    action = None
    success = None

    if flask.request.method == "POST":
        action = flask.request.form.get("action")
        if action == "remove_access":
            success = remove_access_to_meeting_doctors(user_id)
        elif action == "add_access":
            success = add_access_to_meeting_doctors(user_id)

    if user_id:
        es_user = user_logic.get_user(user_id).db_model
        user_exists = es_user is not None
        if user_exists:
            user_has_access_to_meeting_doctors = es_user.has_access_to_meeting_doctors
            user_has_insurance_profile = es_user.insurance_profile is not None
            if user_has_access_to_meeting_doctors:
                meeting_doctors_user = get_meeting_doctors_user(es_user.id)

    html = flask.render_template_string(
        _MEETING_DOCTORS_TEMPLATE,
        user_id=user_id,
        user_exists=user_exists,
        user_has_access_to_meeting_doctors=user_has_access_to_meeting_doctors,
        user_has_insurance_profile=user_has_insurance_profile,
        meeting_doctors_user=(
            json.dumps(meeting_doctors_user, indent=2, ensure_ascii=False)
            if meeting_doctors_user
            else ""
        ),
        action=action,
        success=success,
    )
    return html

nursery_installments

nursery_installments()

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route("/nursery_installments", methods=["GET", "POST"])
@obs.api_call()
def nursery_installments():  # type: ignore[no-untyped-def]  # noqa: D103
    from collections import defaultdict

    from components.es.external.payment_gateway.helpers import (
        generate_nursery_installment_reference,
        generate_public_reference_id,
    )
    from components.es.internal.models.es_company import EsCompany
    from components.es.subcomponents.healthy_benefits.internal.models.brokers.nursery_installment import (
        NurseryInstallmentModelBroker,
    )
    from components.es.subcomponents.healthy_benefits.internal.models.brokers.nursery_installment_plan import (
        NurseryInstallmentPlanModelBroker,
    )
    from components.healthy_benefits.public.subscription import (
        get_subscription_by_id,
    )

    first_day_of_month = today().date().replace(day=1)

    active_installment_plans = (
        NurseryInstallmentPlanModelBroker.get_all_active_installment_plans(
            first_day_of_month
        )
    )
    due_installments = NurseryInstallmentModelBroker.get_all_nursery_installments(
        due_after_date=first_day_of_month
    )

    amounts_per_company: dict[UUID, int] = defaultdict(int)
    for installment in due_installments:
        enrollment = get_enrollment(installment.installment_plan.enrollment_ref)
        subscription = get_subscription_by_id(
            subscription_id=enrollment.subscription_id
        )

        company_id = UUID(subscription.owner_ref)
        amounts_per_company[company_id] += installment.amount

    def get_company_name(company_id: UUID) -> str:
        company = current_session.get(EsCompany, company_id)
        return company.name  # type: ignore[union-attr]

    def get_company_payment_account_external_id(company_id: UUID) -> str | None:
        try:
            from components.es.external.payment_gateway.company_payment_account import (
                get_company_account,
            )

            account = get_company_account(company_id)
            return account.external_id
        except ValueError:
            # Company doesn't have a payment account
            return None

    if flask.request.method == "POST":
        action = flask.request.form.get("action")
        datetime_now = datetime.now()
        if action == "create_installments":
            due_installment_plan_ids = {
                installment.installment_plan_id for installment in due_installments
            }
            for installment_plan in active_installment_plans:
                if installment_plan.id not in due_installment_plan_ids:
                    # Generate public reference id for tracking
                    public_id = generate_public_reference_id()

                    description = (  # Use nursery name as description for in-app display
                        installment_plan.nursery.name
                    )
                    reference = generate_nursery_installment_reference(public_id)

                    # Create installment
                    installment = (
                        NurseryInstallmentModelBroker.create_nursery_installment(
                            installment_plan_id=installment_plan.id,
                            amount=installment_plan.monthly_amount,
                            due_date=first_day_of_month,
                            description=description,
                            reference=reference,
                        )
                    )

                    # Record debit transfer for the member's nursery enrollment
                    enrollment = get_enrollment(
                        installment.installment_plan.enrollment_ref
                    )
                    subscription = get_subscription_by_id(
                        subscription_id=enrollment.subscription_id
                    )

                    company_id = UUID(subscription.owner_ref)
                    subscription_version = mandatory(
                        get_ongoing_or_future_subscription(
                            company_ref=str(company_id), at_date=datetime_now.date()
                        ).versions.get_ongoing_period(on_date=datetime_now.date())
                    )

                    recover_benefit_funds(
                        company_id=company_id,
                        subscription_version=subscription_version,
                        enrollment=enrollment,
                        amount=installment_plan.monthly_amount,
                        occurred_at=datetime_now,
                        description=description,
                        reference=reference,
                    )

            current_session.commit()
            return redirect(url_for("admin_tools.nursery_installments"))

        elif action == "download_csv_installments":
            file = io.BytesIO()
            w = writer(file)
            w.writerow(  # type: ignore[no-untyped-call]
                [
                    "company_name",
                    "beneficiary_name",
                    "nursery_name",
                    "nursery_address",
                    "nursery_phone",
                    "nursery_countact_email",
                    "nursery_iban",
                    "amount_cents",
                    "due_date",
                    "reference",
                ]
            )
            for installment in due_installments:
                installment_plan = installment.installment_plan
                enrollment = get_enrollment(installment_plan.enrollment_ref)
                subscription = get_subscription_by_id(
                    subscription_id=enrollment.subscription_id
                )
                company_id = UUID(subscription.owner_ref)
                company_name = get_company_name(company_id)
                beneficiary = installment_plan.beneficiary
                nursery = installment_plan.nursery
                w.writerow(  # type: ignore[no-untyped-call]
                    [
                        company_name,
                        " ".join([beneficiary.first_name, beneficiary.last_name]),
                        nursery.name,
                        ", ".join(
                            [
                                nursery.address.street,
                                nursery.address.postal_code,
                                nursery.address.city,
                            ]
                        ),
                        nursery.phone_number,
                        nursery.contact_email,
                        nursery.iban_code,
                        installment.amount,
                        installment.due_date,
                        installment.reference,
                    ]
                )

            file.seek(0)

            return send_file(
                file,
                f"nursery-installments-{first_day_of_month.isoformat()}.csv",
                mimetype="text/csv",
            )

        elif action == "download_csv_companies":
            file = io.BytesIO()
            w = writer(file)
            w.writerow(  # type: ignore[no-untyped-call]
                [
                    "company_name",
                    "payment_account_external_id",
                    "amount_cents",
                ]
            )
            for company_id, amount in amounts_per_company.items():
                # Get payment account external_id through proper component interface
                payment_account_external_id = get_company_payment_account_external_id(
                    company_id
                )

                w.writerow(  # type: ignore[no-untyped-call]
                    [
                        get_company_name(company_id),
                        payment_account_external_id,
                        amount,
                    ]
                )

            file.seek(0)

            return send_file(
                file,
                f"nursery-company-transfers-{first_day_of_month.isoformat()}.csv",
                mimetype="text/csv",
            )

    return flask.render_template_string(
        _NURSERY_INSTALLMENTS_TEMPLATE,
        nb_active_installment_plans=len(active_installment_plans),
        nb_due_installments=len(due_installments),
        due_installments=due_installments,
        amounts_per_company=amounts_per_company,
        get_company_name=get_company_name,
        get_company_payment_account_external_id=get_company_payment_account_external_id,
    )

push_to_algolia_for_rq

push_to_algolia_for_rq(prod)

Source code in components/es/public/blueprints/admin_tools.py

def push_to_algolia_for_rq(prod) -> None:  # type: ignore[no-untyped-def]  # noqa: D103
    components.es.internal.business_logic.medical_network.push_network_to_algolia.push_medical_network_to_algolia(
        prod=prod,
        should_push_specialties=False,
        should_push_network=False,
        should_push_services=True,
    )

show_command

show_command(command_full_name)

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route("/commands/<path:command_full_name>")
@obs.api_call()
def show_command(command_full_name):  # type: ignore[no-untyped-def]  # noqa: D103
    from shared.blueprints.commands import render_show_command

    return render_show_command(command_full_name=command_full_name)

sync_medical_services

sync_medical_services()

Source code in components/es/public/blueprints/admin_tools.py

@admin_tools_blueprint.route(
    "/sync_medical_services_with_algolia", methods=["GET", "POST"]
)
@obs.api_call()
def sync_medical_services():  # type: ignore[no-untyped-def]  # noqa: D103
    medical_service_information_flask_url = (
        f"{current_config['BASE_URL']}/admin/medicalserviceinformation/"
    )
    synced = False
    if flask.request.method == "POST":
        current_rq.get_queue(MEDICAL_NETWORK_QUEUE).enqueue(
            push_to_algolia_for_rq,
            prod=is_production_mode(),
        )
        synced = True

    return flask.render_template_string(
        _SYNC_MEDICAL_SERVICES_WITH_ALGOLIA_TEMPLATE,
        synced=synced,
        medical_service_information_flask_url=medical_service_information_flask_url,
    )

data_consistency_blueprint

data_consistency_blueprint module-attribute

data_consistency_blueprint = CustomBlueprint(
    "data_consistency", __name__
)

register_blueprint

register_blueprint(state)

Source code in components/es/public/blueprints/data_consistency_blueprint.py

@data_consistency_blueprint.record_once
def register_blueprint(state) -> None:  # type: ignore[no-untyped-def]  # noqa: ARG001, D103
    # Register commands
    from components.es.internal.data_consistency.commands.register import (
        register_data_consistency_commands,
    )

    register_data_consistency_commands(data_consistency_blueprint)

es_api_blueprint

es_api_blueprint module-attribute

es_api_blueprint = CustomBlueprint(
    "es_api_blueprint", __name__
)

es_core_blueprint

es_core_blueprint module-attribute

es_core_blueprint = CustomBlueprint(
    "es-core",
    __name__,
    cli_group=None,
    template_folder="templates",
    static_folder="static",
    static_url_path="/static",
)

register_blueprint

register_blueprint(state)

Source code in components/es/public/blueprints/es_core_blueprint.py

@es_core_blueprint.record_once
def register_blueprint(state) -> None:  # type: ignore[no-untyped-def]  # noqa: ARG001, D103
    import components.es.internal.models  # noqa: F401
    from components.es.internal.commands.affiliation_connectors.connector_credentials import (  # noqa: F401
        connector_credentials,
    )
    from components.es.internal.commands.affiliation_connectors.factorial_connector import (  # noqa: F401
        factorial_connector,
    )
    from components.es.internal.commands.amedics import (  # noqa: F401
        amedics,
    )
    from components.es.internal.commands.company import (  # noqa: F401
        company,
    )
    from components.es.internal.commands.company_onboarding import (  # noqa: F401
        company_onboarding,
    )
    from components.es.internal.commands.concierge import (  # noqa: F401
        concierge,
    )
    from components.es.internal.commands.contract_option import (  # noqa: F401
        contract_option,
    )
    from components.es.internal.commands.creu_blanca import (  # noqa: F401
        creu_blanca,
    )
    from components.es.internal.commands.dependent import (  # noqa: F401
        dependent,
    )
    from components.es.internal.commands.dkv import dkv  # noqa: F401
    from components.es.internal.commands.employee_onboarding import (  # noqa: F401
        employee_onboarding,
    )
    from components.es.internal.commands.employee_transfer import (  # noqa: F401
        employee_transfer,
    )
    from components.es.internal.commands.employment import (  # noqa: F401
        es_employment,
    )
    from components.es.internal.commands.ex_employees import (  # noqa: F401
        ex_employees,
    )
    from components.es.internal.commands.gdpr import gdpr  # noqa: F401
    from components.es.internal.commands.health_contract import (  # noqa: F401
        health_contract,
    )
    from components.es.internal.commands.house_keeping import (  # noqa: F401
        house_keeping,
    )
    from components.es.internal.commands.iban import iban  # noqa: F401
    from components.es.internal.commands.life_cycle import (  # noqa: F401
        life_cycle,
    )
    from components.es.internal.commands.medical_network import (  # noqa: F401
        medical_network,
    )
    from components.es.internal.commands.meeting_doctors import (  # noqa: F401
        meeting_doctors,
    )
    from components.es.internal.commands.monitoring import (  # noqa: F401
        monitoring,
    )
    from components.es.internal.commands.online_booking import (  # noqa: F401
        online_booking,
    )
    from components.es.internal.commands.pay_csv import (  # noqa: F401
        pay_csv,
    )
    from components.es.internal.commands.payment_transfer import (  # noqa: F401
        payment_transfer,
    )
    from components.es.internal.commands.premium_entry import (  # noqa: F401
        premium_entry,
    )
    from components.es.internal.commands.push_notifications import (  # noqa: F401
        push_notifications,
    )
    from components.es.internal.commands.reimbursement import (  # noqa: F401
        upsell_reimbursement,
    )
    from components.es.internal.commands.reimbursement_payment import (  # noqa: F401
        reimbursement_payment,
    )
    from components.es.internal.commands.revolut_rotate_access_token import (  # noqa: F401
        revolut_rotate_access_token,
    )
    from components.es.internal.commands.segment import (  # noqa: F401
        segment,
    )
    from components.es.internal.commands.signed_documents import (  # noqa: F401
        signed_documents,
    )
    from components.es.internal.commands.travel_assistance_csv import (  # noqa: F401
        travel_assistance_csv,
    )
    from components.es.internal.commands.tuotempo import (  # noqa: F401
        tuotempo,
    )
    from components.es.internal.commands.user import user  # noqa: F401
    from components.es.internal.services.revolut import (
        revolut_webhook_endpoint,  # noqa: F401
    )

components.es.public.clinic

adapter

EsClinicAdapter

Bases: ClinicAdapter

get_allowlist_of_dermatology_medical_admins_ids

get_allowlist_of_dermatology_medical_admins_ids()

Source code in components/es/public/clinic/adapter.py

def get_allowlist_of_dermatology_medical_admins_ids(self) -> list[str]:  # noqa: D102
    # No dermatology session in Es
    return []

get_app_base_user_data

get_app_base_user_data(app_user_id)

Source code in components/es/public/clinic/adapter.py

def get_app_base_user_data(self, app_user_id: str) -> BaseUserData:  # noqa: D102
    from components.es.internal.business_logic.user_v2.service import user_service

    user = user_service.get_or_raise_user(UUID(app_user_id))
    return BaseUserData(
        first_name=normalize_name(user.profile.first_name),
        last_name=normalize_name(user.profile.last_name),
    )

get_app_user_available_health_services

get_app_user_available_health_services(app_user_id)

Source code in components/es/public/clinic/adapter.py

def get_app_user_available_health_services(  # noqa: D102
    self,
    app_user_id: str,
) -> list[AvailableHealthService]:
    available_health_services = [
        AvailableHealthService(name=AvailableHealthServiceName.DATO_CONTENT),
        AvailableHealthService(name=AvailableHealthServiceName.HEALTH_PROGRAM),
        AvailableHealthService(name=AvailableHealthServiceName.THERAPY_SESSION),
        AvailableHealthService(name=AvailableHealthServiceName.THERAPIST),
    ]

    if self.has_access_to_orientation_call(app_user_id):
        # If the user has therapy sessions included, we add the orientation call
        available_health_services.append(
            AvailableHealthService(name=AvailableHealthServiceName.ORIENTATION_CALL)
        )

    return available_health_services

get_app_user_data

get_app_user_data(
    app_user_id, compute_key_account_info=False
)

Source code in components/es/public/clinic/adapter.py

def get_app_user_data(  # noqa: D102
    self,
    app_user_id: str,
    compute_key_account_info: bool = False,  # noqa: ARG002
) -> UserData:
    from components.es.internal.business_logic.user_v2.service import user_service
    from components.es.public.user import get_visible_dependents

    user = user_service.get_or_raise_user(UUID(app_user_id))

    dependent_users = user_service.get_or_raise_users(
        {
            enrollment.insurance_profile.user_id
            for enrollment in get_visible_dependents(user.id)
        }
    )

    dependents = [
        Dependent(
            app_user_id=str(dependent_user.id),
            first_name=normalize_name(dependent_user.profile.first_name),
            last_name=normalize_name(dependent_user.profile.last_name),
            age=int(dependent_user.profile.age)
            if dependent_user.profile.age
            else None,
            gender=dependent_user.profile.gender_compat,
            birth_date=dependent_user.profile.birth_date,
            dependent_type=None,  # ES doesn't use dependent type
        )
        for dependent_user in dependent_users
        if dependent_user != user
    ]

    return UserData(
        first_name=normalize_name(user.profile.first_name),
        last_name=normalize_name(user.profile.last_name),
        gender=user.profile.gender_compat,
        email=user.profile.email,
        profile_id=user.profile.id,
        birth_date=user.profile.birth_date,
        phone=user.profile.phone_number,
        country=user.profile.address.country if user.profile.address else None,
        address=format_address_as_one_line(user.profile.address)
        if user.profile.address
        else None,
        ssn=None,  # Spain doesn't use SSN
        lang=user.profile.lang_compat,
        is_key_account_or_large_company_and_not_alaner=False,  # ES doesn't use this
        is_alaner=user.db_model.is_alaner,
        dependents=dependents,
    )

get_booking_session_package

get_booking_session_package(app_user_id, session_type)

Source code in components/es/public/clinic/adapter.py

def get_booking_session_package(  # noqa: D102
    self, app_user_id: str, session_type: TherapistBookingSessionType
) -> BookingSessionPackage | None:
    if session_type == TherapistBookingSessionType.therapy:
        return _get_therapy_session_package(app_user_id=UUID(app_user_id))
    elif session_type == TherapistBookingSessionType.physiotherapy:
        return _get_physiotherapy_session_package(app_user_id=UUID(app_user_id))
    elif session_type == TherapistBookingSessionType.orientation:
        return _get_orientation_call_session_package()

    return None

get_coverage_status

get_coverage_status(app_user_id)

Source code in components/es/public/clinic/adapter.py

def get_coverage_status(self, app_user_id: str) -> CoverageStatus | None:  # noqa: D102
    from components.es.public.clinic.clinic_eligibility import get_coverage_status

    return get_coverage_status(user_id=UUID(app_user_id))

get_last_active_id_verification_request_for_user

get_last_active_id_verification_request_for_user(
    app_user_id,
)

ES implementation of getting the last active ID verification request for a user. Since Spain doesn't support ID verification for the clinic, this will fail if we try to call it.

Source code in components/es/public/clinic/adapter.py

def get_last_active_id_verification_request_for_user(
    self,
    app_user_id: str,
) -> None:
    """
    ES implementation of getting the last active ID verification request for a user.
    Since Spain doesn't support ID verification for the clinic, this will fail if we try to call it.
    """
    raise NotImplementedError(
        "ES doesn't support ID verification for clinic users."
    )

has_access_to_orientation_call

has_access_to_orientation_call(app_user_id)

Source code in components/es/public/clinic/adapter.py

def has_access_to_orientation_call(  # noqa: D102
    self, app_user_id: str
) -> bool:
    from components.es.internal.business_logic.company.subscribed_therapy_sessions import (
        get_company_covered_therapy_sessions_limit,
    )
    from components.es.internal.business_logic.enrollment.queries.therapy_sessions_limit import (
        get_user_covered_therapy_sessions_limit,
    )
    from components.es.public.company import get_company_id_of_user

    company_id = get_company_id_of_user(user_id=UUID(app_user_id))
    nb_credits_included = (
        get_company_covered_therapy_sessions_limit(company_id=company_id)
        if company_id is not None
        else get_user_covered_therapy_sessions_limit(UUID(app_user_id))
    )

    # Check if the user has access to the orientation call feature based on the app version
    # @remirubis Rémi mistakenly released the global flow without the new implementations in version 1.390.0,
    # so we want to make sure that members can't take advantage of it before 1.392.0.
    app_version = _get_app_version()
    if not version_is_above_or_equal(
        app_version or "", MIN_APP_VERSION_FOR_ORIENTATION_CALL
    ):
        return False

    return nb_credits_included > 0

has_app_user_permission

has_app_user_permission(app_user_id, permission)

Source code in components/es/public/clinic/adapter.py

def has_app_user_permission(  # noqa: D102
    self, app_user_id: str, permission: EmployeePermission
) -> bool:
    from components.es.public.user import get_user as get_es_user

    es_user = get_es_user(UUID(app_user_id)).db_model
    return has_permission(es_user, permission)

is_app_user_admin_of_company

is_app_user_admin_of_company(app_user_id, app_company_id)

Source code in components/es/public/clinic/adapter.py

def is_app_user_admin_of_company(  # noqa: D102
    self, app_user_id: str, app_company_id: str
) -> bool:
    from components.es.public.company import user_is_company_admin

    return user_is_company_admin(
        user_id=UUID(app_user_id), company_id=UUID(app_company_id)
    )

release_date_of_conversations_created_for_therapy_sessions class-attribute instance-attribute

release_date_of_conversations_created_for_therapy_sessions = datetime(
    2025, 9, 23
)

request_id_verification_request_for_user

request_id_verification_request_for_user(
    app_user_id, user_info, commit=True
)

ES implementation of getting or requesting ID verification for a clinic user. Since Spain doesn't support ID verification for clinic users, this will raise a NotImplementedError if it's called.

Source code in components/es/public/clinic/adapter.py

def request_id_verification_request_for_user(
    self,
    app_user_id: str,
    user_info: ClinicUserDataForIdVerification,
    commit: bool = True,
) -> None:
    """
    ES implementation of getting or requesting ID verification for a clinic user.
    Since Spain doesn't support ID verification for clinic users, this will raise a NotImplementedError if it's called.
    """
    raise NotImplementedError(
        "ES doesn't support ID verification for clinic users."
    )

should_request_id_verification_for_user

should_request_id_verification_for_user(app_user_id)

ES implementation of checking if ID verification should be requested for a clinic user. Since Spain doesn't support ID verification for clinic users, this will always return False.

Source code in components/es/public/clinic/adapter.py

def should_request_id_verification_for_user(
    self,
    app_user_id: str,  # noqa: ARG002
) -> bool:
    """
    ES implementation of checking if ID verification should be requested for a clinic user.
    Since Spain doesn't support ID verification for clinic users, this will always return False.
    """
    return False

update_app_user_phone

update_app_user_phone(app_user_id, phone)

Source code in components/es/public/clinic/adapter.py

def update_app_user_phone(self, app_user_id: str, phone: str | None) -> None:  # noqa: D102
    from components.es.internal.business_logic.user_v2.service import user_service

    user_service.update_user_profile(
        user_id=UUID(app_user_id), profile_data={"phone": phone}, commit=True
    )

update_app_user_ssn

update_app_user_ssn(app_user_id, ssn, commit=False)

Source code in components/es/public/clinic/adapter.py

def update_app_user_ssn(  # noqa: D102
    self, app_user_id: str, ssn: str | None, commit: bool = False
) -> None:
    # ES doesn't use SSN
    pass

upload_invoice_as_insurance_document

upload_invoice_as_insurance_document(
    file, app_user_id, upload_invoice_data
)

Source code in components/es/public/clinic/adapter.py

def upload_invoice_as_insurance_document(  # noqa: D102
    self,
    file: IO,  # type: ignore[type-arg]  # noqa: ARG002
    app_user_id: str,  # noqa: ARG002
    upload_invoice_data: UploadInvoiceData,  # noqa: ARG002
) -> bool:
    # ES doesn't support insurance document uploads
    return False

user_has_24_hour_response_guarantee

user_has_24_hour_response_guarantee(app_user_id)

Source code in components/es/public/clinic/adapter.py

def user_has_24_hour_response_guarantee(  # noqa: D102
    self,
    app_user_id: str,  # noqa: ARG002
) -> bool:
    # ES doesn't support 24 hour response guarantee
    return False

validate_session_duration

validate_session_duration(session_duration)

Source code in components/es/public/clinic/adapter.py

def validate_session_duration(  # noqa: D102
    self,
    session_duration: int,
) -> None:
    # 45min sessions are banned in ES
    if session_duration in UNAUTHORIZED_SESSION_LENGTHS:
        raise ValueError(
            f"Sessions of duration {session_duration} minutes are not available for booking in ES"
        )

MIN_APP_VERSION_FOR_ORIENTATION_CALL module-attribute

MIN_APP_VERSION_FOR_ORIENTATION_CALL = '1.392.0'

PHYSIOTHERAPY_SESSION_PRICE_FOR_USER_WITHOUT_ACCESS_IN_CENTS module-attribute

PHYSIOTHERAPY_SESSION_PRICE_FOR_USER_WITHOUT_ACCESS_IN_CENTS = (
    2000
)

THERAPY_SESSION_PRICE_IN_CENTS module-attribute

THERAPY_SESSION_PRICE_IN_CENTS = 3000

UNAUTHORIZED_SESSION_LENGTHS module-attribute

UNAUTHORIZED_SESSION_LENGTHS = [45]

clinic_adapter module-attribute

clinic_adapter = EsClinicAdapter()

clinic_eligibility

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

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

get_coverage_status

get_coverage_status(user_id)

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

Source code in components/es/public/clinic/clinic_eligibility.py

def get_coverage_status(user_id: UUID) -> CoverageStatus | None:
    """
    Return the start and optionally the end date of the current or upcoming period of eligibility to the clinic restricted services.
    """
    from components.es.internal.business_logic.enrollment.queries.base_enrollment import (
        get_user_active_base_enrollments,
    )

    user_active_enrollments = get_user_active_base_enrollments(user_id=user_id)
    insurance_enrollment = first_or_none(
        enrollment
        for enrollment in user_active_enrollments
        # only consider health insurance enrollments
        if enrollment.is_health_insurance_product()
        # exclude past enrollments
        and (enrollment.end_date is None or enrollment.end_date >= date.today())
    )

    if not insurance_enrollment:
        return None

    return CoverageStatus(
        start_date=insurance_enrollment.start_date,
        end_date=insurance_enrollment.end_date,
    )

components.es.public.command_log

get_command_logs

get_command_logs(start_at, end_at)

Source code in components/es/public/command_log.py

def get_command_logs(start_at: datetime, end_at: datetime) -> list[CommandLog]:  # noqa: D103
    logs = current_session.scalars(
        select(EsCommandLog).filter(
            EsCommandLog.created_at >= start_at, EsCommandLog.created_at < end_at
        )
    )

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

components.es.public.company

get_account_ids_from_company_ids

get_account_ids_from_company_ids(company_ids)

Extract account IDs list from company IDs.

Source code in components/es/public/company.py

def get_account_ids_from_company_ids(company_ids: Iterable[str]) -> list[str]:
    """Extract account IDs list from company IDs."""
    company_to_account_mapping = get_company_id_to_account_id(company_ids)
    account_ids = set(company_to_account_mapping.values())
    return sorted(account_ids)

get_companies

get_companies(company_ids)

Source code in components/es/public/company.py

def get_companies(company_ids: list[uuid.UUID]) -> list[EsCompanyEntity]:  # noqa: D103
    companies = current_session.query(EsCompany).filter(EsCompany.id.in_(company_ids))  # noqa: ALN085
    return [
        EsCompanyEntity(id=company.id, display_name=company.display_name)
        for company in companies
    ]

get_company_display_name

get_company_display_name(company_id)

Source code in components/es/public/company.py

def get_company_display_name(company_id: uuid.UUID) -> str:  # noqa: D103
    company = get_or_raise_missing_resource(EsCompany, company_id)
    return company.display_name

get_company_id_of_user

get_company_id_of_user(user_id)

Source code in components/es/public/company.py

def get_company_id_of_user(user_id: uuid.UUID) -> uuid.UUID | None:  # noqa: D103
    try:
        subscriptor = get_user_active_or_future_subscription(
            user_id=user_id
        ).subscriptor
        if isinstance(subscriptor, CompanySubscriptor):
            return subscriptor.subscriptor_id
        return None

    except NoActiveContractException:
        return None

get_company_id_to_account_id

get_company_id_to_account_id(company_ids)

Source code in components/es/public/company.py

def get_company_id_to_account_id(  # noqa: D103
    company_ids: Iterable[str],
) -> dict[str, str]:
    companies = current_session.query(EsCompany).filter(EsCompany.id.in_(company_ids))  # noqa: ALN085
    return {str(company.id): str(company.account_id) for company in companies}

get_company_ids_for_account_id

get_company_ids_for_account_id(account_id)

Source code in components/es/public/company.py

def get_company_ids_for_account_id(account_id: uuid.UUID) -> list[uuid.UUID]:  # noqa: D103
    companies = (
        current_session.query(EsCompany)  # noqa: ALN085
        .filter(EsCompany.account_id == account_id)
        .all()
    )
    return [company.id for company in companies]

get_company_period_at_date

get_company_period_at_date(company_id, at_date)

Retrieve the validity period for a specific company at the provided date.

This function loads the CompanyPayrollPeriod for the given company_id at the specified at_date, then returns the current_period from that object.

:param company_id: The UUID of the company. :param at_date: The date at which to retrieve the company period. :return: A ValidityPeriod instance representing the current period for the specified company at the given date.

Source code in components/es/public/company.py

def get_company_period_at_date(company_id: uuid.UUID, at_date: date) -> ValidityPeriod:
    """
    Retrieve the validity period for a specific company at the provided date.

    This function loads the CompanyPayrollPeriod for the given company_id
    at the specified at_date, then returns the current_period from that object.

    :param company_id: The UUID of the company.
    :param at_date: The date at which to retrieve the company period.
    :return: A ValidityPeriod instance representing the current period
             for the specified company at the given date.
    """
    company_payroll_period = CompanyPayrollPeriod.load_from_company(
        company_id=company_id, at_date=at_date
    )
    return company_payroll_period.current_period

get_employees_invite_emails

get_employees_invite_emails(company_id)

Source code in components/es/public/company.py

def get_employees_invite_emails(company_id: uuid.UUID) -> set[str]:  # noqa: D103
    active_employments = (
        current_session.query(EsEmployment)  # noqa: ALN085
        .filter(
            EsEmployment.company_id == company_id,
            EsEmployment.is_active,
            EsEmployment.invite_email.isnot(None),
        )
        .all()
    )
    return {employment.invite_email for employment in active_employments}  # type: ignore[misc]

get_user_admined_company_ids

get_user_admined_company_ids(user_id)

Source code in components/es/public/company.py

def get_user_admined_company_ids(user_id: uuid.UUID) -> list[uuid.UUID]:  # noqa: D103
    return [company.id for company in get_user_admined_companies(user_id=user_id)]

is_company

is_company(company_id)

Source code in components/es/public/company.py

def is_company(company_id: uuid.UUID) -> bool:  # noqa: D103
    return current_session.get(EsCompany, company_id) is not None

is_employee_in_one_of_companies

is_employee_in_one_of_companies(
    user_id, company_ids, not_ended_on_date
)

Source code in components/es/public/company.py

def is_employee_in_one_of_companies(  # noqa: D103
    user_id: uuid.UUID,
    company_ids: list[uuid.UUID],
    not_ended_on_date: date,
) -> bool:
    return current_session.query(  # type: ignore[no-any-return] # noqa: ALN085
        current_session.query(EsEmployment)  # noqa: ALN085
        .filter(
            EsEmployment.user_id == user_id,
            EsEmployment.company_id.in_(company_ids),
            ~EsEmployment.is_ended_on(not_ended_on_date),
        )
        .exists()
    ).scalar()

user_is_company_admin

user_is_company_admin(user_id, company_id)

Source code in components/es/public/company.py

def user_is_company_admin(user_id: uuid.UUID, company_id: uuid.UUID) -> bool:  # noqa: D103
    return company_logic.user_is_admin(user_id=user_id, company_id=company_id)

components.es.public.contracting

legal_documents

get_cp_individual_data_from_individual_legal_document_info

get_cp_individual_data_from_individual_legal_document_info(
    legal_documents_info,
)

Returns the generated CpIndividualData from the instance of IndividualLegalDocumentsInfo provided.

    Args:
        legal_documents_info (UUID): The health contract id of the member's health contract.

    Returns:
        (CpIndividualData): The data class containing particular conditions data

Source code in components/es/public/contracting/legal_documents.py

def get_cp_individual_data_from_individual_legal_document_info(
    legal_documents_info: IndividualLegalDocumentsInfo,
) -> CpIndividualData:
    """
    Returns the generated CpIndividualData from the instance of IndividualLegalDocumentsInfo
    provided.

            Args:
                legal_documents_info (UUID): The health contract id of the member's health contract.

            Returns:
                (CpIndividualData): The data class containing particular conditions data
    """
    user = user_service.get_or_raise_user(legal_documents_info.user_id)

    if user.profile.address is None:
        raise BaseErrorCode.missing_resource(
            message="The user does not have an address", user_id=user.id
        )

    iban = iban_logic.get_last_iban_set_for_user(user_id=legal_documents_info.user_id)
    contract_info = legal_documents_info.contract_info

    product_options: dict[ProductType | OptionType, OptionCpData] = (
        ex_employees_logic.generate_cp_option_data_from_product_options(
            insurance_products=contract_info.insurance_products,
            options=contract_info.options,
        )
    )

    cp_individual_data = CpIndividualData(
        user_name=user.profile.full_name,
        user_street=user.profile.address.street,
        user_postal_code=mandatory(user.profile.address.postal_code),
        user_city=mandatory(user.profile.address.locality),
        iban_code=iban.iban,
        user_nif=user.db_model.insurance_profile.nif_number
        if user.db_model.insurance_profile
        else None,
        health_contract_price=contract_info.price,
        payment_method=contract_info.payment_method,
        contract_start_date=contract_info.contract_start_date,
        contract_end_date=contract_info.contract_end_date,
        manzana_product_data=product_options.get(ProductType.ALAN_MANZANA)
        if ProductType.ALAN_MANZANA in product_options
        else None,
        fresa_product_data=product_options.get(ProductType.ALAN_FRESA)
        if ProductType.ALAN_FRESA in product_options
        else None,
        reimbursements_add_on_data=product_options.get(OptionType.REIMBURSEMENTS)
        if OptionType.REIMBURSEMENTS in product_options
        else None,
        pec_add_on_data=product_options.get(OptionType.PRE_EXISTING_CONDITIONS)
        if OptionType.PRE_EXISTING_CONDITIONS in product_options
        else None,
        pharmacy_add_on_data=product_options.get(OptionType.PHARMACY)
        if OptionType.PHARMACY in product_options
        else None,
        physio_nutrition_add_on_data=product_options.get(OptionType.PHYSIO_NUTRITION)
        if OptionType.PHYSIO_NUTRITION in product_options
        else None,
        optical_add_on_data=product_options.get(OptionType.OPTICAL)
        if OptionType.OPTICAL in product_options
        else None,
    )

    return cp_individual_data

get_sepa_mandate_data_from_individual_legal_document_info

get_sepa_mandate_data_from_individual_legal_document_info(
    legal_documents_info,
)

Returns the generated SepaMandateData from the instance of IndividualLegalDocumentsInfo provided.

    Args:
        legal_documents_info (UUID): The health contract id of the member's health contract.

    Returns:
        (SepaMandateData): The data class containing the information needed to generate
        a SEPA mandate.

Source code in components/es/public/contracting/legal_documents.py

def get_sepa_mandate_data_from_individual_legal_document_info(
    legal_documents_info: IndividualLegalDocumentsInfo,
) -> SepaMandateData:
    """
    Returns the generated SepaMandateData from the instance of IndividualLegalDocumentsInfo
    provided.

            Args:
                legal_documents_info (UUID): The health contract id of the member's health contract.

            Returns:
                (SepaMandateData): The data class containing the information needed to generate
                a SEPA mandate.
    """
    user = user_service.get_or_raise_user(legal_documents_info.user_id)
    iban = iban_logic.get_last_iban_set_for_user(user_id=legal_documents_info.user_id)

    if user.profile.address is None:
        raise BaseErrorCode.missing_resource(
            message="The user does not have an address", user_id=user.id
        )

    return SepaMandateData(
        name=user.profile.full_name,
        street=user.profile.address.street,
        postal_code=mandatory(user.profile.address.postal_code),
        city=mandatory(user.profile.address.locality),
        iban_code=iban.iban,
        sepa_unique_id=iban.id.hex,
    )

components.es.public.customer_health_partner

employees_count

get_employees_count

get_employees_count(*, account_id, latest_start_date)

Source code in components/es/public/customer_health_partner/employees_count.py

def get_employees_count(*, account_id: UUID, latest_start_date: date) -> int:  # noqa: D103
    from components.es.internal.models.es_company import EsCompany
    from components.es.internal.models.es_employment import (
        EsEmployment,
    )

    all_company_ids_from_account = [
        str(company_id)
        for (company_id,) in current_session.query(EsCompany.id).filter(  # noqa: ALN085
            EsCompany.account_id == account_id
        )
    ]

    total_number_of_employees = (
        current_session.query(EsEmployment.id)  # noqa: ALN085
        .filter(
            EsEmployment.company_id.in_(all_company_ids_from_account),
            EsEmployment.is_cancelled is not True,  # type: ignore[arg-type,comparison-overlap]
            EsEmployment.start_date <= latest_start_date,
            EsEmployment.end_date.is_(None),
        )
        .join(EsEmployment.user)
        .distinct()
        .count()
    )

    return total_number_of_employees

get_admin_traits

get_admin_traits_to_notify

get_admin_traits_to_notify(admin_id, company_ids)

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

Source code in components/es/public/customer_health_partner/get_admin_traits.py

def get_admin_traits_to_notify(
    admin_id: str | None,
    company_ids: list[str],  # noqa: ARG001
) -> Sequence[AdminTraits]:
    """
    Return the list of AdminTraits for admins who should be notified about the well-being assessment report.
    """
    users_traits_of_users_to_notify: list[AdminTraits] = []

    # we notify only the admin who created the assessment
    if admin_id is not None:
        admin_user = get_or_raise_missing_resource(EsUser, admin_id)
        users_traits_of_users_to_notify.append(AdminTraits(admin_user))

    return users_traits_of_users_to_notify

get_company_ids_for_account_company_ids

get_company_ids_for_account_company_ids

get_company_ids_for_account_company_ids(
    company_ids_in_account,
)

Source code in components/es/public/customer_health_partner/get_company_ids_for_account_company_ids.py

def get_company_ids_for_account_company_ids(  # noqa: D103
    company_ids_in_account: list[str],
) -> set[str]:
    from components.es.internal.models.es_account import EsAccount
    from components.es.internal.models.es_company import (
        EsCompany,
    )

    account: EsAccount = mandatory(
        current_session.get(EsCompany, company_ids_in_account[0]).account  # type: ignore[union-attr]
    )
    account_company_ids = {str(company.id) for company in account.companies}
    return account_company_ids

components.es.public.employment

es_country_gateway

EsCountryGateway

Bases: CountryGateway[EsExtendedValues]

Spanish implementation of the Employment Component's CountryGateway

are_companies_in_same_account

are_companies_in_same_account(company_id_1, company_id_2)

Source code in components/es/public/employment/es_country_gateway.py

@override
def are_companies_in_same_account(
    self, company_id_1: str, company_id_2: str
) -> bool:
    from components.es.internal.models.es_company import EsCompany

    company_1 = get_or_raise_missing_resource(EsCompany, company_id_1)
    company_2 = get_or_raise_missing_resource(EsCompany, company_id_2)

    return company_1.account_id == company_2.account_id

get_account_name

get_account_name(account_id)

Source code in components/es/public/employment/es_country_gateway.py

@override
def get_account_name(self, account_id: UUID) -> str:
    from components.es.internal.models.es_account import EsAccount

    return get_or_raise_missing_resource(EsAccount, account_id).name

get_company_information

get_company_information(company_id)

Source code in components/es/public/employment/es_country_gateway.py

@override
def get_company_information(
    self,
    company_id: str,
) -> CompanyInformation:
    from components.es.internal.models.es_company import EsCompany

    company = get_or_raise_missing_resource(EsCompany, company_id)
    return CompanyInformation(
        display_name=company.display_name,
        account_id=company.account_id,
    )

get_employee_identifier_for_country

get_employee_identifier_for_country(extended_values)

Source code in components/es/public/employment/es_country_gateway.py

@override
def get_employee_identifier_for_country(
    self, extended_values: EsExtendedValues
) -> str | None:
    return extended_values.get("invite_nif_number")

get_employment_consumers

get_employment_consumers()

Gets all employment consumers contributed by this country.

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

Source code in components/es/public/employment/es_country_gateway.py

@override
def get_employment_consumers(self) -> set[EmploymentConsumer[EsExtendedValues]]:
    """
    Gets all employment consumers contributed by this country.

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

    return {es_health_affiliation_employment_change_consumer}

get_upstream_retry_handler

get_upstream_retry_handler(source_type)

Source code in components/es/public/employment/es_country_gateway.py

@override
def get_upstream_retry_handler(
    self, source_type: SourceType
) -> UpstreamBlockedMovementRetryFunction[EsExtendedValues] | None:
    from components.es.internal.business_logic.affiliation_connectors.factorial_connector.factorial_connector import (
        retry_es_factorial_blocked_movement,
    )
    from components.es.internal.business_logic.employee_onboarding.auto_enrollment.bulk_auto_enroll_from_csv import (
        retry_bulk_auto_enroll_blocked_movement,
    )
    from components.es.internal.business_logic.employment.global_employment.upstreams.employee_invitation import (
        retry_admin_dashboard_invitation,
    )

    retry_handlers: dict[
        SourceType, UpstreamBlockedMovementRetryFunction[EsExtendedValues]
    ] = {
        SourceType.es_factorial: retry_es_factorial_blocked_movement,
        SourceType.es_admin_dashboard: retry_admin_dashboard_invitation,
        SourceType.es_bulk_auto_enroll: retry_bulk_auto_enroll_blocked_movement,
    }
    return retry_handlers.get(source_type)

get_user_admined_company_ids

get_user_admined_company_ids(user_id)

Source code in components/es/public/employment/es_country_gateway.py

@override
def get_user_admined_company_ids(self, user_id: str) -> list[str]:
    from components.es.internal.business_logic.company.company_logic import (
        get_user_admined_companies,
    )

    return [
        str(admined_company.id)
        for admined_company in get_user_admined_companies(UUID(user_id))
    ]

get_user_full_name

get_user_full_name(user_id)

Source code in components/es/public/employment/es_country_gateway.py

@override
def get_user_full_name(self, user_id: str) -> str | None:
    from components.es.internal.business_logic.user_v2.service import user_service

    user = user_service.get_user_profile(user_id)
    return user.full_name if user else None

es_extended_values

EsEmploymentDeclaration module-attribute

EsEmploymentDeclaration = EmploymentDeclaration[
    EsExtendedValues
]

EsExtendedValues

Bases: ExtendedValuesDict

Spanish extended values stored in the Employment Component

connector_external_id instance-attribute

connector_external_id

coverage_start_date instance-attribute

coverage_start_date

gross_annual_salary instance-attribute

gross_annual_salary

healthy_benefits_start_date instance-attribute

healthy_benefits_start_date

hiring_date instance-attribute

hiring_date

initial_employment_id instance-attribute

initial_employment_id

invite_auto_enroll instance-attribute

invite_auto_enroll

invite_birth_date instance-attribute

invite_birth_date

invite_email instance-attribute

invite_email

invite_identifier_document_type instance-attribute

invite_identifier_document_type

invite_nif_number instance-attribute

invite_nif_number

offboarding_email_send_date instance-attribute

offboarding_email_send_date

population_id instance-attribute

population_id

tax_regime instance-attribute

tax_regime

termination_reason_type instance-attribute

termination_reason_type

components.es.public.employment_consumer

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

es_health_affiliation_employment_change_consumer

es_health_affiliation_employment_change_consumer(
    employment_change, event_bus_orchestrator
)

Source code in components/es/public/employment_consumer.py

def es_health_affiliation_employment_change_consumer(  # noqa: D103
    employment_change: EmploymentChange["EsExtendedValues"],
    event_bus_orchestrator: EventBusOrchestrator,
) -> None:
    if employment_change.country_code != CountryCode.es:
        return

    from components.es.internal.business_logic.employment.global_employment.on_employment_change import (
        on_employment_change,
    )

    on_employment_change(
        employment_change, event_bus_orchestrator=event_bus_orchestrator
    )

components.es.public.enrollment

BaseEnrollment dataclass

BaseEnrollment(
    id,
    user_id,
    type,
    beneficiary_type,
    subscription_ref,
    is_individual,
    start_date=isodate_field(),
    version_start_date=isodate_field(),
    coverage_start_date=isodate_field(),
    end_date=optional_isodate_field(),
    is_cancelled=False,
    insurance_configuration=None,
    healthy_benefits_configuration=None,
)

Bases: DataClassJsonMixin

beneficiary_type instance-attribute

beneficiary_type

coverage_start_date class-attribute instance-attribute

coverage_start_date = isodate_field()

end_date class-attribute instance-attribute

end_date = optional_isodate_field()

healthy_benefits_configuration class-attribute instance-attribute

healthy_benefits_configuration = None

id instance-attribute

id

insurance_configuration class-attribute instance-attribute

insurance_configuration = None

is_cancelled class-attribute instance-attribute

is_cancelled = False

is_health_insurance_product

is_health_insurance_product()

Source code in components/es/internal/business_logic/enrollment/entities/base_enrollment.py

def is_health_insurance_product(self) -> bool:
    return self.type in [BaseEnrollmentType.manzana, BaseEnrollmentType.fresa]

is_healthy_benefit

is_healthy_benefit()

Source code in components/es/internal/business_logic/enrollment/entities/base_enrollment.py

def is_healthy_benefit(self) -> bool:
    return self.type in [
        BaseEnrollmentType.food,
        BaseEnrollmentType.nursery,
        BaseEnrollmentType.transport,
        BaseEnrollmentType.learning,
    ]

is_individual instance-attribute

is_individual

start_date class-attribute instance-attribute

start_date = isodate_field()

subscription_ref instance-attribute

subscription_ref

type instance-attribute

type

user_id instance-attribute

user_id

version_start_date class-attribute instance-attribute

version_start_date = isodate_field()

BaseEnrollmentType

Bases: AlanBaseEnum

alan_mind class-attribute instance-attribute

alan_mind = 'alan_mind'

food class-attribute instance-attribute

food = 'food'

fresa class-attribute instance-attribute

fresa = 'alan_fresa'

learning class-attribute instance-attribute

learning = 'learning'

manzana class-attribute instance-attribute

manzana = 'alan_manzana'

nursery class-attribute instance-attribute

nursery = 'nursery'

optical class-attribute instance-attribute

optical = 'optical'

pharmacy class-attribute instance-attribute

pharmacy = 'pharmacy'

physio_nutrition class-attribute instance-attribute

physio_nutrition = 'physio_nutrition'

pre_existing_conditions class-attribute instance-attribute

pre_existing_conditions = 'pec'

reimbursements class-attribute instance-attribute

reimbursements = 'reimbursements'

transport class-attribute instance-attribute

transport = 'transport'

get_user_active_base_enrollments

get_user_active_base_enrollments(
    user_id, at_date=None, include_dependents=False
)

Return a list of user active enrollments, to insurance, or any healthy benefit. Enrollments starting in the future are NOT returned. If include_dependents is True, all enrollments on the user's insurance policy are returned. Note: we don't have healthy benefits for dependents yet.

Source code in components/es/internal/business_logic/enrollment/queries/base_enrollment.py

def get_user_active_base_enrollments(
    user_id: UUID, at_date: date | None = None, include_dependents: bool = False
) -> list[BaseEnrollment]:
    """
    Return a list of user active enrollments, to insurance, or any healthy benefit.
    Enrollments starting in the future are NOT returned.
    If include_dependents is True, all enrollments on the user's insurance policy are returned.
    Note: we don't have healthy benefits for dependents yet.
    """
    return get_user_base_enrollments(
        user_id,
        with_future=False,
        at_date=at_date,
        include_dependents=include_dependents,
    )

components.es.public.entities

AddOnType

Bases: AlanBaseEnum

ALAN_MIND class-attribute instance-attribute

ALAN_MIND = ALAN_MIND

OPTICAL class-attribute instance-attribute

OPTICAL = OPTICAL

PHARMACY class-attribute instance-attribute

PHARMACY = PHARMACY

PHYSIO_NUTRITION class-attribute instance-attribute

PHYSIO_NUTRITION = PHYSIO_NUTRITION

PRE_EXISTING_CONDITIONS class-attribute instance-attribute

PRE_EXISTING_CONDITIONS = PRE_EXISTING_CONDITIONS

REIMBURSEMENTS class-attribute instance-attribute

REIMBURSEMENTS = REIMBURSEMENTS

BaseEnrollmentType

Bases: AlanBaseEnum

alan_mind class-attribute instance-attribute

alan_mind = 'alan_mind'

food class-attribute instance-attribute

food = 'food'

fresa class-attribute instance-attribute

fresa = 'alan_fresa'

learning class-attribute instance-attribute

learning = 'learning'

manzana class-attribute instance-attribute

manzana = 'alan_manzana'

nursery class-attribute instance-attribute

nursery = 'nursery'

optical class-attribute instance-attribute

optical = 'optical'

pharmacy class-attribute instance-attribute

pharmacy = 'pharmacy'

physio_nutrition class-attribute instance-attribute

physio_nutrition = 'physio_nutrition'

pre_existing_conditions class-attribute instance-attribute

pre_existing_conditions = 'pec'

reimbursements class-attribute instance-attribute

reimbursements = 'reimbursements'

transport class-attribute instance-attribute

transport = 'transport'

ES_DEFAULT_SESSIONS_PER_MEMBER module-attribute

ES_DEFAULT_SESSIONS_PER_MEMBER = 20

EsCompany dataclass

EsCompany(id, display_name)

Bases: DataClassJsonMixin

display_name instance-attribute

display_name

id instance-attribute

id

EsUserDetails dataclass

EsUserDetails(
    id,
    first_name,
    last_name,
    email,
    birth_date=optional_isodate_field(),
    gender=None,
    insurance_profile_id=None,
)

Bases: DataClassJsonMixin

birth_date class-attribute instance-attribute

birth_date = optional_isodate_field()

email instance-attribute

email

first_name instance-attribute

first_name

gender class-attribute instance-attribute

gender = None

id instance-attribute

id

insurance_profile_id class-attribute instance-attribute

insurance_profile_id = None

last_name instance-attribute

last_name

ManzanaServiceConfigurationKey

Bases: AlanBaseEnum

free_therapy_sessions_limit class-attribute instance-attribute

free_therapy_sessions_limit = 'free_therapy_sessions_limit'

has_copayments class-attribute instance-attribute

has_copayments = 'has_copayments'

has_waiting_periods class-attribute instance-attribute

has_waiting_periods = 'has_waiting_periods'

waiting_periods_from_date class-attribute instance-attribute

waiting_periods_from_date = 'waiting_periods_from_date'

OptionType

Bases: AlanBaseEnum

ALAN_FRESA class-attribute instance-attribute

ALAN_FRESA = 'alan_fresa'

ALAN_MANZANA class-attribute instance-attribute

ALAN_MANZANA = 'alan_manzana'

ALAN_MIND class-attribute instance-attribute

ALAN_MIND = 'alan_mind'

CUN class-attribute instance-attribute

CUN = 'cun'

OPTICAL class-attribute instance-attribute

OPTICAL = 'optical'

PHARMACY class-attribute instance-attribute

PHARMACY = 'pharmacy'

PHYSIO_NUTRITION class-attribute instance-attribute

PHYSIO_NUTRITION = 'physio_nutrition'

PRE_EXISTING_CONDITIONS class-attribute instance-attribute

PRE_EXISTING_CONDITIONS = 'pec'

REIMBURSEMENTS class-attribute instance-attribute

REIMBURSEMENTS = 'reimbursements'

ProductServiceConfigurationKey module-attribute

ProductServiceConfigurationKey = (
    ManzanaServiceConfigurationKey
)

ProductType

Bases: AlanBaseEnum

ALAN_FRESA class-attribute instance-attribute

ALAN_FRESA = ALAN_FRESA

ALAN_MANZANA class-attribute instance-attribute

ALAN_MANZANA = ALAN_MANZANA

name property

name

components.es.public.events

subscription

subscribe_to_es_global_events

subscribe_to_es_global_events()

Events subscriptions that should be listened by every runtime for Spain.

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

def subscribe_to_es_global_events() -> None:
    """
    Events subscriptions that should be listened by every runtime for Spain.
    """
    from components.es.internal.events.subscribers import (
        update_email_and_or_address_in_dkv,
    )
    from components.es.internal.queue_name import DKV_QUEUE
    from components.global_profile.public.events import (
        ProfileAddressChanged,
        ProfileEmailChanged,
    )
    from shared.messaging.broker import get_message_broker

    message_broker = get_message_broker()

    message_broker.subscribe_async(
        ProfileAddressChanged,
        update_email_and_or_address_in_dkv,
        queue_name=DKV_QUEUE,
    )
    message_broker.subscribe_async(
        ProfileEmailChanged,
        update_email_and_or_address_in_dkv,
        queue_name=DKV_QUEUE,
    )

subscribe_to_events

subscribe_to_events()

All event subscriptions for Spain should be done here.

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

def subscribe_to_events() -> None:
    """
    All event subscriptions for Spain should be done here.
    """
    from components.ca.public.events.subscription import subscribe_to_ca_global_events
    from components.es.internal.business_logic.concierge.register_to_voice_ai import (
        register_voice_ai_topic_subscribers,
    )

    subscribe_to_es_global_events()
    subscribe_to_ca_global_events()
    register_voice_ai_topic_subscribers()

components.es.public.feature

get_feature_metadata_value

get_feature_metadata_value(feature_name, key, default=None)

Source code in components/es/public/feature.py

def get_feature_metadata_value(  # noqa: D103
    feature_name: str,
    key: str,
    default: Any | None = None,
) -> Any | None:
    from components.es.internal.business_logic.feature import (  # noqa: TID251
        feature_logic,
    )

    return feature_logic.get_feature_metadata_value(
        feature_name=feature_name, key=key, default=default
    )

is_feature_active

is_feature_active(feature_name)

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

Source code in components/es/public/feature.py

def is_feature_active(
    feature_name: str,
) -> bool:
    """
    Check if a feature is active
    :param feature_name: the name of the feature
    :return: true if the feature with this name is found and active
    """
    from components.es.internal.business_logic.feature import (  # noqa: TID251
        FeatureLogic,
    )

    return FeatureLogic.is_feature_active(feature_name=feature_name)

is_user_alaner

is_user_alaner(user_id)

Checks if a user is an Alan employee based on their user ID.

In the Spanish context, a user is considered an Alan employee if either: - They have an active AlanEmployee entry associated with their account - They have an active employment contract with "Marmot Iberia company" (Alan Spain)

This dual check is necessary because Alan employees typically have two accounts: 1. An AlanEmployee account 2. A personal account for using the Alan App

Parameters:

Name	Type	Description	Default
user_id	UUID	The unique identifier of the user to check	required

Returns:

Name	Type	Description
bool	bool	True if the user is an Alan employee, False otherwise

Note

The function performs a database query joining the EsUser table with EsAlanEmployee and EsEmployment tables to check both conditions.

Source code in components/es/public/feature.py

@cached_for(seconds=5)
def is_user_alaner(user_id: HybridId) -> bool:
    """
    Checks if a user is an Alan employee based on their user ID.

    In the Spanish context, a user is considered an Alan employee if either:
    - They have an active AlanEmployee entry associated with their account
    - They have an active employment contract with "Marmot Iberia company" (Alan Spain)

    This dual check is necessary because Alan employees typically have two accounts:
    1. An AlanEmployee account
    2. A personal account for using the Alan App

    Args:
        user_id (uuid.UUID): The unique identifier of the user to check

    Returns:
        bool: True if the user is an Alan employee, False otherwise

    Note:
        The function performs a database query joining the EsUser table with
        EsAlanEmployee and EsEmployment tables to check both conditions.
    """
    from components.es.internal.models.es_alan_employee import EsAlanEmployee
    from components.es.internal.models.es_employment import EsEmployment
    from components.es.internal.models.es_user import EsUser

    if isinstance(user_id, int):
        raise BaseErrorCode.invalid_arguments(
            message="In Spain user ids should be UUIDs", argument=user_id
        )

    # Raise if user does not exist
    user_service.get_or_raise_user(user_id)
    return cast(
        "bool",
        current_session.scalar(
            select(
                exists(
                    select(EsUser.id)
                    .outerjoin(EsUser.alan_employee)
                    .outerjoin(EsUser.employments)
                    .filter(
                        EsUser.id == user_id,
                        or_(
                            EsUser.alan_employee.has(EsAlanEmployee.is_active),
                            and_(
                                EsEmployment.is_active,
                                EsEmployment.company_id == MARMOT_IBERIA_COMPANY_ID,
                            ),
                        ),
                    )
                )
            )
        ),
    )

components.es.public.global_customer_dashboard

admin

AdminInvitation dataclass

AdminInvitation(company_id, invitation_id, invite_email)

company_id instance-attribute

company_id

invitation_id instance-attribute

invitation_id

invite_email instance-attribute

invite_email

CAN_INVITE_ADMINS_DATA module-attribute

CAN_INVITE_ADMINS_DATA = [
    {
        "company": "eb3eca81-52bf-4906-8bb8-5e36a2fc90fe",
        "enabledForUsers": [
            "809bf2c1-ae3b-4be8-be93-ac8509c2e1e2",
            "4825942c-8f57-44ec-b50b-49d431c48a21",
            "67859a35-30f3-47fb-bba7-df5340a60628",
            "c34243b3-6743-41a7-9a98-916481213892",
            "c43c2323-cf93-4f74-a552-c63fa0b0c185",
        ],
    },
    {
        "company": "af18ef59-0772-440f-8dd9-6caf506415b3",
        "enabledForUsers": [
            "809bf2c1-ae3b-4be8-be93-ac8509c2e1e2",
            "4825942c-8f57-44ec-b50b-49d431c48a21",
            "67859a35-30f3-47fb-bba7-df5340a60628",
            "c34243b3-6743-41a7-9a98-916481213892",
            "c43c2323-cf93-4f74-a552-c63fa0b0c185",
        ],
    },
    {
        "company": "375b96ba-9b3b-4ef7-9149-6aab10eb3bf5",
        "enabledForUsers": [
            "809bf2c1-ae3b-4be8-be93-ac8509c2e1e2",
            "4825942c-8f57-44ec-b50b-49d431c48a21",
            "67859a35-30f3-47fb-bba7-df5340a60628",
            "c34243b3-6743-41a7-9a98-916481213892",
            "c43c2323-cf93-4f74-a552-c63fa0b0c185",
        ],
    },
    {
        "company": "f640ef11-7204-4716-9a58-4e58c69103a6",
        "enabledForUsers": [
            "809bf2c1-ae3b-4be8-be93-ac8509c2e1e2",
            "4825942c-8f57-44ec-b50b-49d431c48a21",
            "67859a35-30f3-47fb-bba7-df5340a60628",
            "c34243b3-6743-41a7-9a98-916481213892",
            "c43c2323-cf93-4f74-a552-c63fa0b0c185",
        ],
    },
    {
        "company": "e1e611b7-b04f-41e0-824c-0fc5ce1c6ff6",
        "enabledForUsers": [
            "809bf2c1-ae3b-4be8-be93-ac8509c2e1e2",
            "4825942c-8f57-44ec-b50b-49d431c48a21",
            "67859a35-30f3-47fb-bba7-df5340a60628",
            "c34243b3-6743-41a7-9a98-916481213892",
            "c43c2323-cf93-4f74-a552-c63fa0b0c185",
        ],
    },
    {
        "company": "cd52c2b0-818c-4774-8eb7-8c9b45af0e97",
        "enabledForUsers": [
            "809bf2c1-ae3b-4be8-be93-ac8509c2e1e2",
            "4825942c-8f57-44ec-b50b-49d431c48a21",
            "67859a35-30f3-47fb-bba7-df5340a60628",
            "c34243b3-6743-41a7-9a98-916481213892",
            "c43c2323-cf93-4f74-a552-c63fa0b0c185",
        ],
    },
    {
        "company": "3f57c053-470d-4704-b5fd-d3f04bf1c033",
        "enabledForUsers": [
            "809bf2c1-ae3b-4be8-be93-ac8509c2e1e2",
            "4825942c-8f57-44ec-b50b-49d431c48a21",
            "67859a35-30f3-47fb-bba7-df5340a60628",
            "c34243b3-6743-41a7-9a98-916481213892",
            "c43c2323-cf93-4f74-a552-c63fa0b0c185",
        ],
    },
    {
        "company": "b82212d5-ed23-42a8-b483-4f443ec81b2c",
        "enabledForUsers": [
            "809bf2c1-ae3b-4be8-be93-ac8509c2e1e2",
            "4825942c-8f57-44ec-b50b-49d431c48a21",
            "67859a35-30f3-47fb-bba7-df5340a60628",
            "c34243b3-6743-41a7-9a98-916481213892",
            "c43c2323-cf93-4f74-a552-c63fa0b0c185",
        ],
    },
]

edit_admin_entities

edit_admin_entities(
    admin_user_id,
    added_company_ids,
    removed_company_ids,
    save=True,
)

Source code in components/es/public/global_customer_dashboard/admin.py

def edit_admin_entities(  # noqa: D103
    admin_user_id: uuid.UUID,
    added_company_ids: set[uuid.UUID],
    removed_company_ids: set[uuid.UUID],
    save: bool = True,
) -> None:
    check_resource_exists_or_raise(EsUser, admin_user_id)

    _remove_admin_from_companies(
        admin_user_id=admin_user_id, removed_company_ids=removed_company_ids
    )

    for company_id in added_company_ids:
        check_resource_exists_or_raise(EsCompany, company_id)
        if not company_logic.user_is_admin(
            user_id=admin_user_id, company_id=company_id
        ):
            company_admin = EsCompanyAdmin(
                user_id=admin_user_id,
                company_id=company_id,
            )
            current_session.add(company_admin)

    if save:
        current_session.commit()

get_active_or_future_subscriptions_for_company

get_active_or_future_subscriptions_for_company(
    company_id, at_date
)

Source code in components/es/public/global_customer_dashboard/admin.py

def get_active_or_future_subscriptions_for_company(  # noqa: D103
    company_id: uuid.UUID, at_date: date | None
) -> list[CustomerSubscription]:
    current_logger.info(
        "Get active or future subscription for company",
        company_id=company_id,
        at_date=at_date,
    )
    from components.es.internal.business_logic.company.subscription import (
        get_all_company_subscriptions,
    )

    all_subscriptions = get_all_company_subscriptions(company_id, at_date=at_date)

    company_subscriptions: list[CustomerSubscription] = []
    if len(all_subscriptions.health_insurance) > 0:
        company_subscriptions.append(
            _health_insurance_subscription_to_dashboard_subscription(
                all_subscriptions.health_insurance[0]
            )
        )

    if len(all_subscriptions.healthy_benefits) > 0:
        company_subscriptions.append(
            healthy_benefits_subscription_to_customer_subscription(
                all_subscriptions.healthy_benefits[0]
            )
        )

    return company_subscriptions

get_admin

get_admin(admin_user_id, user_id)

Source code in components/es/public/global_customer_dashboard/admin.py

def get_admin(  # noqa: D103
    admin_user_id: uuid.UUID, user_id: uuid.UUID
) -> CustomerDashboardCompanyAdmin | None:
    admin_user_companies = (
        select(EsCompanyAdmin.company_id)
        .filter(EsCompanyAdmin.user_id == admin_user_id, EsCompanyAdmin.is_not_ended)
        .cte("admin_user_companies")
    )

    result = current_session.execute(
        select(EsUser, EsCompanyAdmin)
        .join(
            EsCompanyAdmin
        )  # the left join allows us to get None if the EsUser doesn't exist
        .filter(
            EsCompanyAdmin.user_id == user_id,
            EsCompanyAdmin.is_not_ended,
            EsCompanyAdmin.company_id.in_(select(admin_user_companies.c.company_id)),
        )
        .limit(1)
    ).first()

    if result is None:
        return None
    _, company_admin = result

    admin_user_profile = user_service.get_or_raise_user_profile(user_id)
    email = get_privacy_compliant_email_for_user(
        company_admin_id=company_admin.id, company_id=company_admin.company_id
    )

    return CustomerDashboardCompanyAdmin(
        id=str(user_id),
        first_name=admin_user_profile.first_name,
        last_name=admin_user_profile.last_name or "",
        email=email,
        type=EntityType.company,
    )

get_admined_entities_for_entity_selector_es

get_admined_entities_for_entity_selector_es(user_id)

Retrieves a list of admined entities for a given user in the context of the Entity Selector (ES).

This function determines whether the 'admin_accounts' feature is enabled for the user and, based on that, fetches the admined entities accordingly, using either single companies only or including accounts.

Parameters:

Name	Type	Description	Default
user_id	str	The unique identifier of the user for whom admined entities are being retrieved.	required

Returns:

Type	Description
list[AdminedEntityForEntitySelector]	list[AdminedEntityForEntitySelector]: A list of admined entities that the user has administrative access to
list[AdminedEntityForEntitySelector]	within the Entity Selector context.

Source code in components/es/public/global_customer_dashboard/admin.py

def get_admined_entities_for_entity_selector_es(
    user_id: str,
) -> list[AdminedEntityForEntitySelector]:
    """
    Retrieves a list of admined entities for a given user in the context of the Entity Selector (ES).

    This function determines whether the 'admin_accounts' feature is enabled for the user and, based on that,
    fetches the admined entities accordingly, using either single companies only or including accounts.

    Args:
        user_id (str): The unique identifier of the user for whom admined entities are being retrieved.

    Returns:
        list[AdminedEntityForEntitySelector]: A list of admined entities that the user has administrative access to
        within the Entity Selector context.
    """
    from components.es.internal.business_logic.feature import (  # noqa: TID251
        feature_logic,
    )

    # For now, accounts are disable for ES
    # https://alanhealth.slack.com/archives/C0776E58QBA/p1718377415099289?thread_ts=1718203474.042239&cid=C0776E58QBA
    is_accounts_enabled_for_user = feature_logic.is_feature_name_enabled_for_user_id(
        user_id=uuid.UUID(user_id), feature_name="admin_accounts"
    )

    admined_entities = get_admined_entities_for_entity_selector_global(
        user_id=user_id,
        operational_scope_model=None,
        company_model=EsCompany,
        account_model=EsAccount,
        admined_entities_query_api=get_admined_entities_query_api_es(),
        exclude_accounts=not is_accounts_enabled_for_user,
    )

    # We filter entities that are actively admined by the user (with an ongoing subscription)
    user_actively_admined_companies_ids = get_user_actively_admined_companies_ids(
        user_id=user_id
    )

    filtered_entities: list[AdminedEntityForEntitySelector] = []
    for entity in admined_entities:
        if entity.type == AdminedEntityType.single_company:
            if str(entity.id) in user_actively_admined_companies_ids:
                filtered_entities.append(entity)
        else:
            company_ids = cast("set[uuid.UUID]", entity.companies_ids)
            entity.companies_ids = {
                company_id
                for company_id in company_ids
                if str(company_id) in user_actively_admined_companies_ids
            }

            filtered_entities.append(entity)

    return filtered_entities

get_admined_entities_query_api_es

get_admined_entities_query_api_es()

Get Spain concrete instance of AdminedEntitiesQueryApi, bound to ES models

Source code in components/es/public/global_customer_dashboard/admin.py

def get_admined_entities_query_api_es() -> AdminedEntitiesQueryApi:
    """
    Get Spain concrete instance of AdminedEntitiesQueryApi, bound to ES models
    """
    return AdminedEntitiesQueryApi(
        company_admin_model=EsCompanyAdmin,
        account_admin_model=EsAccountAdmin,
        get_account_id_to_company_ids_fn=get_account_id_to_company_ids,
        get_user_active_admined_companies_ids_fn=get_user_actively_admined_companies_ids,
    )

get_admined_entity_from_id_es

get_admined_entity_from_id_es(*, user_id, entity_id)

Source code in components/es/public/global_customer_dashboard/admin.py

def get_admined_entity_from_id_es(  # noqa: D103
    *,
    user_id: str,
    entity_id: str,
) -> AdminedEntityForEntitySelector:
    admin_entities = get_admined_entities_for_entity_selector_global(
        user_id=user_id,
        operational_scope_model=None,
        company_model=EsCompany,
        account_model=EsAccount,
        admined_entities_query_api=get_admined_entities_query_api_es(),
    )

    # We only care about the id property here, since if the entity is a company, that's the company id,
    # and if the entity is an account, we fill its id field with the account id
    admin_entity = list(
        filter(lambda entity: str(entity.id) == entity_id, admin_entities)
    )

    if len(admin_entity) != 1:
        raise BaseErrorCode.forbidden(
            message="The queried entity id should correspond to an entity the user administrates"
        )

    return admin_entity[0]

get_admins_for_companies

get_admins_for_companies(
    company_ids,
    sort_filter,
    sort_direction,
    cursor=None,
    limit=None,
)

Returns the list of admins to show on the global customer dashboard, for the companies provided as input.

WARNING: This method does not take into account any potential EsAccountAdmin.

Source code in components/es/public/global_customer_dashboard/admin.py

def get_admins_for_companies(
    company_ids: Iterable[uuid.UUID],
    sort_filter: CustomerDashboardCompanyAdminListSortField,
    sort_direction: PaginationSortDirectionType,
    cursor: int | None = None,
    limit: int | None = None,
) -> PaginatedAdminsForAdminDashboard:
    """
    Returns the list of admins to show on the global customer dashboard, for the companies provided as input.

    WARNING: This method does not take into account any potential EsAccountAdmin.
    """
    admins_with_companies = (
        EsCompanyAdminModelBroker.get_admins_for_companies_paginated(
            company_ids, sort_filter, sort_direction, cursor, limit
        )
    )
    admin_user_profile = user_service.get_or_raise_user_profiles(
        {es_user.id for es_user, _ in admins_with_companies}
    )

    user_id_to_invitation_email = get_invitation_email_for_users_and_entities(
        {
            str(user.id): {
                (str(company_id), EntityType.company)
                for company_id in admined_company_ids
            }
            for user, admined_company_ids in admins_with_companies
        }
    )

    res = []
    for user, admined_company_ids in admins_with_companies:
        admined_company_ids_as_set = set(admined_company_ids)

        email = user_id_to_invitation_email[str(user.id)]
        res.append(
            CustomerDashboardAdminWithEntities(
                id=str(user.id),
                first_name=admin_user_profile[user.id].first_name,
                last_name=admin_user_profile[user.id].last_name or "",
                email=email,
                type=EntityType.company,
                company_ids=admined_company_ids_as_set,
            )
        )
    return PaginatedAdminsForAdminDashboard(
        admins=res,
        meta=PageInfoMeta(
            cursor=cursor,
            has_next=len(res) == limit if limit is not None else None,
        ),
    )

get_common_companies_for_admins

get_common_companies_for_admins(
    admin_user_id, target_admin_user_id
)

Returns the companies that two admins have in common. If the two admins are the same, it returns all the companies the admin has access to. Else it returns the companies that both admins have access to.

Source code in components/es/public/global_customer_dashboard/admin.py

def get_common_companies_for_admins(
    admin_user_id: uuid.UUID, target_admin_user_id: uuid.UUID
) -> set[CustomerDashboardCompany]:
    """
    Returns the companies that two admins have in common.
    If the two admins are the same, it returns all the companies the admin has access to.
    Else it returns the companies that both admins have access to.
    """
    check_resource_exists_or_raise(EsUser, admin_user_id)

    company_admins_for_user = EsCompanyAdminModelBroker.get_company_admins_for_user(
        user_id=admin_user_id, with_joined_load_company=True
    )
    admin_companies = [
        company_admin.company for company_admin in company_admins_for_user
    ]

    admin_companies_dataclasses = {
        _company_to_customer_dashboard_company(company=company)
        for company in admin_companies
    }

    if admin_user_id == target_admin_user_id:
        return admin_companies_dataclasses

    check_resource_exists_or_raise(EsUser, target_admin_user_id)

    company_admins_for_target_user = (
        EsCompanyAdminModelBroker.get_company_admins_for_user(
            user_id=target_admin_user_id, with_joined_load_company=True
        )
    )
    target_admin_companies = [
        company_admin.company for company_admin in company_admins_for_target_user
    ]

    target_admin_companies_dataclasses = {
        _company_to_customer_dashboard_company(company=company)
        for company in target_admin_companies
    }
    common_admined_companies = admin_companies_dataclasses.intersection(
        target_admin_companies_dataclasses
    )

    return common_admined_companies

get_companies_by_account_id

get_companies_by_account_id(account_id)

Return list of companies ids, within a given account id

Parameters: - account_id (UUID): The unique identifier of the account for which companies are to be retrieved.

Raises: - BaseError.missing_resource: If the account with the given account_id does not exist or cannot be found.

Source code in components/es/public/global_customer_dashboard/admin.py

def get_companies_by_account_id(account_id: uuid.UUID) -> list[uuid.UUID]:  # noqa: D417
    """
     Return list of companies ids, within a given account id

     Parameters:
    - account_id (UUID): The unique identifier of the account for which companies are to be retrieved.

     Raises:
    - BaseError.missing_resource: If the account with the given `account_id` does not exist or cannot be found.
    """
    from components.es.internal.business_logic.account.account import (
        get_companies_in_account,
    )

    return [company.id for company in get_companies_in_account(account_id)]

get_company_details

get_company_details(company_id)

Source code in components/es/public/global_customer_dashboard/admin.py

def get_company_details(company_id: uuid.UUID) -> CustomerDashboardCompany:  # noqa: D103
    company = get_or_raise_missing_resource(EsCompany, company_id)

    return _company_to_customer_dashboard_company(company=company)

get_company_signed_documents

get_company_signed_documents(company_id, at_date)

For now in Spain we have 2 different subscriptions for Insurance and Healthy benefits This function returns list of documents for both types of subscription

Source code in components/es/public/global_customer_dashboard/admin.py

def get_company_signed_documents(
    company_id: uuid.UUID, at_date: date | None
) -> list[SignedDocument]:
    """
    For now in Spain we have 2 different subscriptions for Insurance and Healthy benefits
    This function returns list of documents for both types of subscription
    """
    _at_date = at_date or date.today()

    from components.es.internal.business_logic.company.subscription import (
        get_company_healthy_benefits_signed_documents,
        get_company_insurance_signed_documents,
    )

    insurance_documents = get_company_insurance_signed_documents(
        company_id=company_id, at_date=_at_date
    )

    healthy_benefits_documents = get_company_healthy_benefits_signed_documents(
        company_id=company_id, at_date=_at_date
    )

    return [
        _insurance_signed_document_to_signed_document(document)
        for document in insurance_documents
    ] + [
        _healthy_benefits_signed_document_to_signed_document(document)
        for document in healthy_benefits_documents
    ]

get_invitation_details

get_invitation_details(invitation_id)

Returns the details of a pending admin invitation, by ID

Source code in components/es/public/global_customer_dashboard/admin.py

def get_invitation_details(
    invitation_id: uuid.UUID,
) -> CustomerDashboardCompanyAdminInvitationWithCompanies:
    """
    Returns the details of a pending admin invitation, by ID
    """
    invitations_query = current_session.query(EsCompanyAdminInvitation).filter(  # noqa: ALN085
        EsCompanyAdminInvitation.id == invitation_id,
        EsCompanyAdminInvitation.company_admin_id.is_(None),
    )
    invitation = invitations_query.first()
    if not invitation:
        raise BaseErrorCode.missing_resource(
            message=f"EsCompanyAdminInvitation '{invitation_id}' does not exist"
        )

    return CustomerDashboardCompanyAdminInvitationWithCompanies(
        id=str(invitation.id),
        invite_email=mandatory(invitation.invite_email),
        invitation_date=invitation.created_at.date(),
        responsibilities=[],
        companies=[_company_to_customer_dashboard_company(company=invitation.company)],
        type=AdminedEntityType.single_company,
    )

get_onboarded_employees

get_onboarded_employees(
    company_ids, cursor, limit, search=None
)

Source code in components/es/public/global_customer_dashboard/admin.py

def get_onboarded_employees(  # noqa: D103
    company_ids: list[uuid.UUID],
    cursor: int,
    limit: int,
    search: str | None = None,
) -> list[Employee]:
    for company_id in company_ids:
        if not company_logic.user_is_admin(
            user_id=g.current_user.id, company_id=company_id
        ):
            raise BaseErrorCode.forbidden(
                message="You must be a company admin to access this resource"
            )

    employees_data: list[BaseEmployeeDetails] = get_employees_details(
        company_ids=company_ids,
        filter_status=EmployeeFilterStatus.covered,
        filter_keywords=[search] if search else None,
        cursor=cursor,
        limit=limit,
    )

    employees: list[Employee] = [
        Employee(
            user_id=str(employee_data.id),
            first_name=employee_data.first_name,
            last_name=employee_data.last_name,
            employment_id=mandatory(employee_data.employment_id),
            company_id=str(employee_data.company.id),
            invite_email=employee_data.email,
            employee_type=EmployeeType.insured,
            external_employee_id=employee_data.payroll_id,
        )
        for employee_data in employees_data
    ]

    return employees

get_or_download_signed_document

get_or_download_signed_document(
    company_id, document_id, document_type
)

Source code in components/es/public/global_customer_dashboard/admin.py

def get_or_download_signed_document(  # noqa: D103
    company_id: uuid.UUID, document_id: uuid.UUID, document_type: SignedDocumentType
) -> DownloadFile:
    current_logger.info(
        "Downloading signed document for company",
        company_id=company_id,
        document_id=document_id,
        document_type=document_type,
    )
    if (
        document_type == SignedDocumentType.healthy_benefits_particular_conditions
        or document_type == SignedDocumentType.healthy_benefits_general_conditions
    ):
        from components.es.internal.business_logic.company.subscription import (
            get_or_download_healthy_benefits_signed_document,
        )

        file_name, file = get_or_download_healthy_benefits_signed_document(
            document_id=document_id, company_id=company_id
        )
    else:
        from components.es.internal.business_logic.company.subscription import (
            get_or_download_insurance_signed_document,
        )

        file_name, file = get_or_download_insurance_signed_document(
            document_id=document_id, company_id=company_id
        )

    return DownloadFile(file_name=file_name, file_to_download=file)

get_pending_admin_invitations_for_companies

get_pending_admin_invitations_for_companies(
    company_ids,
    sort_filter,
    sort_direction,
    cursor=None,
    limit=None,
)

Source code in components/es/public/global_customer_dashboard/admin.py

def get_pending_admin_invitations_for_companies(  # noqa: D103
    company_ids: Iterable[uuid.UUID],
    sort_filter: CustomerDashboardCompanyAdminInvitationsSortField,
    sort_direction: PaginationSortDirectionType,
    cursor: int | None = None,
    limit: int | None = None,
) -> "PaginatedCustomerDashboardAdminInvitation":
    invitations_query = current_session.query(EsCompanyAdminInvitation).filter(  # noqa: ALN085
        EsCompanyAdminInvitation.company_id.in_(company_ids),
        EsCompanyAdminInvitation.company_admin_id.is_(None),
        EsCompanyAdminInvitation.cancelled_at.is_(None),
    )
    invitations_query = paginate_query(
        query=invitations_query,
        sort_filter=sort_filter,
        sort_direction=sort_direction,
        cursor=cursor,
        limit=limit,
        sort_model=EsCompanyAdminInvitation,
    )
    invitations_result = [
        CustomerDashboardAdminInvitationWithEntityIds(
            id=str(invitation.id),
            invite_email=invitation.invite_email,  # type: ignore[arg-type]
            invitation_date=invitation.created_at,
            responsibilities=[],
            type=AdminedEntityType.single_company,
            account_id=None,
            company_ids=[invitation.company_id],
        )
        for invitation in invitations_query
    ]

    return PaginatedCustomerDashboardAdminInvitation(
        pending_invitations=invitations_result,
        meta=PageInfoMeta(
            cursor=cursor,
            has_next=len(invitations_result) == limit if limit is not None else None,
        ),
    )

get_pending_onboardings_for_admined_entity_selector_es

get_pending_onboardings_for_admined_entity_selector_es(
    user_id,
)

Source code in components/es/public/global_customer_dashboard/admin.py

def get_pending_onboardings_for_admined_entity_selector_es(  # noqa: D103
    user_id: uuid.UUID,
) -> list[PendingOnboarding]:
    from components.es.internal.business_logic.company_onboarding.company_onboarding_logic import (
        company_onboarding_logic,
    )

    company_onboardings = company_onboarding_logic.get_company_onboardings_for_admin(
        user_id
    )

    pending_onboardings = []
    for company_onboarding in company_onboardings:
        if company_onboarding.onboarding_complete is False:
            pending_onboardings.append(
                PendingOnboarding(
                    id=company_onboarding.id,
                    display_name=company_onboarding.company_name,
                    onboarding_url=company_onboarding.onboarding_url,
                    type=PendingOnboardingType.es_company_onboarding,
                    company_id=company_onboarding.company_id,  # type: ignore[arg-type]
                )
            )

    return pending_onboardings

get_user_actively_admined_companies_ids

get_user_actively_admined_companies_ids(user_id)

Returns the list of company admins that user currently administrates. Companies with non active subscription are not included.

Source code in components/es/public/global_customer_dashboard/admin.py

def get_user_actively_admined_companies_ids(user_id: str) -> set[str]:
    """
    Returns the list of company admins that user currently administrates.
    Companies with non active subscription are not included.
    """
    # We filter entities that are actively admined by the user (with an ongoing subscription)
    user_actively_admined_companies = get_user_admined_companies(
        user_id=uuid.UUID(user_id), only_active_or_future=True
    )
    return {str(company.id) for company in user_actively_admined_companies}

healthy_benefits_subscription_to_customer_subscription

healthy_benefits_subscription_to_customer_subscription(
    subscription,
)

Source code in components/es/public/global_customer_dashboard/admin.py

def healthy_benefits_subscription_to_customer_subscription(  # noqa: D103
    subscription: HealthyBenefitsSubscription,
) -> CustomerSubscription:
    return CustomerSubscription(
        id=str(subscription.id),
        company_id=str(subscription.owner_ref),
        start_date=subscription.start_date,
        end_date=subscription.end_date,
        scope=CustomerSubscriptionScope.healthy_benefits,
    )

promote_user_to_admin

promote_user_to_admin(company_ids, invited_user_id)

Source code in components/es/public/global_customer_dashboard/admin.py

def promote_user_to_admin(  # noqa: D103
    company_ids: list[uuid.UUID], invited_user_id: uuid.UUID
) -> None:
    for company_id in company_ids:
        if not company_logic.user_is_admin(
            user_id=g.current_user.id, company_id=company_id
        ):
            raise BaseErrorCode.forbidden(
                message="You must be a company admin to access this resource"
            )

    if (
        is_employee_in_one_of_companies(
            user_id=invited_user_id,
            company_ids=company_ids,
            not_ended_on_date=utctoday(),
        )
        is False
    ):
        raise BaseErrorCode.forbidden(message="User must an employee of the company")

    for company_id in company_ids:
        company_logic.promote_user_to_admin(
            company_id=company_id, invited_user_id=invited_user_id
        )

remove_admin

remove_admin(
    admin_user_id, target_admin_user_id, save=True
)

Source code in components/es/public/global_customer_dashboard/admin.py

def remove_admin(  # noqa: D103
    admin_user_id: uuid.UUID, target_admin_user_id: uuid.UUID, save: bool = True
) -> None:
    check_resource_exists_or_raise(EsUser, admin_user_id)
    check_resource_exists_or_raise(EsUser, target_admin_user_id)

    # remove admin rights on companies the current admin has access to:
    common_companies = get_common_companies_for_admins(
        admin_user_id=admin_user_id, target_admin_user_id=target_admin_user_id
    )
    common_company_ids = {uuid.UUID(company.id) for company in common_companies}
    _remove_admin_from_companies(
        admin_user_id=target_admin_user_id,
        removed_company_ids=common_company_ids,
    )

    if save:
        current_session.commit()

remove_admin_invitation

remove_admin_invitation(invitation_id, save=True)

Source code in components/es/public/global_customer_dashboard/admin.py

def remove_admin_invitation(invitation_id: uuid.UUID, save: bool = True) -> None:  # noqa: D103
    company_logic.cancel_admin_invitation(invitation_id=invitation_id, commit=save)

send_admin_invitation_email

send_admin_invitation_email(invitation_id)

Source code in components/es/public/global_customer_dashboard/admin.py

def send_admin_invitation_email(invitation_id: uuid.UUID) -> None:  # noqa: D103
    company_logic.send_admin_invitation_email(invitation_id=invitation_id)

user_can_admin_entities

user_can_admin_entities(user_id, company_ids, account_ids)

Source code in components/es/public/global_customer_dashboard/admin.py

def user_can_admin_entities(  # noqa: D103
    user_id: uuid.UUID,
    company_ids: set[uuid.UUID],
    account_ids: set[uuid.UUID],
) -> bool:
    check_resource_exists_or_raise(EsUser, user_id)

    can_admin_companies = company_logic.user_can_admin_all(
        user_id=user_id, company_ids=list(company_ids)
    )
    can_admin_accounts = current_session.query(EsAccountAdmin).filter(  # noqa: ALN085
        EsAccountAdmin.user_id == user_id,
        EsAccountAdmin.account_id.in_(list(account_ids)),
        EsAccountAdmin.is_not_ended,
    ).count() == len(account_ids)

    return can_admin_companies and can_admin_accounts

user_can_invite_admins

user_can_invite_admins(
    admin_user_id, company_id, account_id
)

Source code in components/es/public/global_customer_dashboard/admin.py

def user_can_invite_admins(  # noqa: D103
    admin_user_id: uuid.UUID, company_id: str | None, account_id: str | None
) -> bool:
    # account admins can always invite admins
    if account_id is not None:
        return True

    company_config = first_or_none(
        CAN_INVITE_ADMINS_DATA,
        lambda data, company_id=company_id: data.get("company") == company_id,  # type: ignore[misc]
    )

    # If the feature is NOT enabled for the company, the user can invite admins
    if company_config is None:
        return True

    enabled_for_users = company_config.get("enabledForUsers", [])

    if not enabled_for_users:
        return True

    # If the feature is enabled for the company, but not for the user, the user can't invite admins
    if str(admin_user_id) not in enabled_for_users:
        return False

    return True

customer_health_partner

get_company_ids_by_external_ids

get_company_ids_by_external_ids(external_ids)

Get company ids by external ids (Tax number: CIF also called NIF)

:param external_ids: set[str] - The external ids to get the company ids for. :return: dict[str, str | None] - A dictionary mapping external ids to company ids.

Source code in components/es/public/global_customer_dashboard/customer_health_partner.py

def get_company_ids_by_external_ids(external_ids: set[str]) -> dict[str, str | None]:
    """
    Get company ids by external ids (Tax number: CIF also called NIF)

    :param external_ids: set[str] - The external ids to get the company ids for.
    :return: dict[str, str | None] - A dictionary mapping external ids to company ids.
    """
    companies: list[EsCompany] = (
        current_session.query(EsCompany).filter(EsCompany.nif.in_(external_ids)).all()  # noqa: ALN085
    )
    nif_to_company_id: dict[str, str] = {
        company.nif: str(company.id) for company in companies
    }
    return {vat: nif_to_company_id.get(vat, None) for vat in external_ids}

get_company_name_from_companies

get_company_name_from_companies(company_ids)

Source code in components/es/public/global_customer_dashboard/customer_health_partner.py

def get_company_name_from_companies(company_ids: set[UUID]) -> str:  # noqa: D103
    companies = (
        current_session.query(EsCompany).filter(EsCompany.id.in_(company_ids)).all()  # noqa: ALN085
    )
    if len(companies) == 0:
        raise ValueError("No company found for company ids")
    return companies[0].name

get_company_names_by_ids

get_company_names_by_ids(*, company_ids)

Source code in components/es/public/global_customer_dashboard/customer_health_partner.py

def get_company_names_by_ids(*, company_ids: set[str]) -> dict[str, str]:  # noqa: D103
    companies = (
        current_session.query(EsCompany).filter(EsCompany.id.in_(company_ids)).all()  # noqa: ALN085
    )
    return {str(company.id): company.display_name for company in companies}

get_user_email_from_id

get_user_email_from_id(*, user_id)

Source code in components/es/public/global_customer_dashboard/customer_health_partner.py

def get_user_email_from_id(*, user_id: str) -> str | None:  # noqa: D103
    user_profile = user_service.get_user_profile(user_id)

    return user_profile.email if user_profile else None

get_user_info_from_id

get_user_info_from_id(*, user_id)

Source code in components/es/public/global_customer_dashboard/customer_health_partner.py

def get_user_info_from_id(  # noqa: D103
    *,
    user_id: str,
) -> WellbeingAssessmentDisplayUserInfo:
    user_profile = user_service.get_or_raise_user_profile(user_id)

    if user_profile.email is None:
        raise ValueError(f"User {user_id} has no email")

    return WellbeingAssessmentDisplayUserInfo(
        id=str(user_id),
        full_name=user_profile.full_name,
        name=user_profile.full_name,
        email=user_profile.email,
        normalized_full_name=normalize_name(user_profile.full_name),
        pro_email=user_profile.email,
    )

get_wellbeing_assessment_feature_summary

get_wellbeing_assessment_feature_summary(*, account_id)

Source code in components/es/public/global_customer_dashboard/customer_health_partner.py

def get_wellbeing_assessment_feature_summary(  # noqa: D103
    *, account_id: str
) -> WellbeingAssessmentAvailability:
    account = get_or_raise_missing_resource(EsAccount, account_id)

    return get_wellbeing_assessment_feature_summary_logic(
        company_ids=[str(c.id) for c in account.companies],
        can_create_assessment=True,
    )

validate_company_external_id

validate_company_external_id(external_id)

Source code in components/es/public/global_customer_dashboard/customer_health_partner.py

def validate_company_external_id(external_id: str) -> bool:  # noqa: D103
    return is_cif(maybe_cif=external_id)

employee

cancel_employee_transfer

cancel_employee_transfer(transfer_id)

Deletes an employee transfer. :param transfer_id: the id of the employee transfer to cancel :raises BaseErrorCode.missing_resource: If the transfer does not exist

Source code in components/es/public/global_customer_dashboard/employee.py

def cancel_employee_transfer(transfer_id: UUID) -> None:
    """
    Deletes an employee transfer.
    :param transfer_id: the id of the employee transfer to cancel
    :raises BaseErrorCode.missing_resource: If the transfer does not exist
    """
    employee_transfer = EsEmployeeTransferModelBroker.get_or_raise(transfer_id)
    current_session.delete(employee_transfer)

get_employee_scheduled_transfer_or_none

get_employee_scheduled_transfer_or_none(
    user_id, company_id
)

Get the scheduled transfer for an employee.

:param user_id: The user id of the employee. :param company_id: The company id of the employee. :return: The scheduled transfer, or None if there is no scheduled transfer. :raises BaseErrorCode.invalid_arguments: If the scheduled transfer is not for the provided company as initial company

Source code in components/es/public/global_customer_dashboard/employee.py

def get_employee_scheduled_transfer_or_none(
    user_id: UUID, company_id: UUID
) -> ScheduledEmployeeTransfer | None:
    """
    Get the scheduled transfer for an employee.

    :param user_id: The user id of the employee.
    :param company_id: The company id of the employee.
    :return: The scheduled transfer, or None if there is no scheduled transfer.
    :raises BaseErrorCode.invalid_arguments: If the scheduled transfer is not for the provided company as initial company
    """
    from components.es.internal.business_logic.employment.queries.employee_transfer import (
        get_user_pending_employment_transfer,
    )

    # TODO: @pablo.martinez-perez should we also retrieve the primary user_id before getting the scheduled transfers?
    scheduled_employee_transfer = get_user_pending_employment_transfer(user_id)

    if not scheduled_employee_transfer:
        return None

    if scheduled_employee_transfer.initial_company_id != company_id:
        raise BaseErrorCode.invalid_arguments(
            message=f"The scheduled transfer of employee <{user_id}> is not for the provided initial company <{company_id}>"
        )

    return _to_global_dashboard_schedule_employee_transfer(scheduled_employee_transfer)

schedule_employee_transfer

schedule_employee_transfer(
    user_id,
    current_company_id,
    destination_company_id,
    at_date,
    *,
    should_cancel_initial_employment=False,
    should_execute_transfer_immediately_if_at_date_is_past=True,
    commit,
    dry_run
)

Schedule an employee transfer.

The transfer will be executed immediately in a queue, in case the at_date is prior to today, unless you specify differently using

setting the should_execute_transfer_immediately_if_at_date_is_past to False.

:param user_id: The user id of the employee to transfer.. :param current_company_id: The id of the company the employee is currently employed at. :param destination_company_id: The id of the company to transfer the employee to. :param at_date: The date when the transfer will occur. Should be in the future. :param should_cancel_initial_employment: Whether the initial employment should be cancelled. :param should_execute_transfer_immediately_if_at_date_is_past: Whether the transfer should be executed immediately if the at_date is in the past. :param commit: Whether the transfer should be committed. Useful only if at_date is in the past. :param dry_run: Whether the transfer should execute the external side effects. Useful only if at_date is in the past. :return: The created employee transfer.

Source code in components/es/public/global_customer_dashboard/employee.py

def schedule_employee_transfer(
    user_id: UUID,
    current_company_id: UUID,
    destination_company_id: UUID,
    at_date: date,
    *,
    should_cancel_initial_employment: bool = False,
    should_execute_transfer_immediately_if_at_date_is_past: bool = True,
    commit: bool,
    dry_run: bool,
) -> ScheduledEmployeeTransfer:
    """
    Schedule an employee transfer.

    Important: The transfer will be executed immediately in a queue, in case the at_date is prior to today, unless you specify differently using
     setting the should_execute_transfer_immediately_if_at_date_is_past to False.

    :param user_id: The user id of the employee to transfer..
    :param current_company_id: The id of the company the employee is currently employed at.
    :param destination_company_id: The id of the company to transfer the employee to.
    :param at_date: The date when the transfer will occur. Should be in the future.
    :param should_cancel_initial_employment: Whether the initial employment should be cancelled.
    :param should_execute_transfer_immediately_if_at_date_is_past: Whether the transfer should be executed immediately if the at_date is in the past.
    :param commit: Whether the transfer should be committed. Useful only if at_date is in the past.
    :param dry_run: Whether the transfer should execute the external side effects. Useful only if at_date is in the past.
    :return: The created employee transfer.
    """
    from components.es.internal.business_logic.employment.actions.employee_transfer.schedule_employee_transfer import (
        schedule_employee_transfer_to_another_company,
    )

    log = current_logger.bind(
        user_id=user_id,
        current_company_id=current_company_id,
        destination_company_id=destination_company_id,
        at_date=at_date,
        should_cancel_initial_employment=should_cancel_initial_employment,
        should_execute_transfer_immediately_if_at_date_is_past=should_execute_transfer_immediately_if_at_date_is_past,
    )

    user_employment = EsEmploymentModelBroker.get_user_active_employment(
        user_id=user_id, company_id=current_company_id, at_date=at_date
    )

    if user_employment is None:
        raise BaseErrorCode.invalid_arguments(
            message=f"User <{user_id} not employed at the current company {current_company_id} at date {at_date}"
        )

    with side_effects(trigger_on_exit=True):
        created_employee_transfer = schedule_employee_transfer_to_another_company(
            user_employment.id,
            destination_company_id,
            at_date,
            should_cancel_initial_employment=should_cancel_initial_employment,
        )
        # We want to commit the employee transfer as we'll send employee an email when transfer is scheduled
        current_session.commit()

    log = log.bind(employee_transfer_id=created_employee_transfer.id)

    # If transfer scheduled for today or in the past, execute it immediately
    if at_date <= date.today():
        if should_execute_transfer_immediately_if_at_date_is_past:
            from components.es.internal.business_logic.employment.actions.employee_transfer.execute_employee_transfer import (
                execute_employee_transfer,
            )

            log.info("Transfer date is prior to today, executing employee transfer now")
            execute_employee_transfer(
                created_employee_transfer.id, commit=commit, dry_run=dry_run
            )
            log.info(
                "Employee transfer has correctly been executed",
            )
        else:
            log.info(
                "Transfer date is prior to today, but transfer has NOT been executed now. "
                "Transfer will be executed in the next scheduled job"
            )

    return _to_global_dashboard_schedule_employee_transfer(created_employee_transfer)

update_employee_transfer_date

update_employee_transfer_date(
    transaction_id, new_transfer_date
)

Updates the transfer date of an employee transfer. :param transaction_id: the id of the employee transfer to update :param new_transfer_date: the new transfer date :raises BaseErrorCode.missing_resource: If the transfer does not exist

Source code in components/es/public/global_customer_dashboard/employee.py

def update_employee_transfer_date(
    transaction_id: UUID, new_transfer_date: date
) -> None:
    """
    Updates the transfer date of an employee transfer.
    :param transaction_id: the id of the employee transfer to update
    :param new_transfer_date: the new transfer date
    :raises BaseErrorCode.missing_resource: If the transfer does not exist
    """
    employee_transfer = EsEmployeeTransferModelBroker.get_or_raise(transaction_id)
    employee_transfer.transfer_date = new_transfer_date

    current_session.commit()

employee_enrollments

get_employee_enrollments

get_employee_enrollments(user_id, company_id, at_date=None)

Retrieve list of all active enrollments for an employee at a given date Raises an error if the user is not employed by the company at the given date

Source code in components/es/public/global_customer_dashboard/employee_enrollments.py

def get_employee_enrollments(
    user_id: UUID, company_id: UUID, at_date: date | None = None
) -> list[EmployeeServiceEnrollment]:
    """
    Retrieve list of all active enrollments for an employee at a given date
    Raises an error if the user is not employed by the company at the given date
    """
    from components.es.internal.business_logic.enrollment.queries.base_enrollment import (
        get_user_active_base_enrollments,
    )

    _at_date = at_date or date.today()

    employment = EsEmploymentModelBroker.get_user_active_employment(
        user_id=user_id, company_id=company_id, at_date=_at_date
    )
    if employment is None:
        raise BaseErrorCode.missing_resource(
            message="User is not employed by company at this date"
        )

    enrollments = get_user_active_base_enrollments(user_id)

    return [
        _to_global_dashboard_employee_service_enrollment(
            enrollment, user_id, company_id
        )
        for enrollment in enrollments
    ]

member

AdditionalParamsHealthInsuranceOnly

Bases: TypedDict

lang instance-attribute

lang

payroll_id instance-attribute

payroll_id

with_healthy_benefits instance-attribute

with_healthy_benefits

AdditionalParamsHealthyBenefits

Bases: TypedDict

annual_salary instance-attribute

annual_salary

hiring_date instance-attribute

hiring_date

lang instance-attribute

lang

payroll_id instance-attribute

payroll_id

population_id instance-attribute

population_id

tax_regime instance-attribute

tax_regime

with_healthy_benefits instance-attribute

with_healthy_benefits

CSVHeaderDefinition dataclass

CSVHeaderDefinition(
    field_name, order, header_by_lang, required=True
)

Definition of CSV headers for bulk invitation templates.

field_name instance-attribute

field_name

header_by_lang instance-attribute

header_by_lang

order instance-attribute

order

required class-attribute instance-attribute

required = True

CSV_HEADERS_DEFINITIONS module-attribute

CSV_HEADERS_DEFINITIONS = [
    CSVHeaderDefinition(
        field_name="email",
        order=1,
        header_by_lang={
            "es": "Correo electrónico del empleado (mismo correo que para el seguro)",
            "en": "Employee email (Match email used for insurance)",
        },
    ),
    CSVHeaderDefinition(
        field_name="annual_salary",
        order=2,
        header_by_lang={
            "es": "Salario bruto anual (redondeado al entero más cercano, ej: 41852)",
            "en": "Gross annual salary (Rounded to the nearest integer, ex: 41852)",
        },
    ),
    CSVHeaderDefinition(
        field_name="hiring_date",
        order=3,
        header_by_lang={
            "es": "Fecha de contratación (AAAA-MM-DD)",
            "en": "Hiring date (YYYY-MM-DD)",
        },
    ),
    CSVHeaderDefinition(
        field_name="tax_regime",
        order=4,
        header_by_lang={
            "es": "Régimen fiscal (opciones disponibles: General / Navarra / Vizcaya y Guipuzcoa / Alava)",
            "en": "Tax regime (Available choices: General / Navarre / Biscay and Gipuzkoa / Alava)",
        },
    ),
    CSVHeaderDefinition(
        field_name="lang",
        order=5,
        header_by_lang={
            "es": "Idioma (es / en)",
            "en": "Language (en / es)",
        },
    ),
    CSVHeaderDefinition(
        field_name="payroll_id",
        order=6,
        header_by_lang={
            "es": "ID de empleado (opcional)",
            "en": "Employee ID (optional)",
        },
        required=False,
    ),
]

EditEmployeeInvitationParams

Bases: TypedDict

Parameters for editing an existing employee invitation.

lang instance-attribute

lang

payroll_id instance-attribute

payroll_id

EsEmployeeDetails dataclass

EsEmployeeDetails(
    first_name,
    last_name,
    email,
    enrollments,
    offboarding_email_sent,
    hiring_date=isodate_field(),
    offboarding_email_send_date=optional_isodate_field(),
    employment_end_date=optional_isodate_field(),
    insurance_period=None,
    gender=None,
    annual_salary_label=None,
    tax_regime=None,
    payroll_id=None,
)

Bases: DataClassJsonMixin

annual_salary_label class-attribute instance-attribute

annual_salary_label = None

email instance-attribute

email

employment_end_date class-attribute instance-attribute

employment_end_date = optional_isodate_field()

enrollments instance-attribute

enrollments

first_name instance-attribute

first_name

gender class-attribute instance-attribute

gender = None

hiring_date class-attribute instance-attribute

hiring_date = isodate_field()

insurance_period class-attribute instance-attribute

insurance_period = None

last_name instance-attribute

last_name

offboarding_email_send_date class-attribute instance-attribute

offboarding_email_send_date = optional_isodate_field()

offboarding_email_sent instance-attribute

offboarding_email_sent

payroll_id class-attribute instance-attribute

payroll_id = None

tax_regime class-attribute instance-attribute

tax_regime = None

HEADERS_BY_LANG module-attribute

HEADERS_BY_LANG = {
    "es": [
        "Correo electrónico del empleado (mismo correo que para el seguro)",
        "Salario bruto anual (redondeado al entero más cercano, ej: 41852)",
        "Fecha de contratación (AAAA-MM-DD)",
        "Régimen fiscal (opciones disponibles: General / Navarra / Vizcaya y Guipuzcoa / Alava)",
        "Idioma (es / en)",
    ],
    "en": [
        "Employee email (Match email used for insurance)",
        "Gross annual salary (Rounded to the nearest integer, ex: 41852)",
        "Hiring date (YYYY-MM-DD)",
        "Tax regime (Available choices: General / Navarre / Biscay and Gipuzkoa / Alava)",
        "Language (en / es)",
    ],
}

LINEAR_SPAIN_ENG_ON_CALL_LABEL module-attribute

LINEAR_SPAIN_ENG_ON_CALL_LABEL = (
    "b0c45d28-20bf-4304-bb59-793707922e43"
)

LINEAR_SPAIN_TEAM_ID module-attribute

LINEAR_SPAIN_TEAM_ID = (
    "94bcb412-8c0e-4e3d-98ec-c0ede8b4212a"
)

MemberInvitationParams dataclass

MemberInvitationParams(invite_email, additional_params)

Parameters for inviting a new member to the company.

additional_params instance-attribute

additional_params

invite_email instance-attribute

invite_email

PAYROLL_HEADER_BY_LANG module-attribute

PAYROLL_HEADER_BY_LANG = {
    "es": "ID de empleado (opcional)",
    "en": "Employee ID (optional)",
}

TEMPLATE_NAME_BY_LANG module-attribute

TEMPLATE_NAME_BY_LANG = {
    "es": "invitacion_beneficios_modelo.csv",
    "en": "invitation_benefits_template.csv",
}

UpdateEmployeeDetailsParams2

Bases: TypedDict

Parameters for updating employee details.

annual_salary instance-attribute

annual_salary

tax_regime instance-attribute

tax_regime

anonymize_salary

anonymize_salary(salary)

Anonymize the salary of an employee.

Returns: - str: Returns "..." if the salary is anonymized. - None: Returns None if the salary input is None.

Source code in components/es/public/global_customer_dashboard/member.py

def anonymize_salary(salary: int | None) -> str | None:
    """
    Anonymize the salary of an employee.

    Returns:
    - str: Returns "..." if the salary is anonymized.
    - None: Returns None if the salary input is None.
    """
    if salary is None:
        return None

    return "..."

cancel_employee_invitation

cancel_employee_invitation(company_id, invitation_id)

Cancel a pending employee invitation by ID.

Attempts to cancel the invitation through two possible sources: 1. Employee onboarding system (EsEmployeeOnboardingModelBroker) 2. Employment component blocked movements (from ES admin dashboard)

Parameters:

Name	Type	Description	Default
company_id	UUID	UUID of the company the invitation belongs to	required
invitation_id	UUID	UUID of the invitation to cancel	required

Raises:

Type	Description
missing_resource	If no invitation found with the given ID

Returns:

Type	Description
None	None

Source code in components/es/public/global_customer_dashboard/member.py

def cancel_employee_invitation(
    company_id: uuid.UUID,
    invitation_id: uuid.UUID,
) -> None:
    """
    Cancel a pending employee invitation by ID.

    Attempts to cancel the invitation through two possible sources:
    1. Employee onboarding system (EsEmployeeOnboardingModelBroker)
    2. Employment component blocked movements (from ES admin dashboard)

    Args:
        company_id: UUID of the company the invitation belongs to
        invitation_id: UUID of the invitation to cancel

    Raises:
        BaseErrorCode.missing_resource: If no invitation found with the given ID

    Returns:
        None
    """
    employee_onboarding = EsEmployeeOnboardingModelBroker.get(id=invitation_id)
    if employee_onboarding:
        from components.es.internal.business_logic.employee_onboarding.employee_onboarding import (
            employee_onboarding_logic,
        )

        employee_onboarding_logic.cancel_pending_invite(
            employee_onboarding_id=employee_onboarding.id
        )
        return

    from components.employment.public.business_logic.actions.blocked_movement import (
        cancel_pending_blocked_movement,
    )
    from components.employment.public.business_logic.queries.blocked_movement import (
        get_pending_blocked_movements_for_company_id,
    )

    company_blocked_movements = get_pending_blocked_movements_for_company_id(
        company_id=str(company_id), from_sources={SourceType.es_admin_dashboard}
    )
    blocked_movement_for_invitation = next(
        (bm for bm in company_blocked_movements if str(bm.id) == str(invitation_id)),
        None,
    )

    if blocked_movement_for_invitation:
        cancel_pending_blocked_movement(
            blocked_movement_id=blocked_movement_for_invitation.id,
            commit=True,
        )
        return

    raise BaseErrorCode.missing_resource(
        message=f"Invitation {invitation_id} is not found"
    )

check_employee_bulk_invitation_csv

check_employee_bulk_invitation_csv(file, additional_params)

Check the validity of the employee bulk invitation CSV file.

Source code in components/es/public/global_customer_dashboard/member.py

def check_employee_bulk_invitation_csv(
    file: FileStorage | None,
    additional_params: dict,  # type: ignore[type-arg]
) -> list[EmployeeInvitationResult[object]]:
    """
    Check the validity of the employee bulk invitation CSV file.

    """
    if not file:
        raise BaseErrorCode.invalid_arguments(
            message="Missing invitation file for healthy benefits bulk invite"
        )
    with_healthy_benefits = additional_params.pop("with_healthy_benefits", False)

    if not with_healthy_benefits:
        raise NotImplementedError(
            "Bulk invitation is only available for healthy benefits"
        )

    if not file:
        raise BaseErrorCode.invalid_arguments(
            message="Missing invitation file for healthy benefits bulk invite"
        )

    results = check_employee_bulk_invitation_csv_logic(
        csv_file=io.BytesIO(file.stream.read()),
        fieldnames=get_csv_field_names_by_order(),
    )

    mapped_results = [
        map_es_employee_invitation_result_to_employee_invitation_result(result)
        for result in results
    ]

    return mapped_results

create_bulk_members_invitations

create_bulk_members_invitations(
    company_id, start_date, additional_params, file=None
)

Source code in components/es/public/global_customer_dashboard/member.py

def create_bulk_members_invitations(  # noqa: D103
    company_id: str,
    start_date: date,
    additional_params: dict,  # type: ignore[type-arg]
    file: FileStorage | None = None,
) -> None:
    with_healthy_benefits = additional_params.pop("with_healthy_benefits", False)

    if not with_healthy_benefits:
        raise NotImplementedError(
            "Bulk invitation is only available for healthy benefits"
        )

    if not file:
        raise BaseErrorCode.invalid_arguments(
            message="Missing invitation file for healthy benefits bulk invite"
        )

    company = get_or_raise_missing_resource(EsCompany, uuid.UUID(company_id))
    front_url = current_config["FRONT_END_BASE_URL"]

    company_marmot_url = f"{front_url}/marmot/company/{company.id}"

    # Upload the file to S3 into manual_uploads folder
    s3_file_uri = RemoteFileClient.upload(
        file_path_or_file_obj=file,  # type: ignore[arg-type]
        upload_dir="manual_upload",
    )

    # Create a Linear card with the file url
    linear_client = LinearClient()
    jinja_template = Template(
        """
A new bulk invitation CSV for healthy benefits has been uploaded for [{{company_name}}]({{company_marmot_url}}).

Redirect this issue to the relevant OPS team member.

You can find the file [here]({{file_url}}).
The start date is {{start_date}}.

Head to the [company's marmot page]({{company_marmot_url}}), on the Healthy Benefits tab, to upload the file and invite the employees.
"""
    )
    description = jinja_template.render(
        company_id=company_id,
        file_url=s3_file_uri,
        start_date=start_date,
        company_marmot_url=company_marmot_url,
        company_name=company.display_name,
    )
    linear_client.create_issue(
        title=f"New healthy benefits bulk invitation for {company.display_name}",
        description=description,
        team_id=LINEAR_SPAIN_TEAM_ID,
        label_ids=[LINEAR_SPAIN_ENG_ON_CALL_LABEL],
    )

create_bulk_members_invitations_json

create_bulk_members_invitations_json(
    company_id, start_date, additional_params, employees
)

Source code in components/es/public/global_customer_dashboard/member.py

def create_bulk_members_invitations_json(  # noqa: D103
    company_id: str,
    start_date: date,
    additional_params: dict,  # type: ignore[type-arg]
    employees: list[dict],  # type: ignore[type-arg]
) -> list[EmployeeInvitationResult[object]]:
    with_healthy_benefits = additional_params.pop("with_healthy_benefits", False)

    if not with_healthy_benefits:
        raise NotImplementedError(
            "Bulk invitation is only available for healthy benefits"
        )

    results = bulk_invite_employees_to_healthy_benefits_from_json(
        dry_run=False,
        company_id=uuid.UUID(company_id),
        employees=employees,
        insurance_start_date=start_date,
        healthy_benefits_start_date=start_date,
        invite_to_healthy_benefits_if_already_enrolled=True,
    )
    mapped_results = [
        map_es_employee_invitation_result_to_employee_invitation_result(result)
        for result in results
    ]
    return mapped_results

create_member_invitation

create_member_invitation(
    company_id, invite_email, start_date, additional_params
)

Create a new member invitation in the company. - Creates EmployeeOnboarding in ES - Sends an email to the employee with the invitation link

Parameters:

Name	Type	Description	Default
company_id	str	The ID of the company.	required
invite_email	str	The email address of the employee to invite.	required
start_date	date	The start date for the employee's coverage.	required
additional_params	AdditionalParamsHealthyBenefits | AdditionalParamsHealthInsuranceOnly	Additional parameters for the invitation.	required

Returns:

Name	Type	Description
tuple	CustomerDashboardEmployeesInvitationResponses	A tuple containing: - CustomerDashboardEmployeeInvitation: The invitation details if successful. - CustomerDashboardEmployeeInvitationError: The error details if the invitation failed.

Source code in components/es/public/global_customer_dashboard/member.py

def create_member_invitation(
    company_id: str,
    invite_email: str,
    start_date: date,
    additional_params: AdditionalParamsHealthyBenefits
    | AdditionalParamsHealthInsuranceOnly,
) -> CustomerDashboardEmployeesInvitationResponses:
    """
    Create a new member invitation in the company.
    - Creates EmployeeOnboarding in ES
    - Sends an email to the employee with the invitation link

    Args:
        company_id (str): The ID of the company.
        invite_email (str): The email address of the employee to invite.
        start_date (date): The start date for the employee's coverage.
        additional_params (AdditionalParamsHealthyBenefits | AdditionalParamsHealthInsuranceOnly): Additional parameters for the invitation.

    Returns:
        tuple: A tuple containing:
            - CustomerDashboardEmployeeInvitation: The invitation details if successful.
            - CustomerDashboardEmployeeInvitationError: The error details if the invitation failed.
    """
    return create_member_invitations(
        company_id=company_id,
        start_date=start_date,
        invitations=[
            MemberInvitationParams(
                invite_email=invite_email,
                additional_params=additional_params,
            )
        ],
    )

create_member_invitations

create_member_invitations(
    company_id, start_date, invitations
)

Create new member invitations in the company. For each invitation: - Creates EmployeeOnboarding in ES - Sends an email to the employee with the invitation link

Returns a tuple of two lists: - A list of CustomerDashboardEmployeeInvitation objects for successfully invited employees - A list of CustomerDashboardEmployeeInvitationError objects for failed invitations

Source code in components/es/public/global_customer_dashboard/member.py

def create_member_invitations(
    company_id: str, start_date: date, invitations: list[MemberInvitationParams]
) -> CustomerDashboardEmployeesInvitationResponses:
    """
    Create new member invitations in the company. For each invitation:
    - Creates EmployeeOnboarding in ES
    - Sends an email to the employee with the invitation link

    Returns a tuple of two lists:
    - A list of CustomerDashboardEmployeeInvitation objects for successfully invited employees
    - A list of CustomerDashboardEmployeeInvitationError objects for failed invitations
    """
    log = current_logger.bind(
        company_id=company_id,
        start_date=start_date,
    )

    invited_employees = []
    invitation_errors = []

    for invitation in invitations:
        additional_params = invitation.additional_params
        invite_email = invitation.invite_email
        try:
            with_healthy_benefits = additional_params.get(
                "with_healthy_benefits", False
            )

            annual_salary = None
            population_id = None
            tax_regime = None
            hiring_date = None
            if with_healthy_benefits:
                additional_params_hb = cast(
                    "AdditionalParamsHealthyBenefits", additional_params
                )
                annual_salary = additional_params_hb.get("annual_salary", None)
                population_id_str = additional_params_hb.get("population_id", None)
                population_id = (
                    uuid.UUID(population_id_str) if population_id_str else None
                )
                tax_regime_str = additional_params_hb.get("tax_regime", None)
                tax_regime = TaxRegime(tax_regime_str) if tax_regime_str else None

                hiring_date_str = additional_params_hb.get("hiring_date")
                hiring_date = (
                    date.fromisoformat(hiring_date_str) if hiring_date_str else None
                )

            invitation_result = invite_employee_by_email_in_company(
                company_id=uuid.UUID(company_id),
                invite_email=invite_email,
                coverage_start_date=start_date,
                healthy_benefits_start_date=start_date,
                lang=additional_params.get("lang", "es"),
                annual_salary=annual_salary,
                external_employee_id=additional_params.get("payroll_id"),
                population_id=population_id,
                tax_regime=tax_regime,
                hiring_date=hiring_date or start_date,
                source_type=SourceType.es_admin_dashboard,
                send_email=True,
            )

            if (
                invitation_result
                and invitation_result.status == EsInvitationStatus.SUCCESS
            ):
                employee_onboarding = EsEmployeeOnboardingModelBroker.get_by_email(
                    email=invite_email, company_id=uuid.UUID(company_id)
                )
                if not employee_onboarding:
                    log.error(
                        "EmployeeOnboarding not found after successful invitation. This is probably a bug in the Employment consumer",
                    )
                    raise BaseErrorCode.model_validation(
                        message="EmployeeOnboarding should be created for invited employee. "
                    )

                invited_employees.append(
                    CustomerDashboardEmployeeInvitation(
                        id=str(employee_onboarding.id),
                        invite_email=invite_email,
                        invitation_date=mandatory(employee_onboarding.created_at),
                        invitation_email_sent_at=employee_onboarding.invitation_email_sent_at,
                    )
                )

            else:
                invitation_errors.append(
                    CustomerDashboardEmployeeInvitationError(
                        invite_email=invite_email,
                        error_type=CustomerDashboardEmployeeInvitationErrorType.employment_blocked_movement,
                    )
                )

        except (BaseErrorCode, Exception) as e:
            log.exception(
                "Error inviting employee in company",
                exc_info=e,
            )
            if (
                isinstance(e, BaseErrorCode)
                and e.alancode
                == BaseErrorCode.user_with_email_already_exists().alancode
            ):
                invitation_errors.append(
                    CustomerDashboardEmployeeInvitationError(
                        invite_email=invite_email,
                        error_type=CustomerDashboardEmployeeInvitationErrorType.already_exists,
                    )
                )
            else:
                invitation_errors.append(
                    CustomerDashboardEmployeeInvitationError(
                        invite_email=invite_email,
                        error_type=CustomerDashboardEmployeeInvitationErrorType.unknown_error,
                    )
                )

    log.info(
        "Employees invited",
        invited_count=len(invited_employees),
        errors_count=len(invitation_errors),
    )
    try:
        log.info("Committing transaction after inviting employees")
        current_session.commit()
    except Exception as e:
        log.exception(
            "Error committing transaction after inviting employees",
            exc_info=e,
        )
        raise BaseErrorCode.model_validation(
            message="Error committing transaction after inviting employees. Please try again later."
        )

    result = CustomerDashboardEmployeesInvitationResponses(
        invitations=invited_employees,
        errors=invitation_errors,
    )
    return result

edit_employee_invitation

edit_employee_invitation(
    invitation_id, company_id, additional_params
)

Edit an employee invitation by ID.

Attempts to edit the invitation through the Employee onboarding system (EsEmployeeOnboardingModelBroker)

Parameters:

Name	Type	Description	Default
invitation_id	UUID	UUID of the invitation to edit	required
company_id	UUID	UUID of the company the invitation belongs to	required
additional_params	EditEmployeeInvitationParams	dict containing the fields to update	required

Raises:

Type	Description
missing_resource	If no invitation found with the given ID

Returns:

Type	Description
None	None

Source code in components/es/public/global_customer_dashboard/member.py

def edit_employee_invitation(
    invitation_id: uuid.UUID,
    company_id: uuid.UUID,
    additional_params: EditEmployeeInvitationParams,
) -> None:
    """
    Edit an employee invitation by ID.

    Attempts to edit the invitation through the Employee onboarding system (EsEmployeeOnboardingModelBroker)

    Args:
        invitation_id: UUID of the invitation to edit
        company_id: UUID of the company the invitation belongs to
        additional_params: dict containing the fields to update

    Raises:
        BaseErrorCode.missing_resource: If no invitation found with the given ID

    Returns:
        None
    """
    employee_onboarding = EsEmployeeOnboardingModelBroker.get(id=invitation_id)
    payroll_id = additional_params.get("payroll_id")
    if employee_onboarding:
        employee_onboarding.payroll_id = payroll_id
        if employee_onboarding.user_id:
            if is_employment_component_integration_enabled(company_id=company_id):
                from components.es.internal.business_logic.employment.global_employment.upstreams.employee_invitation import (
                    edit_employment_invitation as edit_employment_invitation_with_employment_component,
                )

                edit_employment_invitation_with_employment_component(
                    user_id=employee_onboarding.user_id,
                    company_id=company_id,
                    invite_email=employee_onboarding.invite_email,
                    coverage_start_date=mandatory(
                        employee_onboarding.coverage_start_date
                    ),
                    lang=employee_onboarding.lang,
                    payroll_id=payroll_id,
                )
            else:
                employment = EsEmploymentModelBroker.get_user_employment_in_company(
                    user_id=employee_onboarding.user_id, company_id=company_id
                )
                if employment:
                    employment.payroll_id = payroll_id

        current_session.commit()
        return

    from components.employment.public.business_logic.actions.blocked_movement import (
        retry_core_blocked_movement,
        retry_upstream_blocked_movement,
    )
    from components.employment.public.business_logic.queries.blocked_movement import (
        get_pending_blocked_movements_for_company_id,
    )

    company_blocked_movements = get_pending_blocked_movements_for_company_id(
        company_id=str(company_id), from_sources={SourceType.es_admin_dashboard}
    )
    blocked_movement_for_invitation = next(
        (bm for bm in company_blocked_movements if str(bm.id) == str(invitation_id)),
        None,
    )
    if blocked_movement_for_invitation:
        if isinstance(blocked_movement_for_invitation, CoreBlockedMovement):
            original_employment_declaration: EsEmploymentDeclaration = (
                EmploymentDeclaration.from_dict(
                    mandatory(blocked_movement_for_invitation.context).get(
                        "employment_declaration"
                    )
                )
            )
            retry_core_blocked_movement(
                blocked_movement_id=blocked_movement_for_invitation.id,
                employment_declaration_override=replace(
                    original_employment_declaration,
                    external_employee_id=additional_params.get("payroll_id", None),
                ),
                commit=True,
            )
        else:
            retry_upstream_blocked_movement(
                upstream_blocked_movement_id=blocked_movement_for_invitation.id,
                upstream_data_override={
                    "external_employee_id": additional_params.get("payroll_id", None)
                },
                commit=True,
            )
        return

    raise BaseErrorCode.missing_resource(
        message=f"Invitation {invitation_id} is not found"
    )

generate_bulk_invitation_template

generate_bulk_invitation_template(company_id, lang)

Source code in components/es/public/global_customer_dashboard/member.py

def generate_bulk_invitation_template(  # noqa: D103
    company_id: str,
    lang: str,
) -> tuple[io.BytesIO, str]:
    if lang not in TEMPLATE_NAME_BY_LANG:
        supported_langs = list(TEMPLATE_NAME_BY_LANG.keys())
        raise BaseErrorCode.invalid_arguments(
            message=f"Invalid language: {lang}. Supported languages: {supported_langs}"
        )

    company = get_or_raise_missing_resource(EsCompany, company_id)
    headers = get_csv_headers_by_lang(lang, company.use_employee_payroll_id)

    csv_row = dict.fromkeys(headers, "")
    temp_file = temp_csv_file_from_rows(
        csv_rows=[csv_row],
        fieldnames=headers,
        encoding="utf-8",
        delimiter=",",
        write_header=True,
    )
    return io.BytesIO(temp_file.read()), TEMPLATE_NAME_BY_LANG[lang]

get_csv_field_names_by_order

get_csv_field_names_by_order()

Get all field names from CSV header definitions sorted by their order.

Source code in components/es/public/global_customer_dashboard/member.py

def get_csv_field_names_by_order() -> list[str]:
    """
    Get all field names from CSV header definitions sorted by their order.

    """
    return [
        header.field_name
        for header in sorted(CSV_HEADERS_DEFINITIONS, key=lambda h: h.order)
    ]

get_csv_headers_by_lang

get_csv_headers_by_lang(lang, company_use_payroll_id=False)

Source code in components/es/public/global_customer_dashboard/member.py

def get_csv_headers_by_lang(  # noqa: D103
    lang: str, company_use_payroll_id: bool = False
) -> list[str]:
    headers = [
        header.header_by_lang[lang]
        for header in sorted(CSV_HEADERS_DEFINITIONS, key=lambda h: h.order)
        if lang in header.header_by_lang and (header.required or company_use_payroll_id)
    ]

    return headers

get_employee_details

get_employee_details(user_id, company_id)

Retrieve comprehensive employee details for a user within a Spanish company.

Fetches and aggregates employee information including personal details, employment status, enrollments, insurance periods, and healthy benefits profiles. This is used in the global customer dashboard to display member information.

Parameters:

Name	Type	Description	Default
user_id	UUID	The unique identifier of the user/employee.	required
company_id	UUID	The unique identifier of the company.	required

Returns:

Type	Description
EmployeeDetails[EsEmployeeDetails]	EmployeeDetails containing the user's personal information, employment details,
EmployeeDetails[EsEmployeeDetails]	enrollments, salary information, and insurance period.

Raises:

Type	Description
missing_resource	If no employment is found for the user in the specified company.

Source code in components/es/public/global_customer_dashboard/member.py

def get_employee_details(
    user_id: uuid.UUID, company_id: uuid.UUID
) -> EmployeeDetails[EsEmployeeDetails]:
    """Retrieve comprehensive employee details for a user within a Spanish company.

    Fetches and aggregates employee information including personal details, employment status,
    enrollments, insurance periods, and healthy benefits profiles. This is used in the global
    customer dashboard to display member information.

    Args:
        user_id: The unique identifier of the user/employee.
        company_id: The unique identifier of the company.

    Returns:
        EmployeeDetails containing the user's personal information, employment details,
        enrollments, salary information, and insurance period.

    Raises:
        BaseErrorCode.missing_resource: If no employment is found for the user in the
            specified company.
    """
    from components.es.internal.business_logic.company.CompanyPayrollPeriod import (
        CompanyPayrollPeriod,
    )
    from components.es.internal.business_logic.enrollment.queries.base_enrollment import (
        get_user_active_base_enrollments,
    )
    from components.es.internal.business_logic.enrollment.queries.enrollment import (
        get_latest_continuous_insurance_period,
    )
    from components.es.internal.business_logic.user_v2.service import user_service
    from components.es.subcomponents.healthy_benefits.protected.business_logic.healthy_benefits_profile import (
        get_healthy_benefits_profile,
    )

    user_profile = user_service.get_or_raise_user_profile(user_id)
    base_enrollments = get_user_active_base_enrollments(user_id, at_date=date.today())
    employee_employment = EsEmploymentModelBroker.get_last_user_employment(
        user_id=user_id, company_id=company_id
    )

    active_or_future_insurance_enrollment = (
        EsEnrollmentModelBroker.get_user_active_or_future_enrollment(
            user_id=user_id, at_date=date.today()
        )
    )

    insurance_period = None
    if active_or_future_insurance_enrollment is not None:
        latest_continuous_insurance_period = get_latest_continuous_insurance_period(
            enrollment_id=active_or_future_insurance_enrollment.id
        )

        if latest_continuous_insurance_period is not None:
            insurance_period = ValidityPeriod(
                start_date=latest_continuous_insurance_period.start_date,
                end_date=latest_continuous_insurance_period.end_date,
            )

    if employee_employment is None:
        raise BaseErrorCode.missing_resource(
            message=f"No employment found for user {user_id} in company {company_id}"
        )

    company_payroll_period = CompanyPayrollPeriod.load_from_company(
        company_id=company_id, at_date=date.today()
    )
    next_payroll_start_date = mandatory(
        company_payroll_period.current_period.day_after_end
    )
    current_healthy_benefits_profile: Optional[HealthyBenefitsProfile] = None
    next_healthy_benefits_profile: Optional[HealthyBenefitsProfile] = None

    try:
        # Get the healthy benefits profile for the next payroll start date since we update it at the next payroll start date
        current_healthy_benefits_profile = get_healthy_benefits_profile(
            user_id=user_id,
            at_date=date.today(),
            include_future=True,
        )
        next_healthy_benefits_profile = get_healthy_benefits_profile(
            user_id=user_id,
            at_date=next_payroll_start_date,
            include_future=True,
        )
    except BaseErrorCode:
        # If the future profile is None, we check if there is a current one to return it
        if current_healthy_benefits_profile is not None:
            next_healthy_benefits_profile = current_healthy_benefits_profile

    return EmployeeDetails(
        user_id=str(user_id),
        details=EsEmployeeDetails(
            first_name=user_profile.first_name,  # type: ignore[arg-type]
            last_name=user_profile.last_name,  # type: ignore[arg-type]
            email=employee_employment.invite_email or "",
            gender=user_profile.gender_compat,
            enrollments=base_enrollments,
            hiring_date=employee_employment.start_date,
            offboarding_email_send_date=employee_employment.offboarding_email_send_date,
            offboarding_email_sent=employee_employment.offboarding_email_sent,
            payroll_id=employee_employment.payroll_id,
            annual_salary_label=anonymize_salary(
                salary=next_healthy_benefits_profile.annual_salary
            )
            if next_healthy_benefits_profile
            else None,
            tax_regime=(
                next_healthy_benefits_profile.tax_regime
                if next_healthy_benefits_profile
                else None
            ),
            employment_end_date=employee_employment.end_date,
            insurance_period=insurance_period,
        ),
    )

map_es_employee_invitation_result_to_employee_invitation_result

map_es_employee_invitation_result_to_employee_invitation_result(
    es_result,
)

Source code in components/es/public/global_customer_dashboard/member.py

def map_es_employee_invitation_result_to_employee_invitation_result(  # noqa: D103
    es_result: EsEmployeeInvitationResult,
) -> EmployeeInvitationResult[
    HealthyBenefitsInvitationParams
    | HealthyBenefitsNewInvitationParams
    | HealthyBenefitsNewInvitationParamsWithPayroll
    | object
]:
    return EmployeeInvitationResult(
        id=es_result.id,
        email=es_result.email,
        employee=es_result.employee,
        success=es_result.success,
        error=es_result.error,
        error_keys=es_result.error_keys,
        meta=es_result.meta,
    )

update_employee_details

update_employee_details(
    user_id, company_id, details, *, commit=True
)

Update the details of an employee in a Spanish company.

This function handles updating employee information tax regime and annual salary.

Parameters:

Name	Type	Description	Default
user_id	UUID	UUID of the employee to update	required
company_id	UUID	UUID of the Spanish company where the employee works	required
details	UpdateEmployeeDetailsParams2	Dictionary containing the employee details to update: - tax_regime: Tax regime to apply to the employee - annual_salary: New annual salary (optional, will use existing if None)	required
commit	bool	Whether to commit the database transaction (default: True)	True

Returns:

Type	Description
EmployeeDetails[EsEmployeeDetails]	Updated EmployeeDetails object containing the employee's current information

Raises:

Type	Description
missing_resource	If no employment is found for the user in the company
invalid_arguments	If annual salary is not provided, and cannot be inferred from existing profile

Note

• Annual salary defaults to current profile value if not provided
• Updates are applied to the next payroll period start date

Source code in components/es/public/global_customer_dashboard/member.py

def update_employee_details(
    user_id: uuid.UUID,
    company_id: uuid.UUID,
    details: UpdateEmployeeDetailsParams2,
    *,
    commit: bool = True,
) -> EmployeeDetails[EsEmployeeDetails]:
    """
    Update the details of an employee in a Spanish company.

    This function handles updating employee information tax regime and annual salary.

    Args:
        user_id: UUID of the employee to update
        company_id: UUID of the Spanish company where the employee works
        details: Dictionary containing the employee details to update:
            - tax_regime: Tax regime to apply to the employee
            - annual_salary: New annual salary (optional, will use existing if None)
        commit: Whether to commit the database transaction (default: True)

    Returns:
        Updated EmployeeDetails object containing the employee's current information

    Raises:
        BaseErrorCode.missing_resource: If no employment is found for the user in the company
        BaseErrorCode.invalid_arguments: If annual salary is not provided, and cannot be inferred from existing profile

    Note:
        - Annual salary defaults to current profile value if not provided
        - Updates are applied to the next payroll period start date
    """
    from components.es.subcomponents.healthy_benefits.internal.business_logic.healthy_benefits_profile.healthy_benefits_profile_queries import (
        get_healthy_benefits_profile_or_none,
    )

    log = current_logger.bind(
        user_id=user_id,
        company_id=company_id,
    )
    log.debug("Updating employee details")

    user_current_profile = get_healthy_benefits_profile_or_none(
        user_id=user_id, at_date=date.today()
    )

    annual_salary = details.get("annual_salary")
    if annual_salary is None and user_current_profile is not None:
        annual_salary = user_current_profile.annual_salary

    if annual_salary is None:
        # We should never be in this case as this logic is called to always update an existing profile.
        # We should improve it in the future so we can update the profile without annual salary
        # as it is not always required in case user is not covered with Healthy benefits
        raise BaseErrorCode.invalid_arguments(
            message="We can't update the user profile without an Annual salary value."
        )

    from components.es.internal.business_logic.company.CompanyPayrollPeriod import (
        CompanyPayrollPeriod,
    )

    with side_effects():
        employee_employment = EsEmploymentModelBroker.get_user_employment_in_company(
            user_id=user_id,
            company_id=company_id,
            join_load_insurance_profile=True,
        )
        if employee_employment is None:
            raise BaseErrorCode.missing_resource(
                message=f"No employment found for user {user_id} in company {company_id}"
            )
        company_payroll_period = CompanyPayrollPeriod.load_from_company(
            company_id, date.today()
        )

        next_payroll_start_date = company_payroll_period.next().first_day

        insurance_profile = mandatory(employee_employment).user.insurance_profile

        upsert_health_benefits_profile(
            user_id=user_id,
            company_id=company_id,
            annual_salary=annual_salary,
            tax_regime=details["tax_regime"],
            at_date=next_payroll_start_date,
            nif_number=insurance_profile.nif_number if insurance_profile else None,
            identity_document_type=insurance_profile.identity_document_type
            if insurance_profile
            else None,
            commit=False,
        )

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

    return get_employee_details(user_id, company_id)

components.es.public.helpers

flask_admin_link

extend_flask_admin_config

extend_flask_admin_config(config)

Extend the Flask Admin configuration with custom formatters for cross-component links.

This is useful to improve navigation between component models that don't have direct dependencies, for example those using opaque reference strings instead of foreign keys or typed IDs. Those kinds of references are typically app-dependent.

Source code in components/es/public/helpers/flask_admin_link.py

def extend_flask_admin_config(config: AlanAdminConfiguration) -> None:
    """Extend the Flask Admin configuration with custom formatters for
    cross-component links.

    This is useful to improve navigation between component models that don't
    have direct dependencies, for example those using opaque reference strings
    instead of foreign keys or typed IDs. Those kinds of references are
    typically app-dependent.
    """
    for model, spec in config.mounted_models_specs.items():
        match model.__name__:
            # Payment Gateway
            case "ExpenseLimitRule":
                spec.update(
                    column_formatters=dict(
                        **spec.get("column_formatters", {}),
                        reference=foreign_reference_formatter(
                            model="healthybenefitsenrollment", prefix="healthybenefits"
                        ),
                    )
                )
            # Global Healthy Benefits
            case "Discount":
                spec.update(
                    column_formatters=dict(
                        **spec.get("column_formatters", {}),
                        subscription_id=foreign_reference_formatter(
                            model="subscriptionmodel"
                        ),
                    )
                )
            case "HealthyBenefitsEnrollment":
                spec.update(
                    column_formatters=dict(
                        **spec.get("column_formatters", {}),
                        profile_ref=foreign_reference_formatter(
                            model="eshealthybenefitsprofile"
                        ),
                        subscription_ref=foreign_reference_formatter(
                            model="subscriptionmodel"
                        ),
                    )
                )
            case "Population":
                spec.update(
                    column_formatters=dict(
                        **spec.get("column_formatters", {}),
                        entity_ref=foreign_reference_formatter(model="escompany"),
                    )
                )
            case "EsCompany":
                spec.update(
                    column_formatters=dict(
                        **spec.get("column_formatters", {}),
                        payment_account_id=foreign_reference_formatter(
                            model="account", prefix="paymentgateway"
                        ),
                    )
                )

format_entity_name_for_link

format_entity_name_for_link(entity_name)

Source code in components/es/public/helpers/flask_admin_link.py

def format_entity_name_for_link(entity_name: str) -> str:  # noqa: D103
    return entity_name.replace("_", "")

link_to_company

link_to_company(company_id)

Source code in components/es/public/helpers/flask_admin_link.py

def link_to_company(company_id: int) -> str:  # noqa: D103
    return link_to_entity(entity_name="company", entity_id=company_id)

link_to_contract

link_to_contract(contract_id)

Source code in components/es/public/helpers/flask_admin_link.py

def link_to_contract(contract_id: UUID) -> str:  # noqa: D103
    return link_to_entity(entity_name="healthcontract", entity_id=contract_id)

link_to_employment_detail

link_to_employment_detail(employment_detail_id)

Source code in components/es/public/helpers/flask_admin_link.py

def link_to_employment_detail(employment_detail_id: UUID) -> str:  # noqa: D103
    return link_to_entity(
        entity_name="employment_detail", entity_id=employment_detail_id
    )

link_to_entity

link_to_entity(entity_name, entity_id)

Source code in components/es/public/helpers/flask_admin_link.py

def link_to_entity(entity_name: str, entity_id: int | UUID) -> str:  # noqa: D103
    return f"{current_config['BASE_URL']}/admin/{format_entity_name_for_link(entity_name)}/details/?id={entity_id}"

link_to_health_contract_version

link_to_health_contract_version(health_contract_version_id)

Source code in components/es/public/helpers/flask_admin_link.py

def link_to_health_contract_version(health_contract_version_id: UUID) -> str:  # noqa: D103
    return link_to_entity(
        entity_name="health_contract_version", entity_id=health_contract_version_id
    )

link_to_policy

link_to_policy(policy_id)

Source code in components/es/public/helpers/flask_admin_link.py

def link_to_policy(policy_id: int) -> str:  # noqa: D103
    return link_to_entity(entity_name="policy", entity_id=policy_id)

link_to_user

link_to_user(user_id)

Source code in components/es/public/helpers/flask_admin_link.py

def link_to_user(user_id: int) -> str:  # noqa: D103
    return link_to_entity(entity_name="user", entity_id=user_id)

front_end

FRONT_END_PATHS module-attribute

FRONT_END_PATHS = dict(
    MARMOT_URL="/marmot",
    DEPENDENT_INVITE_URL="/partner-onboarding",
    LOGIN_URL="/login",
    PASSWORD_RESET_BASE_URL="/password_reset",
    APP_URL="/dashboard",
    UNSUBSCRIBE_URL="/unsubscribe",
    MARMOT_ACCOUNT_URL="/TODO-TODO-TODO",
    MARMOT_COMPANY_URL="/marmot/company/",
    MARMOT_USER_URL="/marmot/user/",
    CUSTOMER_ADMIN_ONBOARDING_INVITE_URL="/onboarding/customer-admin-onboarding",
    ONBOARDING_RESEND_EMAIL_URL_EN="/en/onboarding/resend-email",
    ONBOARDING_RESEND_EMAIL_URL_ES="/es/onboarding/resend-email",
)

front_end_url module-attribute

front_end_url = FrontendURL(paths=FRONT_END_PATHS)

init_data_loader

init_data_loader

init_data_loader()

Create the base data to be populated in fresh local DBs.

Executed by "flask data init".

Please edit the README.md file in components/es when editing this file.

Source code in components/es/public/helpers/init_data_loader.py

def init_data_loader() -> None:
    """
    Create the base data to be populated in fresh local DBs.

    Executed by "flask data init".

    Please edit the README.md file in components/es when editing this file.
    """
    import sys

    from shared.helpers.env import env, is_development_mode, is_test_mode

    # Disable outside of dev/test
    if not is_test_mode() and not is_development_mode():
        return

    # Disable for Kay
    # See:
    # - https://alanhealth.slack.com/archives/C19FZEB41/p1744704414107429
    # - https://alanhealth.slack.com/archives/C19FZEB41/p1759848762076119?thread_ts=1759820409.751299&cid=C19FZEB41
    for kay_arg in ["-k", "--use-kay-data", "--use-kay", "kay"]:
        if kay_arg in sys.argv:
            return
    for kay_env_var in ["USE_KAY", "USE_SHARED_KAY"]:
        if env.str(kay_env_var, None):
            return

    from components.es.subcomponents.healthy_benefits.internal.business_logic.authorizations.helpers import (
        create_default_entities,
    )

    create_default_entities()

marmot_link

adyen_transfer_link

adyen_transfer_link(transfer_id)

Source code in components/es/public/helpers/marmot_link.py

def adyen_transfer_link(transfer_id: str) -> str:  # noqa: D103
    return f"{current_config['ADYEN_BALANCE_PLATFORM_BACKOFFICE_URL']}/balanceplatform/transfers/{transfer_id}"

link_to_company

link_to_company(company_id)

Source code in components/es/public/helpers/marmot_link.py

def link_to_company(company_id: UUID) -> str:  # noqa: D103
    return link_to_entity(entity_name="company", entity_id=company_id)

link_to_entity

link_to_entity(entity_name, entity_id)

Source code in components/es/public/helpers/marmot_link.py

def link_to_entity(entity_name: str, entity_id: int | UUID) -> str:  # noqa: D103
    return f"{current_config['FRONT_END_BASE_URL']}/marmot/{entity_name}/{entity_id}"

link_to_user

link_to_user(user_id)

Source code in components/es/public/helpers/marmot_link.py

def link_to_user(user_id: UUID) -> str:  # noqa: D103
    return link_to_entity(entity_name="user", entity_id=user_id)

components.es.public.scim_api

adapter

EsScimAdapter

EsScimAdapter()

Bases: GenericScimAdapter

SCIM adapter for es_api.

Source code in components/es/public/scim_api/adapter.py

def __init__(self) -> None:
    super().__init__()
    self.user_service = user_service

create_app_user

create_app_user(first_name, last_name, email)

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

Source code in components/es/public/scim_api/adapter.py

@override
def create_app_user(
    self, first_name: str, last_name: str, email: str
) -> int | uuid.UUID:
    """
    Create a user with the given first and last name. and returns the user ID.
    """
    profile_data: CreateUserProfileData = {
        "first_name": first_name,
        "last_name": last_name,
    }

    user = self.user_service.create_user(profile_data=profile_data)
    return user.id

get_scim_users_data

get_scim_users_data(alan_employees)

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

Source code in components/es/public/scim_api/adapter.py

@override
def get_scim_users_data(
    self,
    alan_employees: list[EsAlanEmployee],  # type: ignore[override]
) -> dict[int | uuid.UUID, AlanEmployeeIdentity]:
    """
    Returns the first and last name of users from a list of AlanEmployee objects.
    """
    user_profiles = self.user_service.get_user_profiles(
        user_ids={alan_employee.user_id for alan_employee in alan_employees}
    )

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

get_user_data

get_user_data(user_id)

Returns user's first and last name by user_id.

Source code in components/es/public/scim_api/adapter.py

@override
def get_user_data(self, user_id: int | uuid.UUID) -> AlanEmployeeIdentity:
    """
    Returns user's first and last name by user_id.
    """
    if not isinstance(user_id, uuid.UUID):
        raise TypeError("User ID must be a UUID")

    user_profile = self.user_service.get_user_profile(user_id=user_id)

    if user_profile is None:
        raise BaseErrorCode.missing_resource(message=f"User {user_id} not found")

    return AlanEmployeeIdentity(
        first_name=user_profile.first_name, last_name=user_profile.last_name
    )

user_service instance-attribute

user_service = user_service

test

test_adapter

adapter

adapter()

Fixture for the EsGenericScimAdapter instance.

Source code in components/es/public/scim_api/test/test_adapter.py

@pytest.fixture
def adapter() -> EsScimAdapter:
    """Fixture for the EsGenericScimAdapter instance."""
    return EsScimAdapter()

profile_service

profile_service()

Fixture for the profile service.

Source code in components/es/public/scim_api/test/test_adapter.py

@pytest.fixture
def profile_service() -> ProfileService:
    """Fixture for the profile service."""
    return ProfileService.create(app_name=AppName.ALAN_ES)

test_create_app_user

test_create_app_user(adapter, profile_service)

Test create_app_user creates a new user correctly.

Source code in components/es/public/scim_api/test/test_adapter.py

@pytest.mark.usefixtures("db")
def test_create_app_user(adapter, profile_service):
    """Test create_app_user creates a new user correctly."""
    user_id = adapter.create_app_user(
        first_name="John", last_name="Doe", email="john.doe@alan.eu"
    )

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

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

test_get_scim_users_data

test_get_scim_users_data(adapter)

Test get_scim_users_data returns correct mapping of user data.

Source code in components/es/public/scim_api/test/test_adapter.py

@pytest.mark.usefixtures("db")
def test_get_scim_users_data(adapter):
    """Test get_scim_users_data returns correct mapping of user data."""
    # Create test data
    user1, user2, employee1, employee2 = _provision_test_data()

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

test_get_user_data

test_get_user_data(adapter)

Test get_user_data returns correct user identity.

Source code in components/es/public/scim_api/test/test_adapter.py

@pytest.mark.usefixtures("db")
def test_get_user_data(adapter):
    """Test get_user_data returns correct user identity."""
    # Create test data
    user1, _, _, _ = _provision_test_data()

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

    # Test with non-UUID
    with pytest.raises(TypeError):
        adapter.get_user_data(user_id=123)

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

components.es.public.services

push_notifications

get_push_notification_logs_for_user

get_push_notification_logs_for_user(
    user_id, notification_names, created_at__gte=None
)

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

Source code in components/es/public/services/push_notifications.py

def get_push_notification_logs_for_user(
    user_id: uuid.UUID,
    notification_names: list[BasePushNotificationName],
    created_at__gte: datetime | None = None,
) -> list[InMemoryPushNotificationLog]:
    """
    Return a list of all the push notification logs ever created for the given user and notification names.
    """
    return [
        InMemoryPushNotificationLog.from_model(notification_log=item)
        for item in current_session.query(EsPushNotificationLog).filter(  # noqa: ALN085
            EsPushNotificationLog.user_id == user_id,
            EsPushNotificationLog.name.in_(notification_names),
            EsPushNotificationLog.created_at >= created_at__gte  # type: ignore[arg-type]
            if created_at__gte
            else True,
        )
    ]

get_push_notification_tokens_for_user

get_push_notification_tokens_for_user(user_id)

Returns list of active push notification tokens for a user, that we can send push notifications to Filters tokens with has_permission=True or has_permission=None

Source code in components/es/public/services/push_notifications.py

def get_push_notification_tokens_for_user(user_id: uuid.UUID) -> list[str]:
    """
    Returns list of active push notification tokens for a user, that we can send push notifications to
    Filters tokens with has_permission=True or has_permission=None
    """
    from components.es.internal.models.brokers.push_notification_token import (
        EsPushNotificationTokenModelBroker,
    )

    return EsPushNotificationTokenModelBroker.get_active_tokens_by_user(user_id)

push_notification_sender_async

push_notification_sender_async(sender)

Source code in components/es/internal/push_notifications/push_notification_sender.py

def push_notification_sender_async(
    sender: Callable[P, PushNotificationParams | None],
) -> Callable[P, None]:
    @wraps(sender)
    def decorated_function(*args, **kwargs) -> None:  # type: ignore[no-untyped-def]
        pn_params: PushNotificationParams | None = sender(*args, **kwargs)

        if pn_params is None:
            return

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

    return decorated_function

push_notification_sender_sync

push_notification_sender_sync(sender)

Source code in components/es/internal/push_notifications/push_notification_sender.py

def push_notification_sender_sync(
    sender: Callable[P, PushNotificationParams | None],
) -> Callable[P, None]:
    @wraps(sender)
    def decorated_function(*args, **kwargs) -> None:  # type: ignore[no-untyped-def]
        pn_params: PushNotificationParams | None = sender(*args, **kwargs)

        if pn_params is None:
            return

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

    return decorated_function

components.es.public.test_data_generator

get_test_data_generation_config

get_test_data_generation_config()

Source code in components/es/internal/admin_tools/test_data_generator/test_data_generation_config.py

def get_test_data_generation_config() -> TestDataGeneratorConfig:
    return TestDataGeneratorConfig(
        patched_factories=patched_factories,
        handlers=get_fixture_handlers(),
        async_queue=LOW_PRIORITY_QUEUE,
    )

components.es.public.user

get_alan_employee_email_for_user_id

get_alan_employee_email_for_user_id(user_id)

Get Alan employee email for user id

Source code in components/es/public/user.py

def get_alan_employee_email_for_user_id(user_id: uuid.UUID) -> str | None:
    """
    Get Alan employee email for user id
    """
    from components.es.internal.business_logic.user_v2.service import user_service
    from components.es.public.feature import is_user_alaner

    if not is_user_alaner(user_id):
        return None

    user = user_service.get_or_raise_user(user_id)
    assert user.db_model.alan_employee is not None

    return user.db_model.alan_employee.alan_email

get_user

get_user(user_id)

Source code in components/es/public/user.py

def get_user(user_id: uuid.UUID) -> User:  # noqa: D103
    from components.es.internal.business_logic.user_v2.service import user_service

    return user_service.get_or_raise_user(user_id)

get_user_address

get_user_address(user_id)

Source code in components/es/public/user.py

def get_user_address(user_id: uuid.UUID) -> Address | None:  # noqa: D103
    from components.es.internal.business_logic.user_v2.service import user_service

    user_profile = user_service.get_or_raise_user_profile(user_id)
    return user_profile.address

get_user_details

get_user_details(user_id)

Get user details from user id

Source code in components/es/public/user.py

def get_user_details(user_id: uuid.UUID) -> EsUserDetails:
    """
    Get user details from user id
    """
    from components.es.internal.business_logic.user_v2.service import user_service

    user_profile = user_service.get_or_raise_user_profile(user_id)
    return EsUserDetails(
        id=user_id,
        first_name=user_profile.first_name,  # type: ignore[arg-type]
        last_name=user_profile.last_name,  # type: ignore[arg-type]
        email=user_profile.email or "",
        birth_date=user_profile.birth_date,
        gender=user_profile.gender_compat,
    )

get_user_details_from_insurance_profiles

get_user_details_from_insurance_profiles(
    insurance_profile_ids,
)

Get user details from insurance profiles

Source code in components/es/public/user.py

def get_user_details_from_insurance_profiles(
    insurance_profile_ids: list[uuid.UUID],
) -> list[EsUserDetails]:
    """
    Get user details from insurance profiles
    """
    from components.es.internal.business_logic.user_v2.service import user_service
    from components.es.internal.models.es_insurance_profile import EsInsuranceProfile

    insurance_profiles: list[EsInsuranceProfile] = (
        current_session.query(EsInsuranceProfile)  # noqa: ALN085
        .filter(EsInsuranceProfile.id.in_(insurance_profile_ids))
        .all()
    )
    user_ids = {insurance_profile.user_id for insurance_profile in insurance_profiles}
    user_profiles = user_service.get_or_raise_user_profiles(user_ids)

    return [
        EsUserDetails(
            id=insurance_profile.user_id,
            first_name=user_profiles[insurance_profile.user_id].first_name,  # type: ignore[arg-type]
            last_name=user_profiles[insurance_profile.user_id].last_name,  # type: ignore[arg-type]
            email=user_profiles[insurance_profile.user_id].email or "",
            birth_date=user_profiles[insurance_profile.user_id].birth_date,
            gender=user_profiles[insurance_profile.user_id].gender_compat,
            insurance_profile_id=insurance_profile.id,
        )
        for insurance_profile in insurance_profiles
        if user_profiles[insurance_profile.user_id].email
    ]

get_user_email

get_user_email(user_id)

Source code in components/es/public/user.py

def get_user_email(user_id: uuid.UUID) -> str | None:  # noqa: D103
    from components.es.internal.business_logic.user_v2.service import user_service

    user_profile = user_service.get_or_raise_user_profile(user_id)
    return user_profile.email

get_user_from_email

get_user_from_email(email)

Source code in components/es/public/user.py

def get_user_from_email(email: str) -> User | None:  # noqa: D103
    from components.es.internal.business_logic.user_v2.service import user_service

    user = user_service.get_user_by_email(email=email)
    return user if user else None

get_user_full_name

get_user_full_name(user_id)

Source code in components/es/public/user.py

def get_user_full_name(user_id: uuid.UUID) -> str:  # noqa: D103
    from components.es.internal.business_logic.user_v2.service import user_service

    user_profile = user_service.get_or_raise_user_profile(user_id)
    return user_profile.full_name

get_visible_dependents

get_visible_dependents(user_id)

Source code in components/es/public/user.py

def get_visible_dependents(user_id: uuid.UUID):  # type: ignore[no-untyped-def]  # noqa: D103
    from components.es.internal.business_logic.dependent import dependent_logic

    return dependent_logic.get_visible_dependents(user_id)

is_user

is_user(user_id)

Source code in components/es/public/user.py

def is_user(user_id: uuid.UUID) -> bool:  # noqa: D103
    return get_resource_or_none(EsUser, user_id) is not None

user_can_write

user_can_write(user_id, target_user_id)

Source code in components/es/public/user.py

def user_can_write(user_id: uuid.UUID, target_user_id: uuid.UUID) -> bool:  # noqa: D103
    from components.es.internal.business_logic.dependent import dependent_logic

    if user_id == target_user_id:
        return True

    return dependent_logic.is_user_able_to_modify_user(
        current_user_id=user_id, other_user_id=target_user_id
    )

user_has_access_to_free_physiotherapy_sessions

user_has_access_to_free_physiotherapy_sessions(user_id)

Source code in components/es/public/user.py

def user_has_access_to_free_physiotherapy_sessions(  # noqa: D103
    user_id: uuid.UUID,
) -> bool:
    from components.es.internal.business_logic.enrollment.queries.insurance_affiliation import (
        is_user_enrolled_to_product_or_add_on,
    )

    return is_user_enrolled_to_product_or_add_on(
        user_id=user_id, product_or_add_on=ProductType.ALAN_MANZANA
    )

user_has_access_to_therapy

user_has_access_to_therapy(user_id)

Source code in components/es/public/user.py

def user_has_access_to_therapy(user_id: uuid.UUID) -> bool:  # noqa: D103
    from components.es.internal.business_logic.enrollment.queries.insurance_affiliation import (
        is_user_enrolled_to_product_or_add_on,
    )

    return is_user_enrolled_to_product_or_add_on(
        user_id=user_id, product_or_add_on=ProductType.ALAN_MANZANA
    )