Skip to content

Api reference

components.global_services.public.blueprint

global_services_api_blueprint module-attribute 

global_services_api_blueprint = CustomBlueprint(
    "global_services_api_blueprint", __name__
)

components.global_services.public.controllers

api

create_api 

create_api(app_or_blueprint)
Source code in components/global_services/public/controllers/api.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
def create_api(app_or_blueprint) -> None:  # type: ignore[no-untyped-def]  # noqa: D103
    from shared.api.custom_api import CustomApi

    api = CustomApi(app_or_blueprint)

    from components.global_services.public.controllers.feature_flags import (
        feature_flags_endpoint,
    )
    from components.global_services.public.controllers.intercom import (
        intercom_endpoint,
    )
    from components.global_services.public.controllers.push_notification_token import (
        push_notification_token_endpoint,
    )

    api.add_endpoint(feature_flags_endpoint)
    api.add_endpoint(intercom_endpoint)
    api.add_endpoint(push_notification_token_endpoint)

feature_flags

FeatureFlagsController

Bases: BaseController

get 
get(user)
Source code in components/global_services/public/controllers/feature_flags.py 
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
@view_method(
    auth_strategy=GlobalAuthorizationStrategies().open(),
)
@inject_feature_user
@obs.api_call()
def get(self, user: Optional[FeatureUser]) -> Response:  # noqa: D102
    from flask import request

    from components.global_services.internal.queries.feature_flag import (
        get_feature_flags_cached,
    )

    device_id = (
        request.environ.get("HTTP_X_APP_DEVICE_ID", None) if request else None
    )  # TODO: not implemented yet on the client side (search for X-APP-DEVICE-ID)

    mobile_version = (
        request.environ.get("HTTP_X_APP_VERSION", None)
        if request and request.environ.get("HTTP_X_APP_PLATFORM", None) == "mobile"
        else None
    )

    return make_json_response(
        get_feature_flags_cached(
            feature_user=user,
            device_id=device_id,
            mobile_version=mobile_version,
        )
    )

feature_flags_endpoint module-attribute 

feature_flags_endpoint = Endpoint('feature_flags')

get_by_name 

get_by_name(name, user)
Source code in components/global_services/public/controllers/feature_flags.py 
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
@FeatureFlagsController.action_route(
    "/by_name/<string:name>",
    methods=["GET"],
    auth_strategy=GlobalAuthorizationStrategies().open(),
)
@inject_feature_user
@obs.api_call()
def get_by_name(name, user: Optional[FeatureUser]):  # type: ignore[no-untyped-def]  # noqa: D103
    from flask import request

    from components.global_services.internal.queries.feature_flag import (
        get_feature_flag_by_name_cached,
    )

    device_id = (
        request.environ.get("HTTP_X_APP_DEVICE_ID", None) if request else None
    )  # TODO: not implemented yet on the client side (search for X-APP-DEVICE-ID)

    mobile_version = (
        request.environ.get("HTTP_X_APP_VERSION", None)
        if request and request.environ.get("HTTP_X_APP_PLATFORM", None) == "mobile"
        else None
    )

    return make_json_response(
        get_feature_flag_by_name_cached(
            name=name,
            feature_user=user,
            device_id=device_id,
            mobile_version=mobile_version,
        )
    )

intercom

IntercomController

Bases: BaseController

Controller for Intercom related endpoints.

can_write classmethod 
can_write(user, user_id)

Prevent access to other users data.

Source code in components/global_services/public/controllers/intercom.py 
28
29
30
31
32
33
34
@classmethod
def can_write(cls, user: Authenticatable, user_id: UUID) -> bool:
    """Prevent access to other users data."""
    if not user or not user.id or user.id != user_id:
        return False

    return True

IntercomUserCustomAttributesController

Bases: BaseController

get 
get(user_id)

Get user data used by UCE on Intercom.

