API Reference

components.ca.public.auth

authorization

AuthorizationStrategies

Authorization strategies for CA

alaner_admin class-attribute instance-attribute

alaner_admin = CaAlanerAdminStrategy

authenticated class-attribute instance-attribute

authenticated = CaAuthenticatedStrategy

open class-attribute instance-attribute

open = CaOpenStrategy

owner_only class-attribute instance-attribute

owner_only = CaOwnerOnlyStrategy

CaAlanerAdminStrategy

CaAlanerAdminStrategy(permitted_for=None)

Bases: AlanerAdminStrategy

AlanerAdminStrategy for CA (see class AlanerAdminStrategy in shared/iam/authorization.py)

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

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

CaAuthenticatedStrategy

CaAuthenticatedStrategy(allow_deep_link=False)

Bases: AuthenticatedStrategy

AuthenticatedStrategy for CA (see class AuthenticatedStrategy in shared/iam/authorization.py)

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

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

CaOpenStrategy

CaOpenStrategy(allow_deep_link=False)

Bases: OpenStrategy

OpenStrategy for CA (see class OpenStrategy in shared/iam/authorization.py)

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

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

CaOwnerOnlyStrategy

CaOwnerOnlyStrategy(
    owner_bypass_permitted_for=None, allow_deep_link=False
)

Bases: OwnerOnlyStrategy

OwnerOnlyStrategy for CA (see class OwnerOnlyStrategy in shared/iam/authorization.py)

Source code in components/ca/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,
    )

components.ca.public.blueprints

admin_api_blueprint

admin_api_blueprint module-attribute

admin_api_blueprint = CustomBlueprint("admin_api", __name__)

admin_tools

CaAdminToolsBlueprint

Bases: AdminToolsBlueprint, ServerSideAdminToolBlueprint

route

route(rule, **options)

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

def route(self, rule: str, **options: Any) -> Callable[[_VT], _VT]:  # type: ignore[override] # noqa: D102
    # Required for mypy: incompatible diamond inheritance (ServerSideAdminToolBlueprint is not typed)
    return AdminToolsBlueprint.route(self, rule, **options)

admin_tools_blueprint module-attribute

admin_tools_blueprint = CaAdminToolsBlueprint(
    "admin_tools", __name__
)

config_usage

config_usage()

Default admin tools: see shared admin tools

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

@admin_tools_blueprint.route("/config_usage", methods=["GET"])
def config_usage() -> str:
    """Default admin tools: see shared admin tools"""
    from shared.blueprints.commands import render_config_usage_count_by_key

    return render_config_usage_count_by_key(current_config)

customize_email

customize_email(template_name)

Customize an email template

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

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

    return render_customize_email(template_name)

document_previews

document_previews()

Default admin tools: see shared admin tools

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

@admin_tools_blueprint.route("/document_previews")
def document_previews() -> str:
    """Default admin tools: see shared admin tools"""
    admin_tools_blueprint.register_all_documents()

    return render_template(
        "admin_tools/document_previews.html",
        documents=admin_tools_blueprint.documents.values(),
    )

index

index()

Default admin tools: see shared admin tools

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

@admin_tools_blueprint.route("/")
def index() -> str:
    """Default admin tools: see shared admin tools"""
    return render_template("admin_tools/index.html")

init_blueprint

init_blueprint(state)

Initialize the admin tools blueprint

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

@admin_tools_blueprint.record_once
def init_blueprint(state: BlueprintSetupState) -> None:
    """Initialize the admin tools blueprint"""
    with state.app.app_context():

        def cb(admin_tool: AdminToolsBlueprint) -> None:
            pass

        admin_tools_blueprint.register_document_registration_callback(cb)

list_api_endpoints

list_api_endpoints()

Default admin tools: see shared admin tools

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

@admin_tools_blueprint.route("/api/documentation", methods=["GET"])
def list_api_endpoints() -> str:
    """Default admin tools: see shared admin tools"""
    from shared.blueprints.api_documentation import render_list_api_endpoints

    return render_list_api_endpoints(layout="admin_tools/_layout.html")

list_commands

list_commands()

Default admin tools: see shared admin tools

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

@admin_tools_blueprint.route("/commands/list", methods=["GET"])
def list_commands() -> str:
    """Default admin tools: see shared admin tools"""
    from shared.blueprints.commands import render_list_commands

    return render_list_commands()

list_email

list_email()

Default admin tools: see shared admin tools

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

@admin_tools_blueprint.route("/mails/list", methods=["GET"])
def list_email() -> str:
    """Default admin tools: see shared admin tools"""
    from shared.blueprints.admin_tools.blueprint import render_list_email

    return render_list_email()

load_admin_tools_blueprint

load_admin_tools_blueprint()

Load the admin tools blueprint

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

def load_admin_tools_blueprint() -> AdminToolsBlueprint:
    """Load the admin tools blueprint"""
    return admin_tools_blueprint

marmot_account_company_placeholder_creation

marmot_account_company_placeholder_creation()

Marmot admin tools: account and company placeholder creation

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

@admin_tools_blueprint.route(
    "/marmot/account_company_placeholder_creation", methods=["GET", "POST"]
)
def marmot_account_company_placeholder_creation() -> str:
    """Marmot admin tools: account and company placeholder creation"""
    from components.ca.internal.admin_tools.account_company_placeholder_creation import (
        render_marmot_account_company_placeholder_creation_template,
    )

    return render_marmot_account_company_placeholder_creation_template()

marmot_company_admin_invitation

marmot_company_admin_invitation()

Marmot admin tools: company admin invitation

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

