Skip to content

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 
49
50
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 
24
25
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 
15
16
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 
33
34
35
36
37
38
39
40
41
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 
25
26
27
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 
115
116
117
118
119
120
@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 
89
90
91
92
93
94
95
96
@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 
50
51
52
53
54
55
56
57
58
@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 
44
45
46
47
@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 
33
34
35
36
37
38
39
40
41
@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 
123
124
125
126
127
128
@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 
 99
100
101
102
103
104
@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 
71
72
73
74
75
76
@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 
229
230
231
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 
173
174
175
176
177
178
179
180
181
182
@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 
141
142
143
144
145
146
147
148
149
150
@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 
163
164
165
166
167
168
169
170
@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 
153
154
155
156
157
158
159
160
@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_settlement 

marmot_create_settlement()

Marmot admin tools: create settlement

Source code in components/ca/public/blueprints/admin_tools.py 
197
198
199
200
201
202
203
204
205
206
@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 
131
132
133
134
135
136
137
138
@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 
185
186
187
188
189
190
191
192
193
194
@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_impersonate 

marmot_impersonate()

Marmot admin tools: create settlement

Source code in components/ca/public/blueprints/admin_tools.py 
219
220
221
222
223
224
225
226
@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_onboarding_invitation 

marmot_onboarding_invitation()

Marmot admin tools: onboarding invitation

Source code in components/ca/public/blueprints/admin_tools.py 
209
210
211
212
213
214
215
216
@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()

show_command 

show_command(command_full_name)

Default admin tools: see shared admin tools

Source code in components/ca/public/blueprints/admin_tools.py 
107
108
109
110
111
112
@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 
61
62
63
64
65
66
67
68
@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 
79
80
81
82
83
84
85
86
@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 
13
14
15
@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 
 9
10
11
12
@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.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 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
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.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 
40
41
42
43
44
45
46
47
48
49
@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 
51
52
53
54
55
56
57
@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 
27
28
29
30
31
32
33
34
35
36
37
38
@override
def get_company_information(
    self,
    company_id: str,
) -> CompanyInformation:
    from components.ca.internal.contracting.models.ca_company import CaCompany

    company = get_or_raise_missing_resource(CaCompany, 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/ca/public/employment/ca_country_gateway.py 
70
71
72
73
@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 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@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 
75
76
77
@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 
59
60
61
@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 
63
64
65
66
67
68
@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
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 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
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 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
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.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,
        ProfileEmailChanged,
    )
    from components.payment_gateway.public.enums import PaymentServiceProvider
    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,
    )

    # 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,
        predicate=lambda m: m.provider == PaymentServiceProvider.jpmorgan,
        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 
101
102
103
104
105
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 
21
22
23
24
25
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 
17
18
19
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 
27
28
29
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",
    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 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
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.terminate_coverage_email import (
        send_terminate_coverage_email,
    )
    from components.ca.internal.xodus.mailers.benefits_card_email import (
        send_benefits_card_email,
    )

    return [
        reimbursement_request_success_email,
        send_benefits_card_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 
 5
 6
 7
 8
 9
10
11
12
13
14
15
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 
20
21
22
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 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@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 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@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 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@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 
17
18
19
20
@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 
23
24
25
26
@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 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@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 
75
76
77
78
79
80
81
82
83
84
85
86
@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 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@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 
 8
 9
10
11
12
13
14
15
16
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,
    )