Source code in components/global_services/public/controllers/intercom.py 
76
77
78
79
80
81
82
83
84
85
86
@view_method(
    IntercomController.owner_only(param_name="user_id"),
)
@obs.api_call()
def get(self, user_id: UUID) -> Response:
    """Get user data used by UCE on Intercom."""
    user_custom_attributes = (
        get_app_dependency().get_intercom_user_custom_attributes(user_id=user_id)
    )

    return make_json_response(user_custom_attributes)

IntercomUserHashController

Bases: BaseController

get 
get(user_id, params)

Retrieve a signed hash to 'authenticate' user in Intercom.

Source code in components/global_services/public/controllers/intercom.py 
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
@view_method(IntercomController.owner_only(param_name="user_id"))
@request_argument(
    "platform",
    type=str,
    required=True,
    help="Platform of user",
)
@obs.api_call()
def get(self, user_id: UUID, params: dict[str, Any]) -> str:
    """Retrieve a signed hash to 'authenticate' user in Intercom."""
    platform = params["platform"]
    intercom_identity_verification_secrets_names = current_config.get(
        "INTERCOM_IDENTITY_VERIFICATION_SECRETS_NAMES"
    )

    if (
        not intercom_identity_verification_secrets_names
        or platform not in intercom_identity_verification_secrets_names
    ):
        abort(500, "INTERCOM_MISSING_SECRETS_NAMES_ERROR_CODE")

    intercom_identity_secret = raw_secret_from_config(
        config_key=intercom_identity_verification_secrets_names[platform],
    )

    if not intercom_identity_secret:
        abort(500, "INTERCOM_MISSING_SECRET_ERROR_CODE")

    return hmac.new(
        intercom_identity_secret.encode("utf8"),
        str(user_id).encode("utf8"),
        digestmod=hashlib.sha256,
    ).hexdigest()

intercom_endpoint module-attribute 

intercom_endpoint = Endpoint('intercom')

push_notification_token

PushNotificationTokenControllerCrud

Bases: BaseController

Controller for push notification (firebase) related endpoints.

delete 
delete(user, id)

Deletes the token

See components.global_services.internal.services.push_notifications.push_notification_token

Source code in components/global_services/public/controllers/push_notification_token.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@view_method(GlobalAuthorizationStrategies().authenticated())
@obs.api_call()
def delete(self, user: Authenticatable, id: UUID) -> Response:
    """
    Deletes the token

    See components.global_services.internal.services.push_notifications.push_notification_token
    """
    from components.global_services.internal.services.push_notifications.push_notification_token import (
        push_notification_token_logic,
    )

    if not push_notification_token_logic.can_write(str(user.id), id):
        abort(401, "User is not authorised to edit this token")

    push_notification_token_logic.delete_token(id)

    return make_success_json_response()
post 
post(user, params)

Creates or updates the token

See components.global_services.internal.services.push_notifications.push_notification_token

Source code in components/global_services/public/controllers/push_notification_token.py 
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
@view_method(GlobalAuthorizationStrategies().authenticated())
@request_argument(
    "registration_token",
    required=True,
    type=str,
    help="Registration token obtained from Firebase Cloud Messaging SDK",
)
@request_argument(
    "platform",
    type=str,
    choices=("ios", "android"),
    help="Platform: 'android' or 'ios'",
)
@request_argument(
    "has_permission",
    type=bool,
    help="The user has allowed push notifications",
)
@request_argument(
    "send_to_customer_io",
    type=bool,
    help="Whether the token should be forwarded to Customer.io. Set to False by recent apps as they handle this themselves.",
)
@obs.api_call()
def post(self, user: Authenticatable, params: dict) -> Response:  # type: ignore[type-arg]
    """
    Creates or updates the token

    See components.global_services.internal.services.push_notifications.push_notification_token
    """
    from components.global_services.internal.services.push_notifications.push_notification_token import (
        push_notification_token_logic,
    )

    token, is_new_token = push_notification_token_logic.create_or_update_token(
        user_id=str(user.id),
        registration_token=params["registration_token"],
        has_permission=params.get("has_permission"),
        platform=params.get("platform"),
        send_to_customer_io=params.get("send_to_customer_io", True),
    )

    token_entity = InMemoryPushNotificationToken.from_model(token)

    return make_json_response(
        token_entity.to_dict(), code=201 if is_new_token else 200
    )