@admin_tools_blueprint.route(
    "/marmot/company_admin_invitation", methods=["GET", "POST"]
)
def marmot_company_admin_invitation() -> str:
    """Marmot admin tools: company admin invitation"""
    from components.ca.internal.admin_tools.company_admin_invitation import (
        render_marmot_company_admin_invitation_template,
    )

    return render_marmot_company_admin_invitation_template()

marmot_contract_amendment

marmot_contract_amendment()

Marmot admin tools: contract amendment

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

@admin_tools_blueprint.route("/marmot/contract_amendment", methods=["GET", "POST"])
def marmot_contract_amendment() -> str:
    """Marmot admin tools: contract amendment"""
    from components.ca.internal.admin_tools.contract_amendment import (
        render_marmot_contract_amendment_template,
    )

    return render_marmot_contract_amendment_template()

marmot_contract_creation

marmot_contract_creation()

Marmot admin tools: contract creation

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

@admin_tools_blueprint.route("/marmot/contract_creation", methods=["GET", "POST"])
def marmot_contract_creation() -> str:
    """Marmot admin tools: contract creation"""
    from components.ca.internal.admin_tools.contract_creation import (
        render_marmot_contract_creation_template,
    )

    return render_marmot_contract_creation_template()

marmot_create_reimbursement_request

marmot_create_reimbursement_request()

Marmot admin tools: create reimbursement request

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

@admin_tools_blueprint.route(
    "/marmot/create_reimbursement_request", methods=["GET", "POST"]
)
def marmot_create_reimbursement_request() -> str:
    """Marmot admin tools: create reimbursement request"""
    from components.ca.internal.admin_tools.create_reimbursement_request import (
        render_marmot_create_reimbursement_request_template,
    )

    return render_marmot_create_reimbursement_request_template()

marmot_create_settlement

marmot_create_settlement()

Marmot admin tools: create settlement

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

@admin_tools_blueprint.route("/marmot/create_settlement", methods=["GET", "POST"])
def marmot_create_settlement() -> str:
    """Marmot admin tools: create settlement"""
    from components.ca.internal.admin_tools.create_settlement import (
        render_marmot_create_settlement_template,
    )

    user_id = g.current_user.id

    return render_marmot_create_settlement_template(user_id)

marmot_employee_invitation

marmot_employee_invitation()

Marmot admin tools: employee invitation

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

@admin_tools_blueprint.route("/marmot/employee_invitation", methods=["GET", "POST"])
def marmot_employee_invitation() -> str:
    """Marmot admin tools: employee invitation"""
    from components.ca.internal.admin_tools.employee_invitation import (
        render_marmot_employee_invitation_template,
    )

    return render_marmot_employee_invitation_template()

marmot_employee_termination

marmot_employee_termination()

Marmot admin tools: employee termination

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

@admin_tools_blueprint.route("/marmot/employee_termination", methods=["GET", "POST"])
def marmot_employee_termination() -> str:
    """Marmot admin tools: employee termination"""
    from components.ca.internal.admin_tools.employee_termination import (
        render_marmot_employee_termination_template,
    )

    user_id = g.current_user.id

    return render_marmot_employee_termination_template(user_id)

marmot_exemption_review

marmot_exemption_review()

Marmot admin tools: exemption review

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

@admin_tools_blueprint.route("/marmot/exemption_review", methods=["GET", "POST"])
def marmot_exemption_review() -> str:
    """Marmot admin tools: exemption review"""
    from components.ca.internal.admin_tools.exemption_review import (
        render_marmot_exemption_review_template,
    )

    return render_marmot_exemption_review_template()

marmot_impersonate

marmot_impersonate()

Marmot admin tools: create settlement

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

@admin_tools_blueprint.route("/marmot/impersonate", methods=["GET", "POST"])
def marmot_impersonate() -> str:
    """Marmot admin tools: create settlement"""
    from components.ca.internal.admin_tools.impersonate import (
        render_marmot_impersonate_template,
    )

    return render_marmot_impersonate_template()

marmot_manual_exemption

marmot_manual_exemption()

Marmot admin tools: manual exemption

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

@admin_tools_blueprint.route("/marmot/manual_exemption", methods=["GET", "POST"])
def marmot_manual_exemption() -> str:
    """Marmot admin tools: manual exemption"""
    from components.ca.internal.admin_tools.manual_exemption import (
        render_marmot_manual_exemption_template,
    )

    return render_marmot_manual_exemption_template()

marmot_onboarding_invitation

marmot_onboarding_invitation()

Marmot admin tools: onboarding invitation

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

@admin_tools_blueprint.route("/marmot/onboarding_invitation", methods=["GET", "POST"])
def marmot_onboarding_invitation() -> str:
    """Marmot admin tools: onboarding invitation"""
    from components.ca.internal.admin_tools.onboarding_invitation import (
        render_marmot_onboarding_invitation_template,
    )

    return render_marmot_onboarding_invitation_template()

marmot_upload_exemption_justification

marmot_upload_exemption_justification()

Marmot admin tools: upload exemption justification

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

@admin_tools_blueprint.route(
    "/marmot/upload_exemption_justification", methods=["GET", "POST"]
)
def marmot_upload_exemption_justification() -> str:
    """Marmot admin tools: upload exemption justification"""
    from components.ca.internal.admin_tools.upload_exemption_justification import (
        render_marmot_upload_exemption_justification_template,
    )

    return render_marmot_upload_exemption_justification_template()

show_command

show_command(command_full_name)

Default admin tools: see shared admin tools

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

@admin_tools_blueprint.route("/commands/<path:command_full_name>")
def show_command(command_full_name: str) -> str:
    """Default admin tools: see shared admin tools"""
    from shared.blueprints.commands import render_show_command

    return render_show_command(command_full_name=command_full_name)

upload_file