push_notification_token_endpoint module-attribute 

push_notification_token_endpoint = Endpoint(
    "push_notification_tokens"
)

components.global_services.public.dependencies

COMPONENT_NAME module-attribute 

COMPONENT_NAME = 'global_services'

GlobalServicesDependency

Bases: ABC

get_intercom_user_custom_attributes abstractmethod 

get_intercom_user_custom_attributes(user_id)

Implement get_intercom_user_custom_attributes

All properties in the response should be defined on Intercom (see here ⧉).

Source code in components/global_services/public/dependencies.py 
11
12
13
14
15
16
17
18
@abstractmethod
def get_intercom_user_custom_attributes(self, user_id: UUID) -> dict[str, Any]:
    """
    Implement get_intercom_user_custom_attributes

    All properties in the response should be defined on Intercom (see [here](https://www.intercom.com/help/en/articles/179-create-and-track-custom-data-attributes-cdas)).
    """
    raise NotImplementedError("get_intercom_user_custom_attributes not implemented")

get_app_dependency 

get_app_dependency()
Source code in components/global_services/public/dependencies.py 
29
30
31
32
def get_app_dependency() -> GlobalServicesDependency:  # noqa: D103
    from flask import current_app

    return cast("CustomFlask", current_app).get_component_dependency(COMPONENT_NAME)  # type: ignore[no-any-return]

set_app_dependency 

set_app_dependency(dependency)
Source code in components/global_services/public/dependencies.py 
21
22
23
24
25
26
def set_app_dependency(dependency: GlobalServicesDependency) -> None:  # noqa: D103
    from flask import current_app

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

components.global_services.public.entities

feature_flag

FeatureFlagData dataclass 

FeatureFlagData(feature_metadata, name, is_active=False)

Bases: DataClassJsonMixin

feature_metadata instance-attribute 
feature_metadata
from_feature_flag_model classmethod 
from_feature_flag_model(feature_flag)
Source code in components/global_services/public/entities/feature_flag.py 
18
19
20
21
22
23
24
@classmethod
def from_feature_flag_model(cls, feature_flag: "FeatureFlag") -> "FeatureFlagData":  # noqa: D102
    return cls(
        feature_metadata=feature_flag.feature_metadata,
        name=feature_flag.name,
        is_active=feature_flag.is_active,  # type: ignore[arg-type]
    )
is_active class-attribute instance-attribute 
is_active = False
name instance-attribute 
name

in_memory_push_notification_log

InMemoryPushNotificationLog dataclass 

InMemoryPushNotificationLog(id, name, data, status)

Represents an in-memory version of a push notification log.

data instance-attribute 
data
from_model classmethod 
from_model(notification_log)

Creates an InMemoryPushNotificationLog instance from a BasePushNotificationLog model, it can be PushNotificationLog, BePushNotificationLog or EsPushNotificationLog depending on app_name.

Source code in components/global_services/public/entities/in_memory_push_notification_log.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@classmethod
def from_model(
    cls, notification_log: "BasePushNotificationLog"
) -> "InMemoryPushNotificationLog":
    """
    Creates an InMemoryPushNotificationLog instance from a BasePushNotificationLog model,
    it can be PushNotificationLog, BePushNotificationLog or EsPushNotificationLog depending on app_name.
    """
    return cls(
        id=notification_log.id,
        name=notification_log.name,
        data=notification_log.data,
        status=mandatory(notification_log.status),
    )
id instance-attribute 
id
name instance-attribute 
name
status instance-attribute 
status

in_memory_push_notification_token

InMemoryPushNotificationToken dataclass 

InMemoryPushNotificationToken(registration_token, platform)

Bases: DataClassJsonMixin

Represents an in-memory simplified version of a push notification model

from_model classmethod 
from_model(push_notification_token)

Creates an InMemoryPushNotificationToken from a BasePushNotificationToken model

Source code in components/global_services/public/entities/in_memory_push_notification_token.py 
19
20
21
22
23
24
25
26
27
28
29
@classmethod
def from_model(
    cls, push_notification_token: "BasePushNotificationToken"
) -> "InMemoryPushNotificationToken":
    """
    Creates an InMemoryPushNotificationToken from a BasePushNotificationToken model
    """
    return cls(
        registration_token=push_notification_token.registration_token,
        platform=push_notification_token.platform,
    )
platform instance-attribute 
platform
registration_token instance-attribute 
registration_token

mfa

MfaRequiredSetting dataclass 

MfaRequiredSetting(id, restriction_type)

Bases: DataClassJsonMixin

Define the structure of the MFA required setting

id instance-attribute 
id
restriction_type instance-attribute 
restriction_type

MfaRequiredSettingType

Bases: AlanBaseEnum

Define the list of possible values for MFA restriction

admins_only class-attribute instance-attribute 
admins_only = 'admins_only'
all class-attribute instance-attribute 
all = 'all'
employees_only class-attribute instance-attribute 
employees_only = 'employees_only'

MfaRequiredSettings dataclass 

MfaRequiredSettings(companies=list(), accounts=list())

Bases: DataClassJsonMixin

Define the type of MFA restriction we can apply (accounts is not implemented yet)

accounts class-attribute instance-attribute 
accounts = field(default_factory=list)
companies class-attribute instance-attribute 
companies = field(default_factory=list)

components.global_services.public.enums

base_push_notification_name

BasePushNotificationName

Bases: AlanBaseEnum

Abstract PushNotificationName to keep track of local places where the global push notification system is used

components.global_services.public.queries

feature_flag

get_feature_flag_by_name_cached 

get_feature_flag_by_name_cached(
    name,
    feature_user=None,
    device_id=None,
    mobile_version=None,
)

DEPRECATED: Use LaunchDarkly instead

Source code in components/global_services/internal/queries/feature_flag.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@deprecated("Use LaunchDarkly instead", category=AlanDeprecationWarning)
@cached_for(minutes=CACHE_DURATION_MINUTES, unless=is_caching_disabled)
def get_feature_flag_by_name_cached(
    name: str,
    feature_user: Optional[FeatureUser] = None,
    device_id: Optional[str] = None,
    mobile_version: Optional[str] = None,
) -> Optional[FeatureFlagData]:
    """
    DEPRECATED: Use LaunchDarkly instead
    """
    return get_feature_flag_by_name(
        name=name,
        feature_user=feature_user,
        device_id=device_id,
        mobile_version=mobile_version,
    )

mfa

get_companies_with_mfa_required_settings 

get_companies_with_mfa_required_settings(app_name)

This function retrieves companies with MFA required settings in FeatureFlag enabled.

Source code in components/global_services/public/queries/mfa.py 
42
43
44
45
46
47
48
49
50
def get_companies_with_mfa_required_settings(
    app_name: AppName,
) -> list[MfaRequiredSetting]:
    """This function retrieves companies with MFA required settings in FeatureFlag enabled."""
    mfa_required_settings = get_mfa_required_settings_by_app(app_name=app_name)
    if not mfa_required_settings:
        return []

    return mfa_required_settings.companies

get_mfa_required_settings_by_app 

get_mfa_required_settings_by_app(app_name)

Get the MFA required settings for the given country.

Source code in components/global_services/public/queries/mfa.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def get_mfa_required_settings_by_app(
    app_name: AppName,
) -> Optional[MfaRequiredSettings]:
    """
    Get the MFA required settings for the given country.
    """
    feature_flag = get_feature_flag_by_name_cached(
        name=f"{app_name}-mfa_required_settings"
    )
    if (
        not feature_flag
        or not feature_flag.is_active
        or not feature_flag.feature_metadata
    ):
        return None

    try:
        return MfaRequiredSettings.from_dict(feature_flag.feature_metadata)
    except Exception as e:
        current_logger.error(
            f"Error while parsing MFA required settings: {e}."
            f"Feature flag: {feature_flag}"
        )
        return None