upload_file()

Default admin tools: see shared admin tools

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

@admin_tools_blueprint.route("/upload_file", methods=["GET", "POST"])
def upload_file() -> str:
    """Default admin tools: see shared admin tools"""
    from shared.blueprints.admin_tools.blueprint import (
        render_upload_file,
    )

    return render_upload_file()

view_email

view_email(template_name)

View an email template

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

@admin_tools_blueprint.route("/mails/see/<path:template_name>")
def view_email(template_name: str) -> str:
    """View an email template"""
    from shared.blueprints.admin_tools.blueprint import (
        view_email as shared_view_email,
    )

    return shared_view_email(template_name)

ca_api_blueprint

ca_api_blueprint module-attribute

ca_api_blueprint = CustomBlueprint(
    "ca_api_blueprint", __name__
)

ca_core_blueprint

ca_core_blueprint module-attribute

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

register_blueprint

register_blueprint(state)

Source code in components/ca/public/blueprints/ca_core_blueprint.py

@ca_core_blueprint.record_once
def register_blueprint(state) -> None:  # type: ignore[no-untyped-def]  # noqa: ARG001, D103
    import components.ca.internal.models  # noqa: F401

ca_services_blueprint

ca_services_blueprint module-attribute

ca_services_blueprint = CustomBlueprint(
    "ca_services", __name__
)

record_once

record_once(state)

Source code in components/ca/public/blueprints/ca_services_blueprint.py

@ca_services_blueprint.record_once
def record_once(state: BlueprintSetupState) -> None:  # noqa: ARG001, D103
    # force registration of routes from those packages and files
    import components.ca.internal.services.payment_provider  # noqa: F401, RUF100

components.ca.public.clinic

adapter

CaClinicAdapter

Bases: ClinicAdapter

Adapter for the CA clinic

create_external_user

create_external_user(onboarding_data, profile_service)

Create an external teleconsultation user profile with onboarding data for CA.

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

def create_external_user(
    self,
    onboarding_data: ExternalOnboardingUserData,
    profile_service: ProfileService,
) -> tuple[Any, RefreshTokenType]:
    """Create an external teleconsultation user profile with onboarding data for CA."""
    # CA doesn't support external user creation
    raise NotImplementedError("CA doesn't support external user creation")

currency class-attribute instance-attribute

currency = CAD

get_allowlist_of_dermatology_medical_admins_ids

get_allowlist_of_dermatology_medical_admins_ids()

Get the allowlist of dermatology medical admin IDs.

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

def get_allowlist_of_dermatology_medical_admins_ids(self) -> list[str]:
    """Get the allowlist of dermatology medical admin IDs."""
    # No dermatology sessions in CA
    return []

get_app_base_user_data

get_app_base_user_data(app_user_id)

Get base user data for the CA clinic.

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

def get_app_base_user_data(self, app_user_id: str) -> BaseUserData:
    """Get base user data for the CA clinic."""
    user = UserBroker.get_by_id(UUID(app_user_id)).one_or_none()

    if user is None:
        raise BaseErrorCode.missing_resource(
            message=f"User not found with id {app_user_id}"
        )
    profile_service: ProfileService = ProfileService.create(
        app_name=AppName.ALAN_CA
    )
    profile = profile_service.get_profile(profile_id=user.profile_id)
    if profile is None:
        raise BaseErrorCode.missing_resource(
            message=f"Profile not found with id {user.profile_id}"
        )

    return BaseUserData(
        first_name=normalize_name(profile.first_name),
        last_name=normalize_name(profile.last_name),
    )

get_app_user_available_health_services

get_app_user_available_health_services(app_user_id)

Get available health services for the user.

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

def get_app_user_available_health_services(
    self,
    app_user_id: str,  # noqa: ARG002
) -> list[AvailableHealthService]:
    """Get available health services for the user."""
    # For now, we allow all users to access all health services
    return [
        AvailableHealthService(name=AvailableHealthServiceName.DATO_CONTENT),
        AvailableHealthService(name=AvailableHealthServiceName.HEALTH_PROGRAM),
        AvailableHealthService(name=AvailableHealthServiceName.THERAPY_SESSION),
        AvailableHealthService(name=AvailableHealthServiceName.THERAPIST),
        AvailableHealthService(name=AvailableHealthServiceName.ORIENTATION_CALL),
    ]

get_app_user_data

get_app_user_data(
    app_user_id, compute_key_account_info=False
)

Get the user data for the CA clinic

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

def get_app_user_data(
    self,
    app_user_id: str,
    compute_key_account_info: bool = False,  # noqa: ARG002
) -> UserData:
    """Get the user data for the CA clinic"""
    profile_service: ProfileService = ProfileService.create(
        app_name=AppName.ALAN_CA
    )
    user = UserBroker.get_by_id_with_load_options(UUID(app_user_id)).one()

    policy_with_all_enrollments_and_insurance_profiles = (
        PolicyBroker.get_by_profile_id_with_load_options(
            user.profile_id
        ).one_or_none()
    )
    if policy_with_all_enrollments_and_insurance_profiles is None:
        raise BaseErrorCode.missing_resource(
            message=f"No policy found for user with id {user.id}"
        )

    # The policy contains all the enrollments, along with each enrollment's insurance profile,
    # We'll use the profile ids in the insurance profiles to get all the profiles associated with the policy
    profile_ids = [
        enrollment.insurance_profile.profile_id
        for enrollment in policy_with_all_enrollments_and_insurance_profiles.enrollments
    ]

    profiles = profile_service.get_or_raise_profiles(profile_ids)

    users = UserBroker.get_by_profile_id_in(profile_ids).all()

    profile_id_to_user_id = {
        user.profile_id: user.id for user in users if user.profile_id is not None
    }
    profile_id_to_enrollment_type = {
        enrollment.insurance_profile.profile_id: enrollment.enrollment_type
        for enrollment in policy_with_all_enrollments_and_insurance_profiles.enrollments
    }

    extended_profiles: list[ProfileWithUserIdAndEnrollmentType] = [
        ProfileWithUserIdAndEnrollmentType(
            profile=profile,
            user_id=profile_id_to_user_id.get(profile.id),
            enrollment_type=profile_id_to_enrollment_type.get(profile.id),
        )
        for profile in profiles
    ]

    primary_extended_profile = next(
        (
            extended_profile
            for extended_profile in extended_profiles
            if extended_profile.enrollment_type == EnrollmentType.primary
        ),
        None,
    )
    if primary_extended_profile is None:
        raise BaseErrorCode.missing_resource(message="No primary enrollment found")

    # TODO guillaume.gustin clarify what should happen if user has no address set. Can it happen ?
    administrative_area = (
        primary_extended_profile.profile.address.administrative_area
        if primary_extended_profile.profile.address
        else None
    )

    dependents = [
        Dependent(
            app_user_id=str(extended_profile.user_id),
            first_name=normalize_name(extended_profile.profile.first_name or ""),
            last_name=normalize_name(extended_profile.profile.last_name or ""),
            age=int(extended_profile.profile.age)
            if extended_profile.profile.age
            else None,
            gender=_convert_gender_to_user_gender(extended_profile.profile.gender),
            birth_date=extended_profile.profile.birth_date,
            dependent_type=DependentType.PARTNER
            if extended_profile.enrollment_type == EnrollmentType.partner
            else DependentType.CHILD,
        )
        for extended_profile in extended_profiles
        if extended_profile.user_id is not None
        and extended_profile.user_id != user.id
    ]

    primary_profile = primary_extended_profile.profile
    primary_address = primary_profile.address
    return UserData(
        first_name=normalize_name(primary_profile.first_name),
        last_name=normalize_name(primary_profile.last_name),
        gender=_convert_gender_to_user_gender(primary_profile.gender),
        email=primary_profile.email,
        profile_id=primary_profile.id,
        birth_date=primary_profile.birth_date,
        phone=primary_profile.phone_number,
        country=primary_address.country if primary_address else None,
        administrative_area=administrative_area,
        address=(
            f"{primary_address.street}, {primary_address.locality} {primary_address.postal_code}"
            if primary_address
            else ""
        ),
        ssn=None,  # Canada doesn't use SSN
        lang=Lang(primary_profile.preferred_language.value),
        is_key_account_or_large_company_and_not_alaner=False,  # Canada doesn't use this
        is_alaner=user.is_alaner,
        dependents=dependents,
    )

get_booking_session_package

get_booking_session_package(app_user_id, session_type)

Get the booking session package for the specified session type.

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

def get_booking_session_package(
    self,
    app_user_id: str,  # noqa: ARG002
    session_type: TherapistBookingSessionType,
) -> BookingSessionPackage | None:
    """Get the booking session package for the specified session type."""
    if session_type == TherapistBookingSessionType.therapy:
        return BookingSessionPackage(
            price_in_cents=THERAPY_SESSION_PRICE_IN_CENTS,
            included=BookingSessionPackageCount(count_limit=2),
        )
    elif session_type == TherapistBookingSessionType.orientation:
        return BookingSessionPackage(
            price_in_cents=0,
        )
    else:
        return None

get_coverage_status

get_coverage_status(app_user_id)

Get the coverage status for the user.

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

def get_coverage_status(self, app_user_id: str) -> CoverageStatus | None:
    """Get the coverage status for the user."""
    from components.ca.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,
)

CA implementation of getting the last active ID verification request for a user. Since CA doesn't support ID verification for the clinic, this will raise a NotImplementedError if it's called.

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

def get_last_active_id_verification_request_for_user(
    self,
    app_user_id: str,
) -> None:
    """
    CA implementation of getting the last active ID verification request for a user.
    Since CA doesn't support ID verification for the clinic, this will raise a NotImplementedError if it's called.
    """
    raise NotImplementedError(
        "CA doesn't support ID verification for clinic users."
    )

has_access_to_orientation_call

has_access_to_orientation_call(app_user_id)

Check if user has access to orientation calls.

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

def has_access_to_orientation_call(
    self,
    app_user_id: str,  # noqa: ARG002
) -> bool:
    """Check if user has access to orientation calls."""
    # For now, we allow all users to access orientation calls
    return True

has_app_user_permission

has_app_user_permission(app_user_id, permission)

Check if user has the specified permission.

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

def has_app_user_permission(
    self, app_user_id: str, permission: EmployeePermission
) -> bool:
    """Check if user has the specified permission."""
    user = UserBroker.get_by_id_with_load_options(UUID(app_user_id)).one_or_none()
    if user is None:
        raise BaseErrorCode.missing_resource(
            message=f"User not found with id {app_user_id}"
        )
    return has_permission(user, permission)

is_app_user_admin_of_company

is_app_user_admin_of_company(app_user_id, app_company_id)

Check if user is admin of the specified company.

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

def is_app_user_admin_of_company(
    self, app_user_id: str, app_company_id: str
) -> bool:
    """Check if user is admin of the specified company."""
    from components.ca.internal.customer_dashboard.queries.company import (
        user_can_admin_all,
    )

    return user_can_admin_all(UUID(app_user_id), [UUID(app_company_id)])

is_orientation_session_mandatory

is_orientation_session_mandatory()

Check if orientation sessions are mandatory.

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

def is_orientation_session_mandatory(self) -> bool:
    """Check if orientation sessions are mandatory."""
    return False