get_mfa_restriction_type_for_company 

get_mfa_restriction_type_for_company(app_name, company_id)

This function retrieves the MFA restriction type for a specific company.

Source code in components/global_services/public/queries/mfa.py 
53
54
55
56
57
58
59
60
61
62
def get_mfa_restriction_type_for_company(
    app_name: AppName, company_id: uuid.UUID | int
) -> Optional[MfaRequiredSettingType]:
    """This function retrieves the MFA restriction type for a specific company."""
    mfa_required_companies = get_companies_with_mfa_required_settings(app_name=app_name)
    for company in mfa_required_companies:
        if str(company.id) == str(company_id):
            return company.restriction_type

    return None

is_mfa_required_for_company_admins 

is_mfa_required_for_company_admins(app_name, company_id)

This function checks if a company forced MFA activation for its admins.

Source code in components/global_services/public/queries/mfa.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def is_mfa_required_for_company_admins(
    app_name: AppName, company_id: uuid.UUID | int
) -> bool:
    """This function checks if a company forced MFA activation for its admins."""
    restriction_type = get_mfa_restriction_type_for_company(
        app_name=app_name, company_id=company_id
    )
    if not restriction_type or restriction_type not in [
        MfaRequiredSettingType.all,
        MfaRequiredSettingType.admins_only,
    ]:
        return False

    return True

is_mfa_required_for_company_employees 

is_mfa_required_for_company_employees(app_name, company_id)

This function checks if a company forced MFA activation for its employees.

Source code in components/global_services/public/queries/mfa.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def is_mfa_required_for_company_employees(
    app_name: AppName, company_id: uuid.UUID | int
) -> bool:
    """This function checks if a company forced MFA activation for its employees."""
    restriction_type = get_mfa_restriction_type_for_company(
        app_name=app_name, company_id=company_id
    )
    if not restriction_type or restriction_type not in [
        MfaRequiredSettingType.all,
        MfaRequiredSettingType.employees_only,
    ]:
        return False

    return True

push_notifications

P module-attribute 

P = ParamSpec('P')

get_push_notification_tokens_for_user 

get_push_notification_tokens_for_user(user_id)

Returns the push notification tokens for the user

Source code in components/global_services/public/queries/push_notifications.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def get_push_notification_tokens_for_user(
    user_id: uuid.UUID,
) -> list[str]:
    """
    Returns the push notification tokens for the user
    """
    push_notification_tokens: list[GlobalPushNotificationToken] = (
        current_session.query(GlobalPushNotificationToken)  # noqa: ALN085
        .filter(
            GlobalPushNotificationToken.has_permission.is_(True),
            GlobalPushNotificationToken.user_id == str(user_id),
        )
        .all()
    )

    return [
        push_notification_token.registration_token
        for push_notification_token in push_notification_tokens
    ]

components.global_services.public.services

push_notifications

get_push_notification_functions 

get_push_notification_functions()

Get the push notification functions for the current app.

Hook the get_tokens global function so that consumers don't have to care about it.

push_notification_sender_async, push_notification_sender_sync = get_push_notification_functions()

Source code in components/global_services/internal/services/push_notifications/push_notifications_local.py 
 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
 99