request_id_verification_request_for_user

request_id_verification_request_for_user(
    app_user_id, user_info, commit=True
)

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

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

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

should_request_id_verification_for_user

should_request_id_verification_for_user(app_user_id)

Since CA doesn't support ID verification for clinic users, this will always return False.

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

def should_request_id_verification_for_user(
    self,
    app_user_id: str,  # noqa: ARG002
) -> bool:
    """
    Since CA doesn't support ID verification for clinic users, this will always return False.
    """
    return False

therapy_refundable_cancellation_limit_in_hours

therapy_refundable_cancellation_limit_in_hours()

Get the therapy refundable cancellation limit in hours.

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

def therapy_refundable_cancellation_limit_in_hours(self) -> int | None:
    """Get the therapy refundable cancellation limit in hours."""
    # 48 hours is the cancellation limit for therapy sessions in CA
    return 48

update_app_user_phone

update_app_user_phone(app_user_id, phone)

Update the user's phone number.

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

def update_app_user_phone(self, app_user_id: str, phone: str | None) -> None:
    """Update the user's phone number."""
    profile_service: ProfileService = ProfileService.create(
        app_name=AppName.ALAN_CA
    )
    user = UserBroker.get_by_id(UUID(app_user_id)).one_or_none()
    if user is None:
        raise BaseErrorCode.missing_resource(
            message=f"User not found with id {app_user_id}"
        )

    profile_service.change_phone_number(
        profile_id=user.profile_id, phone_number=phone
    )
    current_session.commit()

update_app_user_ssn

update_app_user_ssn(app_user_id, ssn, commit=False)

Update the user's SSN.

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

def update_app_user_ssn(
    self, app_user_id: str, ssn: str | None, commit: bool = False
) -> None:
    """Update the user's SSN."""
    # CA doesn't use SSN
    pass

upload_invoice_as_insurance_document

upload_invoice_as_insurance_document(
    file, app_user_id, upload_invoice_data
)

Upload an invoice as an insurance document.

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

def upload_invoice_as_insurance_document(
    self,
    file: IO,  # type: ignore[type-arg] # noqa: ARG002
    app_user_id: str,  # noqa: ARG002
    upload_invoice_data: UploadInvoiceData,  # noqa: ARG002
) -> bool:
    """Upload an invoice as an insurance document."""
    # CA 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/ca/public/clinic/adapter.py

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

validate_session_duration

validate_session_duration(session_duration)

Validate the session duration.

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

def validate_session_duration(
    self,
    session_duration: int,
) -> None:
    """Validate the session duration."""
    if session_duration in UNAUTHORIZED_SESSION_LENGTHS:
        raise ValueError(
            f"Sessions of duration {session_duration} minutes are not available for booking in CA"
        )

ProfileWithUserIdAndEnrollmentType dataclass

ProfileWithUserIdAndEnrollmentType(
    profile, user_id=None, enrollment_type=None
)

Wrapper class that adds user_id and enrollment_type as attributes to a Profile.

enrollment_type class-attribute instance-attribute

enrollment_type = None

profile instance-attribute

profile

user_id class-attribute instance-attribute

user_id = None

THERAPY_SESSION_PRICE_IN_CENTS module-attribute

THERAPY_SESSION_PRICE_IN_CENTS = 15000

UNAUTHORIZED_SESSION_LENGTHS module-attribute

UNAUTHORIZED_SESSION_LENGTHS = [45, 60]

clinic_adapter module-attribute

clinic_adapter = CaClinicAdapter()

clinic_eligibility

get_coverage_status

get_coverage_status(user_id)

Return the start and optionally the end date of the current or upcoming period of eligibility for a user.

Source code in components/ca/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 for a user.
    """
    try:
        latest_enrollment = get_latest_enrollment_on_date(
            user_id=user_id, on_date=date.today()
        )
    except NoResultFound:
        # No enrollment found, the user is not covered yet
        return None
    except Exception as e:
        current_logger.error(
            "Error getting enrollment for user on date",
            user_id=user_id,
            date=date.today().isoformat(),
            error=e,
        )
        return None

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

components.ca.public.command_log

queries

get_command_logs

get_command_logs(start_at, end_at)

Get command logs from the database

Source code in components/ca/public/command_log/queries.py

def get_command_logs(start_at: datetime, end_at: datetime) -> list[CommandLog]:
    """Get command logs from the database"""
    logs = current_session.scalars(
        select(CaCommandLog).filter(
            CaCommandLog.created_at >= start_at, CaCommandLog.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="cacommandlog",
        )
        for log in logs
    ]

components.ca.public.contracting

company

get_account_ids_from_company_ids

get_account_ids_from_company_ids(company_ids)

Get account IDs from company IDs.

Parameters:

Name	Type	Description	Default
company_ids	Iterable[str]	Iterable of company ID strings (UUID format)	required

Returns:

Type	Description
list[str]	Sorted list of unique account ID strings

Source code in components/ca/public/contracting/company.py

def get_account_ids_from_company_ids(company_ids: Iterable[str]) -> list[str]:
    """
    Get account IDs from company IDs.

    Args:
        company_ids: Iterable of company ID strings (UUID format)

    Returns:
        Sorted list of unique account ID strings
    """
    from sqlalchemy import select

    from components.ca.internal.contracting.models.ca_company import CaCompany

    companies = current_session.scalars(
        select(CaCompany).where(CaCompany.id.in_([UUID(cid) for cid in company_ids]))
    ).all()

    account_ids = {
        str(company.account_id) for company in companies if company.account_id
    }
    return sorted(account_ids)

get_company_id_of_user

get_company_id_of_user(user_id)

Source code in components/ca/public/contracting/company.py

def get_company_id_of_user(user_id: UUID) -> UUID | None:  # noqa: D103
    from components.ca.internal.contracting.queries.company import (
        get_company_by_user_id_or_none,
    )

    company = get_company_by_user_id_or_none(user_id)
    return company.id if company else None

get_company_ids_from_user_employments

get_company_ids_from_user_employments(user_id)

Get all company IDs where the user is currently employed.

This queries the Employment Component for active (non-cancelled, non-ended) employments and returns their company IDs.

Parameters:

Name	Type	Description	Default
user_id	str	The user ID as a string (UUID format)	required

Returns:

Type	Description
list[str]	Sorted list of unique company IDs as strings

Source code in components/ca/public/contracting/company.py

def get_company_ids_from_user_employments(user_id: str) -> list[str]:
    """
    Get all company IDs where the user is currently employed.

    This queries the Employment Component for active (non-cancelled,
    non-ended) employments and returns their company IDs.

    Args:
        user_id: The user ID as a string (UUID format)

    Returns:
        Sorted list of unique company IDs as strings
    """
    from components.employment.public.business_logic.queries.core_employment_version import (
        get_core_employments_for_user,
    )

    today = utctoday()
    employments = get_core_employments_for_user(user_id)

    # Filter for active (non-cancelled, non-ended) employments
    company_ids = {
        employment.company_id
        for employment in employments
        if not employment.is_cancelled
        and (employment.end_date is None or employment.end_date >= today)
        and employment.start_date <= today
    }

    return sorted(company_ids)

components.ca.public.employment

ca_country_gateway

CaCountryGateway

Bases: CountryGateway[CaExtendedValues]

Canadian 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/ca/public/employment/ca_country_gateway.py

@override
def are_companies_in_same_account(
    self, company_id_1: str, company_id_2: str
) -> bool:
    from components.ca.internal.contracting.models.ca_company import CaCompany

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

    return company_1.account_id == company_2.account_id

get_account_name

get_account_name(account_id)

Source code in components/ca/public/employment/ca_country_gateway.py

@override
def get_account_name(self, account_id: UUID) -> str:
    from components.ca.internal.contracting.models.ca_account import CaAccount

    account = get_or_raise_missing_resource(CaAccount, account_id)

    return account.name

get_company_information

get_company_information(company_id)

Source code in components/ca/public/employment/ca_country_gateway.py

@override
def get_company_information(
    self,
    company_id: str,
) -> CompanyInformation | None:
    from components.ca.internal.contracting.models.ca_company import CaCompany

    company = get_resource_or_none(CaCompany, company_id)
    if not company:
        return None
    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/ca/public/employment/ca_country_gateway.py

@override
def get_employee_identifier_for_country(
    self, extended_values: CaExtendedValues
) -> str | None: ...

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/ca/public/employment/ca_country_gateway.py

@override
def get_employment_consumers(self) -> set[EmploymentConsumer[CaExtendedValues]]:
    """
    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.ca.public.employment.employment_consumer import (
        ca_health_affiliation_employment_change_consumer,
    )

    return {ca_health_affiliation_employment_change_consumer}

get_upstream_retry_handler

get_upstream_retry_handler(source_type)

Source code in components/ca/public/employment/ca_country_gateway.py

@override
def get_upstream_retry_handler(self, source_type: SourceType) -> None:
    return None

get_user_admined_company_ids

get_user_admined_company_ids(user_id)

Source code in components/ca/public/employment/ca_country_gateway.py

@override
def get_user_admined_company_ids(self, user_id: str) -> list[str]:
    raise NotImplementedError("Not implemented in Canada")

get_user_full_name

get_user_full_name(user_id)

Source code in components/ca/public/employment/ca_country_gateway.py

@override
def get_user_full_name(self, user_id: str) -> str | None:
    from components.ca.internal.tech.models.ca_user import CaUser

    user = get_resource_or_none(CaUser, user_id)
    return user.full_name if user else None

ca_extended_values

CaEmploymentDeclaration module-attribute

CaEmploymentDeclaration = EmploymentDeclaration[
    CaExtendedValues
]

CaExtendedValues

Bases: ExtendedValuesDict

Canadian extended values stored in the Employment Component

email instance-attribute

email

occupation instance-attribute

occupation

rwam_certificate_number instance-attribute

rwam_certificate_number

salary instance-attribute

salary

termination_type instance-attribute

termination_type

viewed_terms_at instance-attribute

viewed_terms_at

employment_consumer

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

ca_health_affiliation_employment_change_consumer

ca_health_affiliation_employment_change_consumer(
    employment_change, event_bus_orchestrator
)

Consumer for employment changes

Source code in components/ca/public/employment/employment_consumer.py

def ca_health_affiliation_employment_change_consumer(
    employment_change: EmploymentChange["CaExtendedValues"],
    event_bus_orchestrator: EventBusOrchestrator,  # noqa: ARG001 we do not need
) -> None:
    """
    Consumer for employment changes
    """
    if employment_change.country_code != CountryCode.ca:
        return

    from components.ca.internal.employment.actions.on_employment_change import (
        on_employment_change,
    )

    on_employment_change(
        employment_change=employment_change,
        commit=False,
    )

components.ca.public.events

beneficiary_created

BeneficiaryCreated dataclass

BeneficiaryCreated(enrollment_id)

Bases: Message

This event is published when a beneficiary has been created successfully

enrollment_id instance-attribute

enrollment_id

contract_amendment

ContractAmendmentCreated dataclass

ContractAmendmentCreated(
    company_id, health_subscription_version_payload_id
)

Bases: Message

Event published when a contract amendment is created for a company.