100
def get_push_notification_functions() -> tuple[
    Callable[[Callable[P, GlobalPushNotificationParams | None]], Callable[P, None]],
    Callable[[Callable[P, GlobalPushNotificationParams | None]], Callable[P, None]],
]:
    """
    Get the push notification functions for the current app.

    Hook the get_tokens global function so that consumers don't have to care about it.

    >>> push_notification_sender_async, push_notification_sender_sync = get_push_notification_functions()
    """
    match get_current_app_name():
        case AppName.ALAN_FR:
            from components.fr.public.services.push_notifications import (
                push_notification_sender_async,
                push_notification_sender_sync,
            )

        case AppName.ALAN_BE:
            from components.be.public.services.push_notifications import (
                push_notification_sender_async,
                push_notification_sender_sync,
            )

        case AppName.ALAN_ES:
            from components.es.public.services.push_notifications import (
                push_notification_sender_async,
                push_notification_sender_sync,
            )

        case _:
            # CA: handled in the global code
            raise NotImplementedError(
                f"Push notifications not implemented for {get_current_app_name()}"
            )

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

            if pn_params is None:
                return None

            return PushNotificationParams(
                **asdict(pn_params),
                tokens=get_push_notification_tokens_for_user(
                    pn_params.app_user_id, AppName(pn_params.app_id)
                ),
            )

        return push_notification_sender_async(decorated_function)

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

            if pn_params is None:
                return None

            return PushNotificationParams(
                **asdict(pn_params),
                tokens=get_push_notification_tokens_for_user(
                    pn_params.app_user_id, AppName(pn_params.app_id)
                ),
            )

        return push_notification_sender_sync(decorated_function)

    return (_push_notification_sender_async, _push_notification_sender_sync)

get_push_notification_logs_for_user 

get_push_notification_logs_for_user(
    app_name,
    app_user_id,
    notification_names,
    created_at__gte=None,
)
Source code in components/global_services/internal/services/push_notifications/push_notifications_local.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def get_push_notification_logs_for_user(
    app_name: AppName,
    app_user_id: str,
    notification_names: list[BasePushNotificationName],
    created_at__gte: Optional[datetime] = None,
) -> list[InMemoryPushNotificationLog]:
    from components.global_services.internal.queries.push_notification_logs import (
        get_push_notifications,
    )

    return get_push_notifications(
        app_name=app_name,
        app_user_id=app_user_id,
        notification_names=notification_names,
        created_at__gte=created_at__gte,
    )

get_push_notification_token_objects_for_user 

get_push_notification_token_objects_for_user(
    user_app_id, app_name
)
Source code in components/global_services/internal/services/push_notifications/push_notifications_local.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def get_push_notification_token_objects_for_user(
    user_app_id: str, app_name: AppName
) -> list[InMemoryPushNotificationToken]:
    match app_name:
        case AppName.ALAN_FR:
            from components.fr.internal.business_logic.push_notification_token import (  # noqa: ALN043
                get_push_notification_token_objects_for_user as get_push_notification_token_objects_for_user_fr,
            )

            return get_push_notification_token_objects_for_user_fr(int(user_app_id))

        case _:
            raise NotImplementedError(
                f"Push notifications not implemented for {app_name}"
            )

get_push_notification_tokens_for_user 

get_push_notification_tokens_for_user(
    user_app_id, app_name
)
Source code in components/global_services/internal/services/push_notifications/push_notifications_local.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def get_push_notification_tokens_for_user(
    user_app_id: str, app_name: AppName
) -> list[str]:
    match app_name:
        case AppName.ALAN_FR:
            from components.fr.internal.business_logic.push_notification_token import (  # noqa: ALN043
                get_push_notification_tokens_for_user as get_push_notification_tokens_for_user_fr,
            )

            return get_push_notification_tokens_for_user_fr(int(user_app_id))

        case AppName.ALAN_BE:
            from components.be.public.services.push_notifications import (
                get_push_notification_tokens_for_user as get_push_notification_tokens_for_user_be,
            )

            return get_push_notification_tokens_for_user_be(UUID(user_app_id))

        case AppName.ALAN_ES:
            from components.es.public.services.push_notifications import (
                get_push_notification_tokens_for_user as get_push_notification_tokens_for_user_es,
            )

            return get_push_notification_tokens_for_user_es(UUID(user_app_id))

        case _:
            # CA: handled in the global code
            raise NotImplementedError(
                f"Push notifications not implemented for {app_name}"
            )

push_notification_logic module-attribute 

push_notification_logic = SharedPushNotificationLogic(
    push_notification_log_cls=GlobalPushNotificationLog
)