company_id instance-attribute

company_id

health_subscription_version_payload_id instance-attribute

health_subscription_version_payload_id

subscription

subscribe_to_ca_global_events

subscribe_to_ca_global_events()

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

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

def subscribe_to_ca_global_events() -> None:
    """
    Events subscriptions that should be listened by every runtime for Canada.
    """
    from components.ca.internal.claim_management.claim_engine.steps.reimbursement_payment.business_logic.actions.update_reimbursement_payment import (
        update_reimbursement_payment_status,
    )
    from components.ca.internal.claim_management.claim_engine.steps.reimbursement_payment.jpmorgan import (
        CA_JPMORGAN_WORKSPACE_KEY,
    )
    from components.ca.internal.document_parsing.events.on_document_parsing_validated import (
        on_document_parsing_validated,
    )
    from components.ca.internal.events.subscribers import (
        update_contact_information_in_intercom,
        update_contact_information_in_segment,
        update_identity_information_in_segment,
    )
    from components.ca.internal.events.telus_subscribers import (
        sync_telus_limits_after_beneficiary_created,
        sync_telus_limits_after_contract_amendment,
        sync_telus_limits_after_procedure_processed,
    )
    from components.ca.public.events.beneficiary_created import BeneficiaryCreated
    from components.ca.public.events.contract_amendment import ContractAmendmentCreated
    from components.ca.public.events.telus_procedure import TelusProcedureProcessed
    from components.documents.public.events.document import DocumentParsingValidated
    from components.documents.public.mappers.document_type_to_country_mapper import (
        get_document_types_for_country,
    )
    from components.global_profile.public.events import (
        IdentityInformationChanged,
        PreferredLanguageChanged,
        ProfileAddressChanged,
        ProfileEmailChanged,
    )
    from components.payment_gateway.public.events.transfer_update import (
        PayoutBankTransferUpdated,
    )
    from shared.enums.country import Country
    from shared.messaging.broker import get_message_broker
    from shared.queuing.config import (
        CLAIM_ENGINE_QUEUE,
        LOW_PRIORITY_QUEUE,
        PROFILE_INTERCOM_QUEUE,
    )

    message_broker = get_message_broker()

    # Subscriptions to global profile events
    # CA events
    message_broker.subscribe_async(
        ProfileEmailChanged,
        update_contact_information_in_intercom,
        queue_name=PROFILE_INTERCOM_QUEUE,
    )
    message_broker.subscribe_async(
        ProfileEmailChanged,
        update_contact_information_in_segment,
        queue_name=LOW_PRIORITY_QUEUE,
    )

    message_broker.subscribe_async(
        IdentityInformationChanged,
        update_identity_information_in_segment,
        queue_name=LOW_PRIORITY_QUEUE,
    )

    message_broker.subscribe_async(
        PreferredLanguageChanged,
        update_identity_information_in_segment,
        queue_name=LOW_PRIORITY_QUEUE,
    )

    message_broker.subscribe_async(
        ProfileAddressChanged,
        update_identity_information_in_segment,
        queue_name=LOW_PRIORITY_QUEUE,
    )

    # Subscriptions to document parsing events
    message_broker.subscribe_async(
        DocumentParsingValidated,
        on_document_parsing_validated,
        queue_name=LOW_PRIORITY_QUEUE,
        # Filter only ca insurance documents
        predicate=lambda m: m.document_type
        in get_document_types_for_country(Country.Ca),
    )

    # Subscriptions to Telus events
    message_broker.subscribe_async(
        TelusProcedureProcessed,
        sync_telus_limits_after_procedure_processed,
        queue_name=LOW_PRIORITY_QUEUE,
    )
    message_broker.subscribe_async(
        BeneficiaryCreated,
        sync_telus_limits_after_beneficiary_created,
        queue_name=LOW_PRIORITY_QUEUE,
    )
    message_broker.subscribe_async(
        ContractAmendmentCreated,
        sync_telus_limits_after_contract_amendment,
        queue_name=LOW_PRIORITY_QUEUE,
    )

    # Subscribe to payment gateway events
    message_broker.subscribe_async(
        PayoutBankTransferUpdated,
        update_reimbursement_payment_status,
        # Only for JPMorgan CA workspace
        predicate=lambda m: m.workspace_key == CA_JPMORGAN_WORKSPACE_KEY,
        queue_name=CLAIM_ENGINE_QUEUE,
    )

subscribe_to_events

subscribe_to_events()

All event subscriptions for Canada should be done here.

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

def subscribe_to_events() -> None:
    """
    All event subscriptions for Canada should be done here.
    """
    subscribe_to_ca_global_events()

telus_procedure

TelusProcedureProcessed dataclass

TelusProcedureProcessed(enrollment_id, procedure_id)

Bases: Message

This event is published when a Telus procedure has been processed successfully and an enrollment has been identified for limit synchronization.

enrollment_id instance-attribute

enrollment_id

procedure_id instance-attribute

procedure_id

components.ca.public.helpers

anonymization

CaDbAnonymizerState

Bases: BaseDbAnonymizerState

State for the CA anonymization (used in Turing)

anonymization_queries

anonymization_queries()

Anonymization queries for CA anonymization

Source code in components/ca/public/helpers/anonymization.py

def anonymization_queries(self) -> list[sql.Composable]:
    """Anonymization queries for CA anonymization"""
    return super().anonymization_queries() + signed_bundles_anonymization_queries(
        schema="ca"
    )

available_profile_classes

available_profile_classes()

Available profile classes CA anonymization

Source code in components/ca/public/helpers/anonymization.py

def available_profile_classes(self) -> list[type[BaseDbAnonymizerProfile]]:
    """Available profile classes CA anonymization"""
    return [CaDbAnonymizerProfile, CaDbAnonymizerDocumentParsingProfile]

default_schema

default_schema()

Default database schema CA anonymization

Source code in components/ca/public/helpers/anonymization.py

def default_schema(self) -> str:
    """Default database schema CA anonymization"""
    return "ca"

front_end

FRONT_END_PATHS module-attribute

FRONT_END_PATHS = dict(
    APP_URL="/dashboard",
    UNSUBSCRIBE_URL="/unsubscribe",
    LOGIN_URL="/login",
    PASSWORD_RESET_BASE_URL="/password_reset",
    DEPENDENT_INVITE_URL="/password_creation",
    MARMOT_URL="/marmot",
    MARMOT_CLAIM_MANAGEMENT="/marmot/claim_management",
    MARMOT_ACCOUNT_URL="/TODO-TODO-TODO",
    MARMOT_USER_URL="/marmot/user/",
    MARMOT_COMPANY_URL="/marmot/company/",
    EMPLOYEE_SIGNUP_URL="/ca/employee-onboarding",
    EMPLOYEE_DEEPLINK_URL="/employee_onboarding",
    EXEMPTION_ONBOARDING_URL="/exemption_onboarding",
    COMPANY_ADMIN_ONBOARDING_INVITE_URL="/onboarding/company_admin_signup",
    CUSTOMER_ADMIN_ONBOARDING_INVITE_URL="/onboarding/customer-admin-onboarding",
    PARTNER_ONBOARDING_INVITE_URL="/ca/partner-onboarding",
    PARTNER_DEEPLINK_URL="/partner_onboarding",
)

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

Source code in components/ca/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".
    """
    import sys

    from shared.helpers.env import 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
    if "-k" in sys.argv or "--use-kay-data" in sys.argv or "--use-kay" in sys.argv:
        return

    from components.ca.internal.offer.actions.create_default_entities import (
        create_default_entities,
    )

    create_default_entities()

load_all_mailers

load_all_mailers

load_all_mailers()

Loads all CA mailers, this helps setup register_sample_template_args for the admin tools

Source code in components/ca/public/helpers/load_all_mailers.py

def load_all_mailers() -> list:  # type: ignore[type-arg]
    """
    Loads all CA mailers, this helps setup register_sample_template_args for the admin tools
    """
    from components.ca.internal.claim_management.mailers import (
        reimbursement_request_success_email,
    )
    from components.ca.internal.mailers.exemption_approved_email import (
        send_exemption_approved_email,
    )
    from components.ca.internal.mailers.terminate_coverage_email import (
        send_terminate_coverage_email,
    )

    return [
        reimbursement_request_success_email,
        send_exemption_approved_email,
        send_terminate_coverage_email,
    ]

templating

configure_templating

configure_templating(jinja_env)

Inspired from https://github.com/alan-eu/alan-apps/blob/main/backend/components/fr/internal/helpers/templating.py#L46 ⧉

Source code in components/ca/public/helpers/templating.py

def configure_templating(jinja_env) -> None:  # type: ignore[no-untyped-def]
    """
    Inspired from https://github.com/alan-eu/alan-apps/blob/main/backend/components/fr/internal/helpers/templating.py#L46
    """
    jinja_env.filters.update(
        {
            "french_date": french_date_format,
            "dutch_date": dutch_date_format,
            "amount_with_currency": amount_with_currency,
        }
    )

components.ca.public.scim_api

adapter

CaScimAdapter

CaScimAdapter()

Bases: GenericScimAdapter

SCIM adapter for ca_api.

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

def __init__(self) -> None:
    super().__init__()
    self.profile_service = ProfileService.create(app_name=AppName.ALAN_CA)

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/ca/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_id = self.profile_service.create_profile(
        first_name=first_name, last_name=last_name
    )
    user = CaUser(
        profile_id=profile_id,
    )
    current_session.add(user)
    current_session.flush()
    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/ca/public/scim_api/adapter.py

@override
def get_scim_users_data(
    self,
    alan_employees: list[CaAlanEmployee],  # 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.profile_service.get_profiles(
        profile_ids={
            alan_employee.user.profile_id for alan_employee in alan_employees
        }
    )
    user_profiles_dict = {
        user_profile.id: user_profile for user_profile in user_profiles
    }

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

get_user_data

get_user_data(user_id)

Returns user's first and last name by user_id.

Source code in components/ca/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 = UserBroker.get_by_id(user_id).one_or_none()
    if user is None:
        raise BaseErrorCode.missing_resource(f"User with ID {user_id} not found")
    user_profile = self.profile_service.get_or_raise_profile(
        profile_id=user.profile_id
    )

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

profile_service instance-attribute

profile_service = create(app_name=ALAN_CA)

test

test_adapter

adapter

adapter()

Fixture for the EsGenericScimAdapter instance.

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

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

profile_service

profile_service()

Fixture for the profile service.

Source code in components/ca/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_CA)

test_create_app_user

test_create_app_user(adapter, profile_service)

Test create_app_user creates a new user correctly.

Source code in components/ca/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 = UserBroker.get_by_id(user_id).one_or_none()

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

Test get_scim_users_data returns correct mapping of user data.

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

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

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

Test get_user_data returns correct user identity.

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

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

    # 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.ca.public.test_data_generator

config

get_test_data_generation_config

get_test_data_generation_config()

Get the test data generation config

Source code in components/ca/public/test_data_generator/config.py

def get_test_data_generation_config() -> TestDataGeneratorConfig:
    """
    Get the test data generation config
    """
    return TestDataGeneratorConfig(
        patched_factories=patched_factories,
        handlers=get_fixture_handlers(),
        async_queue=None,
    )