Skip to content

Api reference

components.alan_pay.public.business_logic

alan_pay_actions

AlanPayActions

Public actions interface with the Alan Pay component.

declare_profile staticmethod
declare_profile(app_id, app_user_id, save=True)

Create a new Alan Pay profile for the given user, and return its ID.

Source code in components/alan_pay/public/business_logic/alan_pay_actions.py
@staticmethod
def declare_profile(app_id: str, app_user_id: str, save: bool = True) -> UUID:
    """
    Create a new Alan Pay profile for the given user, and return its ID.
    """
    with no_commit_in_session(
        commit_at_end=save, silent=True, rollback_at_end=False
    ):
        # Create the Alan Pay profile
        profile = save_alan_pay_profile(
            app_id=app_id,
            app_user_id=app_user_id,
            save=False,
        )
        # Flush to enforce the profile ID creation
        current_session.flush()

        # Also create the related transfer history, used to link all card payments and bank transfers to the profile
        save_transfer_history(
            profile_id=profile.id,
            save=False,
        )

    return profile.id
declare_sepa_mandate classmethod
declare_sepa_mandate(
    app_id,
    app_user_id,
    debtor_name,
    debtor_iban,
    debtor_country,
    unique_mandate_reference,
    signature_date,
    save=True,
)

Declare a SEPA mandate for the given user.

Source code in components/alan_pay/public/business_logic/alan_pay_actions.py
@classmethod
def declare_sepa_mandate(
    cls,
    app_id: str,
    app_user_id: str,
    debtor_name: str,
    debtor_iban: str,
    debtor_country: str,
    unique_mandate_reference: str,
    signature_date: date,
    save: bool = True,
) -> None:
    """
    Declare a SEPA mandate for the given user.
    """
    # Fetch target profile
    profile = get_alan_pay_profile(app_id=app_id, app_user_id=app_user_id)
    if profile is None:
        raise ValueError(
            f"Profile not found for app [{app_id}] and user [{app_user_id}]"
        )

    # Map to CCA33
    country_code = cls._get_country_code_from_string(country=debtor_country)

    # Run action
    declare_sepa_mandate(
        alan_pay_profile_id=profile.id,
        debtor_name=debtor_name,
        debtor_iban=debtor_iban,
        debtor_country=country_code,
        unique_mandate_reference=unique_mandate_reference,
        signature_date=signature_date,
        save=save,
    )
offboard_user staticmethod
offboard_user(app_id, app_user_id)

Offboard an existing user from Alan Pay.

Source code in components/alan_pay/public/business_logic/alan_pay_actions.py
@staticmethod
def offboard_user(
    app_id: str,
    app_user_id: str,
) -> None:
    """
    Offboard an existing user from Alan Pay.
    """
    # Fetch target profile
    profile = get_alan_pay_profile(app_id=app_id, app_user_id=app_user_id)
    if profile is None:
        raise ValueError(
            f"Profile not found for app [{app_id}] and user [{app_user_id}]"
        )

    offboard_alan_pay_profile(profile_id=profile.id)
onboard_user staticmethod
onboard_user(
    app_id,
    app_user_id,
    phone_number_verification_request_id,
    phone_number_otp_code,
    first_name,
    last_name,
    birth_date,
    email,
    lang,
)

Onboard a new user on Alan Pay by: - Verifying its phone number using the OTP (One Time Password) they received - Creating a new account membership and a card

Raises an InvalidPhoneNumberVerificationCodeException if the OTP is invalid

Source code in components/alan_pay/public/business_logic/alan_pay_actions.py
@staticmethod
def onboard_user(
    app_id: str,
    app_user_id: str,
    phone_number_verification_request_id: str,
    phone_number_otp_code: str,
    first_name: str,
    last_name: str,
    birth_date: date,
    email: str,
    lang: Lang,
) -> None:
    """
    Onboard a new user on Alan Pay by:
    - Verifying its phone number using the OTP (One Time Password) they received
    - Creating a new account membership and a card

    Raises an InvalidPhoneNumberVerificationCodeException if the OTP is invalid
    """
    # Fetch target profile
    profile = get_alan_pay_profile(app_id=app_id, app_user_id=app_user_id)
    if profile is None:
        raise ValueError(
            f"Profile not found for app [{app_id}] and user [{app_user_id}]"
        )

    onboard_user_from_phone_number_verification(
        profile_id=profile.id,
        request_id=phone_number_verification_request_id,
        code=phone_number_otp_code,
        first_name=first_name,
        last_name=last_name,
        birth_date=birth_date,
        email=email,
        lang=lang,
    )
request_external_transaction staticmethod
request_external_transaction(
    app_id,
    app_user_id,
    transaction_date,
    requested_amount_in_cents,
    external_transaction_id,
    save=True,
)

Request creation of a new external transaction for the given user. - If user has no debt to Alan (ex: no recent card payment), no transaction will be created - If user has a debt to Alan, a new transaction will be created, whose amount will be the min value between the user debt and requested_amount_in_cents

Source code in components/alan_pay/public/business_logic/alan_pay_actions.py
@staticmethod
def request_external_transaction(
    app_id: str,
    app_user_id: str,
    transaction_date: datetime,
    requested_amount_in_cents: int,
    external_transaction_id: str,
    save: bool = True,
) -> InMemoryAlanBalanceTransaction | None:
    """
    Request creation of a new external transaction for the given user.
    - If user has no debt to Alan (ex: no recent card payment), no transaction will be created
    - If user has a debt to Alan, a new transaction will be created, whose amount will be the min value between the user debt and `requested_amount_in_cents`
    """
    # Do NOT create any new transaction if we don't know anything about the user
    profile = get_alan_pay_profile(app_id=app_id, app_user_id=app_user_id)
    if not profile:
        return None

    # Request an addition to the Alan Balance
    transaction = request_external_transaction(
        profile_id=profile.id,
        requested_amount_in_cents=requested_amount_in_cents,
        external_transaction_id=external_transaction_id,
        transaction_date=transaction_date,
        save=save,
    )

    current_session.flush()  # Flush to enforce the transaction ID creation used in the in-memory mapping

    return (
        map_alan_balance_transaction(transaction=transaction)
        if transaction is not None
        else None
    )
set_push_notification_date_for_external_transactions staticmethod
set_push_notification_date_for_external_transactions(
    transaction_ids, push_notification_sent_at
)

Update the date a push notification was sent for the given external transactions.

Source code in components/alan_pay/public/business_logic/alan_pay_actions.py
@staticmethod
def set_push_notification_date_for_external_transactions(
    transaction_ids: list[UUID],
    push_notification_sent_at: datetime,
) -> list[InMemoryAlanBalanceTransaction]:
    """
    Update the date a push notification was sent for the given external transactions.
    """
    transactions = set_push_notification_date_for_external_transactions(
        transaction_ids=transaction_ids,
        push_notification_sent_at=push_notification_sent_at,
    )

    return [
        map_alan_balance_transaction(transaction=transaction)
        for transaction in transactions
    ]
start_phone_number_verification staticmethod
start_phone_number_verification(phone_number, lang)

Start phone number verification by sending a OTP (One Time Password) to the given phone number.

Source code in components/alan_pay/public/business_logic/alan_pay_actions.py
@staticmethod
def start_phone_number_verification(
    phone_number: str,
    lang: Lang,
) -> InMemoryAlanPayPhoneNumberVerification:
    """
    Start phone number verification by sending a OTP (One Time Password) to the given phone number.
    """
    return start_alan_pay_phone_number_verification(
        phone_number=phone_number, lang=lang
    )
view_card_numbers staticmethod
view_card_numbers(app_id, app_user_id)

Returns the numbers of the latest active card of the given user.

Source code in components/alan_pay/public/business_logic/alan_pay_actions.py
@staticmethod
def view_card_numbers(
    app_id: str,
    app_user_id: str,
) -> InMemoryAlanPayCardNumbers:
    """
    Returns the numbers of the latest active card of the given user.
    """
    from components.alan_pay.internal.business_logic.actions.alan_pay_card import (
        get_alan_pay_card_numbers,
    )

    # Fetch target profile
    profile = get_alan_pay_profile(app_id=app_id, app_user_id=app_user_id)
    if profile is None:
        raise ValueError(
            f"Profile not found for app [{app_id}] and user [{app_user_id}]"
        )

    return get_alan_pay_card_numbers(profile_id=profile.id)

alan_pay_queries

AlanPayQueries

Public queries interface with the Alan Pay component.

get_alan_balance staticmethod
get_alan_balance(app_id, app_user_id)

Return the live amount of the Alan Balance for the given user.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_alan_balance(app_id: str, app_user_id: str) -> int:
    """
    Return the live amount of the Alan Balance for the given user.
    """
    profile = get_alan_pay_profile(app_id=app_id, app_user_id=app_user_id)

    return get_alan_balance(profile_id=profile.id) if profile is not None else 0
get_alan_balance_transactions staticmethod
get_alan_balance_transactions(
    app_id, app_user_id, transaction_ids=None
)

Return a list of Alan Balance transactions for the given user.

If transaction_ids is not None, only the transactions with the given ids will be returned.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_alan_balance_transactions(
    app_id: str,
    app_user_id: str,
    transaction_ids: Optional[list[UUID]] = None,
) -> list[InMemoryAlanBalanceTransaction]:
    """
    Return a list of Alan Balance transactions for the given user.

    If transaction_ids is not None, only the transactions with the given ids will be returned.
    """
    profile = get_alan_pay_profile(app_id=app_id, app_user_id=app_user_id)
    transactions = (
        get_alan_balance_transactions(
            profile_id=profile.id, transaction_ids=transaction_ids
        )
        if profile is not None
        else []
    )

    return [
        map_alan_balance_transaction(transaction=transaction)
        for transaction in transactions
    ]
get_alan_balance_transactions_by_external_id staticmethod
get_alan_balance_transactions_by_external_id(
    external_transaction_ids,
)

Return a list of Alan Balance transactions whose external ids are from the given ones.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_alan_balance_transactions_by_external_id(
    external_transaction_ids: list[str],
) -> list[InMemoryAlanBalanceTransaction]:
    """
    Return a list of Alan Balance transactions whose external ids are from the given ones.
    """
    return [
        map_alan_balance_transaction(transaction=transaction)
        for transaction in get_alan_balance_transactions(
            external_transaction_ids=external_transaction_ids
        )
    ]
get_latest_payments staticmethod
get_latest_payments(app_id, app_user_id, limit=10)

Returns a list of the latest payments made by the given Alan Pay profile.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_latest_payments(
    app_id: str, app_user_id: str, limit: int = 10
) -> list[InMemoryAlanPayCardPayment]:
    """
    Returns a list of the latest payments made by the given Alan Pay profile.
    """
    profile = get_alan_pay_profile(app_id=app_id, app_user_id=app_user_id)
    if profile is None:
        return []

    return get_latest_card_payments_for_profile(profile_id=profile.id, limit=limit)
get_merchant_category_code_descriptions staticmethod
get_merchant_category_code_descriptions(lang=Lang.english)

Returns description labels for merchant category codes.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_merchant_category_code_descriptions(
    lang: Lang = Lang.english,
) -> dict[str, str]:
    """
    Returns description labels for merchant category codes.
    """
    return get_merchant_category_code_descriptions(lang=lang)
get_next_recovery_debt classmethod
get_next_recovery_debt(app_id, app_user_id)

Returns the next recovery debt for the given user.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@classmethod
def get_next_recovery_debt(
    cls, app_id: str, app_user_id: str
) -> Optional[InMemoryAlanPayRecoveryDebt]:
    """
    Returns the next recovery debt for the given user.
    """
    next_recovery_session = cls.get_next_recovery_session()

    return (
        cls.get_recovery_debt(
            recovery_session_id=next_recovery_session.id,
            app_id=app_id,
            app_user_id=app_user_id,
        )
        if next_recovery_session is not None
        else None
    )
get_next_recovery_session staticmethod
get_next_recovery_session()

Returns the closest Alan Pay recovery session (can be today).

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_next_recovery_session() -> Optional[InMemoryAlanPayRecoverySession]:
    """
    Returns the closest Alan Pay recovery session (can be today).
    """
    session = get_next_alan_pay_recovery_session(ref_date=utctoday())

    return (
        map_alan_pay_recovery_session(recovery_session=session)
        if session is not None
        else None
    )
get_profile staticmethod
get_profile(app_id, app_user_id)

Returns the Alan Pay profile for the given user info, or null if not found.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_profile(app_id: str, app_user_id: str) -> Optional[InMemoryAlanPayProfile]:
    """
    Returns the Alan Pay profile for the given user info, or null if not found.
    """
    profile = get_alan_pay_profile(app_id=app_id, app_user_id=app_user_id)

    return map_alan_pay_profile(profile=profile) if profile is not None else None
get_profile_by_id staticmethod
get_profile_by_id(profile_id)

Returns a specific Alan Pay profile.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_profile_by_id(profile_id: UUID) -> InMemoryAlanPayProfile:
    """
    Returns a specific Alan Pay profile.
    """
    profile = get_alan_pay_profile_by_id(profile_id=profile_id)

    return map_alan_pay_profile(profile=profile)
get_profiles staticmethod
get_profiles(app_id)

Returns all the existing Alan Pay profiles for the given app.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_profiles(app_id: str) -> list[InMemoryAlanPayProfile]:
    """
    Returns all the existing Alan Pay profiles for the given app.
    """
    query = current_session.query(AlanPaymentCardProfile).filter(  # noqa: ALN085
        AlanPaymentCardProfile.app_id == app_id
    )

    return [map_alan_pay_profile(profile=profile) for profile in query]
get_recovery_debt staticmethod
get_recovery_debt(recovery_session_id, app_id, app_user_id)

Return debt info for the given recovery session and user.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_recovery_debt(
    recovery_session_id: UUID,
    app_id: str,
    app_user_id: str,
) -> Optional[InMemoryAlanPayRecoveryDebt]:
    """
    Return debt info for the given recovery session and user.
    """
    profile = get_alan_pay_profile(app_id=app_id, app_user_id=app_user_id)
    if profile is None:
        return None

    # Find the recovery debt linked to the given recovery session
    for debt in get_alan_pay_recovery_debts(profile_id=profile.id):
        if debt.alan_payment_card_recovery_session_id == recovery_session_id:
            return map_alan_pay_recovery_debt(recovery_debt=debt)

    return None
get_recovery_debt_by_id staticmethod
get_recovery_debt_by_id(recovery_debt_id)

Return info for a specific debt.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_recovery_debt_by_id(recovery_debt_id: UUID) -> InMemoryAlanPayRecoveryDebt:
    """
    Return info for a specific debt.
    """
    debt = get_alan_pay_recovery_debt_by_id(recovery_debt_id=recovery_debt_id)

    return map_alan_pay_recovery_debt(recovery_debt=debt)
get_recovery_debts staticmethod
get_recovery_debts(recovery_session_id)
Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_recovery_debts(  # noqa: D102
    recovery_session_id: UUID,
) -> list[InMemoryAlanPayRecoveryDebt]:
    recovery_session = get_or_raise_missing_resource(
        AlanPaymentCardRecoverySession,
        recovery_session_id,
        options=[
            joinedload(AlanPaymentCardRecoverySession.debts).joinedload(
                AlanPaymentCardRecoveryDebt.recovered_transactions
            )
        ],
    )

    return [
        map_alan_pay_recovery_debt(recovery_debt=recovery_debt)
        for recovery_debt in recovery_session.debts
    ]
get_recovery_debts_for_user staticmethod
get_recovery_debts_for_user(app_id, app_user_id)

Return debt info for the given recovery session and user.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def get_recovery_debts_for_user(
    app_id: str,
    app_user_id: str,
) -> list[InMemoryAlanPayRecoveryDebt]:
    """
    Return debt info for the given recovery session and user.
    """
    profile = get_alan_pay_profile(app_id=app_id, app_user_id=app_user_id)
    if profile is None:
        return []

    return [
        map_alan_pay_recovery_debt(recovery_debt=debt)
        for debt in get_alan_pay_recovery_debts(profile_id=profile.id)
    ]
has_card_provisioning staticmethod
has_card_provisioning(profile_id)

Returns True if the alan pay profile has a provision for its card

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def has_card_provisioning(profile_id: UUID) -> bool:
    """
    Returns True if the alan pay profile has a provision for its card
    """
    return has_card_provisioning(profile_id=profile_id)
is_support_appointment_scheduling_enabled staticmethod
is_support_appointment_scheduling_enabled()

Returns true if members are allowed to schedule appointments with a member of the team (though Google Calendar), false otherwise.

Source code in components/alan_pay/public/business_logic/alan_pay_queries.py
@staticmethod
def is_support_appointment_scheduling_enabled() -> bool:
    """
    Returns true if members are allowed to schedule appointments with a member of the
    team (though Google Calendar), false otherwise.
    """
    return is_alan_pay_appointment_scheduling_enabled()

reset_local_onboarding_flow_action

reset_swan_and_alan_pay_dev_environment_for_user

reset_swan_and_alan_pay_dev_environment_for_user(
    app_user_id, app_user_email, save
)
Source code in components/alan_pay/public/business_logic/reset_local_onboarding_flow_action.py
def reset_swan_and_alan_pay_dev_environment_for_user(  # noqa: D103
    app_user_id: int,
    app_user_email: str | None,
    save: bool,
) -> None:
    if is_production_mode():
        raise RuntimeError(
            "Unauthorized - This method is only be used for development purposes"
        )

    client = SwanClient()
    account_memberships = client.queries.account_memberships()

    ## Start by removing all account memberships for a user, this allows to use the same phonenumber again in the onboarding
    for account_membership in account_memberships:
        if account_membership.email == app_user_email:
            current_logger.info(f"Removing SWAN memberships {account_membership}")
            if save:
                client.mutations.disable_account_membership(account_membership.id)

    # We need to delete the payment card history because we use it to determine eligibility for the onbarding in the app.  We delee the profile as well for consistency
    alan_payment_card_profile = (
        current_session.query(AlanPaymentCardProfile)  # noqa: ALN085
        .filter_by(app_user_id=str(app_user_id))
        .first()
    )

    if alan_payment_card_profile is None:
        current_logger.info(
            f"No payment card profile found for app_user_id {app_user_id}"
        )
        return

    alan_payment_card_profile_history_list: list[AlanPaymentCardProfileHistory] = (
        current_session.query(AlanPaymentCardProfileHistory)  # noqa: ALN085
        .filter_by(alan_payment_card_profile_id=alan_payment_card_profile.id)
        .all()
    )

    for alan_payment_card_profile_history in alan_payment_card_profile_history_list:
        current_logger.info(
            f"Removing AlanPaymentCardProfileHistory: {alan_payment_card_profile_history}"
        )
        current_session.delete(alan_payment_card_profile_history)

    alan_payment_card_balance_transaction_list: list[
        AlanPaymentCardBalanceTransaction
    ] = (
        current_session.query(AlanPaymentCardBalanceTransaction)  # noqa: ALN085
        .filter_by(alan_payment_card_profile_id=alan_payment_card_profile.id)
        .all()
    )

    for (
        alan_payment_card_balance_transaction
    ) in alan_payment_card_balance_transaction_list:
        current_logger.info(
            f"Removing AlanPaymentCardBalanceTransaction: {alan_payment_card_balance_transaction}"
        )
        current_session.delete(alan_payment_card_balance_transaction)

    alan_payment_card_recovery_debt_list: list[AlanPaymentCardRecoveryDebt] = (
        current_session.query(AlanPaymentCardRecoveryDebt)  # noqa: ALN085
        .filter_by(alan_payment_card_profile_id=alan_payment_card_profile.id)
        .all()
    )

    for alan_payment_card_recovery_debt in alan_payment_card_recovery_debt_list:
        current_logger.info(
            f"Removing AlanPaymentCardRecoveryDebt: {alan_payment_card_recovery_debt}"
        )
        current_session.delete(alan_payment_card_recovery_debt)

    current_logger.info(f"Removing AlanPaymentCardProfile: {alan_payment_card_profile}")
    current_session.delete(alan_payment_card_profile)

components.alan_pay.public.commands

app_group

alan_pay_commands module-attribute

alan_pay_commands = AppGroup(
    name="alan_pay",
    help="Main command group for the Alan Pay component",
)

push_notification

app_group

push_notification_commands
push_notification_commands()
Source code in components/alan_pay/public/commands/push_notification/app_group.py
6
7
8
@alan_pay_commands.group("push_notification")
def push_notification_commands() -> None:  # noqa: D103
    pass
register_push_notification_commands
register_push_notification_commands()
Source code in components/alan_pay/public/commands/push_notification/app_group.py
def register_push_notification_commands() -> None:  # noqa: D103
    from components.alan_pay.public.commands.push_notification import (  # noqa: F401
        send_for_non_provisioned_members,
        send_for_recovery_session,
        send_for_transaction,
    )

send_for_non_provisioned_members

send_for_non_provisioned_members
send_for_non_provisioned_members(dry_run)

Send a notification to all members that onboarded recently and didn't add their card in their wallet

Source code in components/alan_pay/public/commands/push_notification/send_for_non_provisioned_members.py
@push_notification_commands.command(requires_authentication=False)
@command_with_dry_run
def send_for_non_provisioned_members(dry_run: bool) -> None:
    """
    Send a notification to all members that onboarded recently and didn't add their card in their wallet
    """
    send_no_provisioning_push_notifications(dry_run=dry_run)

send_for_recovery_session

send_for_recovery_session
send_for_recovery_session(id, dry_run)

Send a recovery push notification to all members that will be debited during the recovery session Should be ran one day after the execution of the recovery session command

Source code in components/alan_pay/public/commands/push_notification/send_for_recovery_session.py
@push_notification_commands.command()
@click.option(
    "--id",
    type=UUID,
    required=True,
)
@command_with_dry_run
def send_for_recovery_session(
    id: UUID,
    dry_run: bool,
) -> None:
    """
    Send a recovery push notification to all members that will be debited during the recovery session
    Should be ran one day after the execution of the recovery session command
    """
    send_recovery_session_push_notifications(
        recovery_session_id=id,
        dry_run=dry_run,
    )

send_for_transaction

send_for_transaction
send_for_transaction(id, force)

Send a push notification for a given alan balance transaction ID.

Source code in components/alan_pay/public/commands/push_notification/send_for_transaction.py
@push_notification_commands.command()
@click.option(
    "--id",
    help="ID of the transaction",
    type=click.UUID,
)
@click.option(
    "--force",
    is_flag=True,
    required=False,
    default=False,
)
def send_for_transaction(
    id: UUID,
    force: bool,
) -> None:
    """Send a push notification for a given alan balance transaction ID."""
    send_alan_balance_transaction_notification(
        alan_balance_transaction_id=id,
        force=force,
    )

recovery

app_group

recovery_commands
recovery_commands()
Source code in components/alan_pay/public/commands/recovery/app_group.py
6
7
8
@alan_pay_commands.group("recovery")
def recovery_commands() -> None:  # noqa: D103
    pass
register_recovery_commands
register_recovery_commands()
Source code in components/alan_pay/public/commands/recovery/app_group.py
def register_recovery_commands() -> None:  # noqa: D103
    from components.alan_pay.public.commands.recovery import (  # noqa: F401
        list_alan_pay_recovery_sessions,
        run_alan_pay_recovery_session,
    )

list_alan_pay_recovery_sessions

list_sessions
list_sessions()

List all existing recovery session.

Source code in components/alan_pay/public/commands/recovery/list_alan_pay_recovery_sessions.py
@recovery_commands.command()
def list_sessions() -> None:
    """
    List all existing recovery session.
    """
    query = (
        current_session.query(AlanPaymentCardRecoverySession)  # noqa: ALN085
        .options(
            joinedload(AlanPaymentCardRecoverySession.debts).joinedload(
                AlanPaymentCardRecoveryDebt.recovered_transactions
            )
        )
        .order_by(AlanPaymentCardRecoverySession.recovery_period_start_date.desc())
    )

    for session in query:
        session_status = session.status
        if session_status == AlanPayRecoverySessionStatus.pending:
            total_amount_to_recover = sum(
                get_debt_amount_to_recover(alan_pay_recovery_debt_id=debt.id)
                for debt in session.debts
            )
            current_logger.info(
                f"{session}, status = %-10s Recovery date = {session.recovery_date}, total amount to recover = %.2f€"
                % (session_status, total_amount_to_recover / 100)
            )

        else:
            total_recovered_amount = sum(
                debt.recovered_amount or 0 for debt in session.debts
            )
            current_logger.info(
                f"{session}, status = %-10s Recovery date = {session.recovery_date}, total recovered amount  = %.2f€"
                % (session_status, total_recovered_amount / 100)
            )

run_alan_pay_recovery_session

run_session
run_session(id, dry_run)

Recover money from the given recovery session.

Source code in components/alan_pay/public/commands/recovery/run_alan_pay_recovery_session.py
@recovery_commands.command()
@click.option("--id", type=UUID, required=True)
@command_with_dry_run
def run_session(id: UUID, dry_run: bool) -> None:
    """
    Recover money from the given recovery session.
    """
    if dry_run:
        current_logger.info(
            "No dry-run available. If you want an overview of the amounts to recover, run `list` command instead."
        )
        return

    session = run_alan_pay_recovery_session(recovery_session_id=id)

    debts_count = len(session.debts)
    total_recovered_amount = sum(debt.recovered_amount for debt in session.debts)  # type: ignore[misc]

    current_logger.info(
        f"{session}: {debts_count} debts, total recovered amount = {total_recovered_amount / 100}€, status={session.status}"
    )

swan

app_group

register_swan_commands
register_swan_commands()
Source code in components/alan_pay/public/commands/swan/app_group.py
def register_swan_commands() -> None:  # noqa: D103
    from components.alan_pay.public.commands.swan import (  # noqa: F401
        spending_limits,
        sync,
    )
swan_commands
swan_commands()
Source code in components/alan_pay/public/commands/swan/app_group.py
6
7
8
@alan_pay_commands.group("swan")
def swan_commands() -> None:  # noqa: D103
    pass

spending_limits

check_spending_limits
check_spending_limits()

This command logs Swan cards whose spending limits are not configured as expected

Source code in components/alan_pay/public/commands/swan/spending_limits.py
@swan_commands.command()
def check_spending_limits() -> None:
    """
    This command logs Swan cards whose spending limits are not configured as expected
    """

    @dataclass
    class CardInfo:
        id: str
        card_holder_name: str
        creation_date: date
        swan_limit_errors: list[str] = field(default_factory=list)
        alan_limit_errors: list[str] = field(default_factory=list)

    client = SwanClient()

    for card in client.queries.cards():
        # Skip non-enabled cards
        if card.status_info.status != CardStatus.enabled:
            continue

        alan_limit, swan_limit = card.spending_limits
        card_info = CardInfo(
            id=card.id,
            card_holder_name=f"{card.account_membership.user.first_name} {card.account_membership.user.last_name}",
            creation_date=card.created_at.as_date(),
        )

        # Alan limit is supposed to be 400€/day
        if alan_limit.type != SpendingLimitType.account_holder:
            card_info.alan_limit_errors.append(f"Type={alan_limit.type}")
        if alan_limit.period != SpendingLimitPeriod.daily:
            card_info.alan_limit_errors.append(f"Period={alan_limit.period}")
        if alan_limit.amount.value != 400:
            card_info.alan_limit_errors.append(f"Amount={alan_limit.amount.value}")
        if alan_limit.amount.currency != "EUR":
            card_info.alan_limit_errors.append(f"Currency={alan_limit.amount.currency}")

        # Swan limit is supposed to be 600€/month
        if swan_limit.type != SpendingLimitType.partner:
            card_info.swan_limit_errors.append(f"Type={swan_limit.type}")
        if swan_limit.period != SpendingLimitPeriod.monthly:
            card_info.swan_limit_errors.append(f"Period={swan_limit.period}")
        if swan_limit.amount.value != 600:
            card_info.swan_limit_errors.append(f"Amount={swan_limit.amount.value}")
        if swan_limit.amount.currency != "EUR":
            card_info.swan_limit_errors.append(f"Currency={swan_limit.amount.currency}")

        # Log errors
        if card_info.swan_limit_errors:
            card_errors = ", ".join(card_info.swan_limit_errors)
            current_logger.warning(
                f"Misconfigured Swan spending limit for card [{card_info.id}] created on [{card_info.creation_date}] for [{card_info.card_holder_name}]: {card_errors}"
            )
        if card_info.alan_limit_errors:
            card_errors = ", ".join(card_info.alan_limit_errors)
            current_logger.warning(
                f"Misconfigured Alan spending limit for card [{card_info.id}] created on [{card_info.creation_date}] for [{card_info.card_holder_name}]: {card_errors}"
            )

sync

sync_payment_supports
sync_payment_supports(dry_run)
Source code in components/alan_pay/public/commands/swan/sync.py
@swan_commands.command()
@command_with_dry_run
def sync_payment_supports(dry_run: bool) -> None:  # noqa: D103
    from components.alan_pay.internal.business_logic.processing.banking_data import (
        sync_banking_payment_supports,
    )

    sync_banking_payment_supports(save=not dry_run)
sync_transactions
sync_transactions(dry_run)
Source code in components/alan_pay/public/commands/swan/sync.py
@swan_commands.command()
@command_with_dry_run
def sync_transactions(dry_run: bool) -> None:  # noqa: D103
    from components.alan_pay.internal.business_logic.processing.banking_data import (
        sync_banking_transactions,
    )

    sync_banking_transactions(save=not dry_run, run_side_effects=not dry_run)

components.alan_pay.public.controllers

support

SupportController

Bases: BaseController

get_support_configuration

get_support_configuration()

Returns a hardcoded configuration for Alan Pay's support.

This endpoint has been added so we can easily switch appointment scheduling on and off in the future.

Source code in components/alan_pay/public/controllers/support.py
@SupportController.action_route(
    "/config",
    methods=["GET"],
    auth_strategy=AuthorizationStrategies.authenticated(),
)
@obs.api_call()
def get_support_configuration() -> Response:
    """
    Returns a hardcoded configuration for Alan Pay's support.

    This endpoint has been added so we can easily switch appointment scheduling on and off in the future.
    """
    from components.alan_pay.internal.business_logic.rules.support import (
        is_alan_pay_appointment_scheduling_enabled,
    )

    @dataclass
    class AppointmentSchedulingConfig(DataClassJsonMixin):
        enabled: bool
        url: str

    @dataclass
    class Config(DataClassJsonMixin):
        appointment_scheduling: AppointmentSchedulingConfig

    config = Config(
        appointment_scheduling=AppointmentSchedulingConfig(
            enabled=is_alan_pay_appointment_scheduling_enabled(),
            url="https://calendar.app.google/sdB52NGjgCb4VHQL6",  # Flora
        )
    )

    return make_json_response(data=config)

support_endpoint module-attribute

support_endpoint = Endpoint('support')

components.alan_pay.public.entities

in_memory_alan_balance_transaction

InMemoryAlanBalanceTransaction dataclass

InMemoryAlanBalanceTransaction(
    id,
    alan_pay_profile_id,
    alan_pay_recovery_debt_id,
    amount_in_cents,
    event_type,
    transaction_date,
    card_payment,
    sepa_debit,
    external_transaction_id,
    push_notification_sent_at,
)
alan_pay_profile_id instance-attribute
alan_pay_profile_id

The id of the Alan Pay profile this transaction is tied to. We only expose the profile id for performance reasons. If a caller component requires the complete profile, it can fetch it using this id.

alan_pay_recovery_debt_id instance-attribute
alan_pay_recovery_debt_id

The id of the Alan Pay debt this transaction will be recovered with. We only expose the debt id for performance reasons. If a caller component requires the complete debt info, it can fetch it using this id.

amount_in_cents instance-attribute
amount_in_cents

The transaction amount in cents. Negative amount means the member owes money to Alan, positive amount means the opposite.

card_payment instance-attribute
card_payment

Additional payment information in case the event source is a card payment.

event_type instance-attribute
event_type

The transaction source.

external_transaction_id instance-attribute
external_transaction_id

The id of the external transaction in case the event type is external_transaction

id instance-attribute
id

The Alan Balance transaction id.

push_notification_sent_at instance-attribute
push_notification_sent_at

The date we sent a push notification to the member for this transaction, if applicable.

sepa_debit instance-attribute
sepa_debit

Additional SEPA debit information in case the event source is recovery.

transaction_date instance-attribute
transaction_date

The date the transaction happened.

in_memory_alan_pay_card_numbers

InMemoryAlanPayCardNumbers dataclass

InMemoryAlanPayCardNumbers(
    pan_url, expiry_date_url, cvv_url, cardholder_name
)
cardholder_name instance-attribute
cardholder_name
cvv_url instance-attribute
cvv_url
expiry_date_url instance-attribute
expiry_date_url
pan_url instance-attribute
pan_url

in_memory_alan_pay_card_payment

InMemoryAlanPayCardPayment dataclass

InMemoryAlanPayCardPayment(
    effective_date,
    amount_in_cents,
    status,
    rejection_reason,
    is_recovered,
    merchant_name,
    merchant_postal_code,
    merchant_city,
    merchant_country,
    merchant_id,
    merchant_category_code,
)
amount_in_cents instance-attribute
amount_in_cents

The payment amount in cents. If the payment is not captured yet, this amount is the authorized one.

effective_date instance-attribute
effective_date

The date the payment was initiated.

is_recovered instance-attribute
is_recovered

True if the payment was recovered, False otherwise.

merchant_category_code instance-attribute
merchant_category_code

The category code of the merchant that initiated the transaction (MCC).

merchant_city instance-attribute
merchant_city
merchant_country instance-attribute
merchant_country
merchant_id instance-attribute
merchant_id

The ID of the merchant that initiated the transaction within the banking network (MID).

merchant_name instance-attribute
merchant_name

The name of the merchant that initiated the transaction.

merchant_postal_code instance-attribute
merchant_postal_code
rejection_reason instance-attribute
rejection_reason

The reason why the payment was rejected, if applicable.

status instance-attribute
status

The status of the payment. Can be authorized, captured, or rejected.

in_memory_alan_pay_phone_number_verification

InMemoryAlanPayPhoneNumberVerification dataclass

InMemoryAlanPayPhoneNumberVerification(request_id)
request_id instance-attribute
request_id

in_memory_alan_pay_profile

InMemoryAlanPayProfile dataclass

InMemoryAlanPayProfile(
    id,
    app_id,
    app_user_id,
    status,
    provisioning_push_notification_sent_at,
)
app_id instance-attribute
app_id

The id of the source application.

app_user_id instance-attribute
app_user_id

The id of the user on the source application.

id instance-attribute
id

The profile id.

provisioning_push_notification_sent_at instance-attribute
provisioning_push_notification_sent_at

The date we sent a push notification to the member if they didn't add the card to the wallet after onboarding, if applicable.

status instance-attribute
status

The status of the Alan Pay profile.

in_memory_alan_pay_recovery_debt

InMemoryAlanPayRecoveryDebt dataclass

InMemoryAlanPayRecoveryDebt(
    id,
    profile,
    recovery_session,
    status,
    transactions,
    amount_in_cents,
    recovered_amount_in_cents,
    refunded_amount_in_cents,
)
amount_in_cents instance-attribute
amount_in_cents

Total debt amount in cents - computed live.

id instance-attribute
id

The recovery debt id.

profile instance-attribute
profile

The profile linked to the recovery debt.

recovered_amount_in_cents instance-attribute
recovered_amount_in_cents

The amount that was effectively recovered for this debt (null until the debt is effectively settled).

recovery_session instance-attribute
recovery_session

The recovery session the recovery debt is included in.

refunded_amount_in_cents instance-attribute
refunded_amount_in_cents

The amount that was effectively refunded for this debt (null until the debt is effectively settled).

status instance-attribute
status

The debt status (basically, was the debt recovered or not).

transactions instance-attribute
transactions

The transactions linked to the recovery debt.

in_memory_alan_pay_recovery_session

InMemoryAlanPayRecoverySession dataclass

InMemoryAlanPayRecoverySession(
    id,
    recovery_period_start_date,
    recovery_period_end_date,
    freeze_date,
    recovery_date,
    status,
)
freeze_date instance-attribute
freeze_date

The cutoff date after which the recovery session is frozen (amounts won't change).

id instance-attribute
id

The recovery session id.

recovery_date instance-attribute
recovery_date

The effective date of the recovery.

recovery_period_end_date instance-attribute
recovery_period_end_date

The (inclusive) end date of the recovery period.

recovery_period_start_date instance-attribute
recovery_period_start_date

The (inclusive) start date of the recovery period.

status instance-attribute
status

The status of the recovery session (pending, succeeded, failed).

in_memory_alan_pay_sepa_debit

InMemoryAlanPaySepaDebit dataclass

InMemoryAlanPaySepaDebit(recovery_session_id, sepa_mandate)
recovery_session_id instance-attribute
recovery_session_id

The id of the linked recovery session.

sepa_mandate instance-attribute
sepa_mandate

The SEPA mandate used for the debit.

Field has been made optional as we started recovery with credit transfers (from members to Alan) at the very beginning of Alan Pay. Yet, to keep things simple, we map these credit transfers to this SEPA debit model.

in_memory_alan_pay_sepa_mandate

InMemoryAlanPaySepaMandate dataclass

InMemoryAlanPaySepaMandate(
    sepa_creditor_identifier,
    debtor_name,
    debtor_iban,
    unique_mandate_reference,
)
debtor_iban instance-attribute
debtor_iban

The debtor IBAN.

debtor_name instance-attribute
debtor_name

The debtor name, as specified in the mandate.

sepa_creditor_identifier instance-attribute
sepa_creditor_identifier

"Identifiant CrΓ©ancier SEPA" in French.

unique_mandate_reference instance-attribute
unique_mandate_reference

"RΓ©fΓ©rence Unique de Mandat" in French.

components.alan_pay.public.enums

alan_balance_transaction_event_type

AlanBalanceTransactionEventType

Bases: AlanBaseEnum

The type of event that triggered the balance transaction.

card_payment class-attribute instance-attribute
card_payment = 'captured_card_payment'

A payment has been captured with the Alan Pay card.

external_transaction class-attribute instance-attribute
external_transaction = 'external_transaction'

Balance transaction issued from a component consumer.

failed_recovery class-attribute instance-attribute
failed_recovery = 'failed_recovery'

Alan tried but failed to recover debt issued from the Alan Pay card from the member.

recovery class-attribute instance-attribute
recovery = 'recovery'

Alan recovered debt issued from the Alan Pay card from the member.

refund class-attribute instance-attribute
refund = 'refund'

Alan refunded a debt issued from the Alan Pay card to the member (ex: payment canceled after we blocked some care reimbursements).

rejected_card_payment class-attribute instance-attribute
rejected_card_payment = 'rejected_card_payment'

A payment has been rejected with the Alan Pay card.

released_card_payment class-attribute instance-attribute
released_card_payment = 'released_card_payment'

A payment that was authorized - but partially (or never) captured.

alan_pay_card_payment_rejection_reason

AlanPayCardPaymentRejectionReason

Bases: AlanBaseEnum

online_payment class-attribute instance-attribute
online_payment = 'online_payment'
other_reason class-attribute instance-attribute
other_reason = 'other_reason'
spending_limit_reached class-attribute instance-attribute
spending_limit_reached = 'spending_limit_reached'
wrong_merchant_category class-attribute instance-attribute
wrong_merchant_category = 'wrong_merchant_category'

alan_pay_card_transaction_status

AlanPayCardTransactionStatus

Bases: AlanBaseEnum

authorized class-attribute instance-attribute
authorized = 'authorized'

A received payment was eventually authorized.

captured class-attribute instance-attribute
captured = 'captured'

An authorized payment was eventually captured.

received class-attribute instance-attribute
received = 'received'

The Alan Pay card was used to initiate a new payment, that still needs to be authorized or rejected.

rejected class-attribute instance-attribute
rejected = 'rejected'

A received payment was eventually rejected.

released class-attribute instance-attribute
released = 'released'

An authorized payment whose remaining authorized amount was eventually released.

alan_pay_profile_status

AlanPayProfileStatus

Bases: AlanBaseEnum

active class-attribute instance-attribute
active = 'active'
inactive class-attribute instance-attribute
inactive = 'inactive'
offboarded class-attribute instance-attribute
offboarded = 'offboarded'

alan_pay_push_notification_name

AlanPayPushNotificationName

Bases: BasePushNotificationName

alan_pay__external_transaction class-attribute instance-attribute
alan_pay__external_transaction = (
    "alan_pay__external_transaction"
)

Used by any caller component when a push notification is sent for one or several external transactions.

alan_pay__post_payment class-attribute instance-attribute
alan_pay__post_payment = 'alan_pay__post_payment'

Sent when a payment is made by a member.

alan_pay__provisioning_reminder class-attribute instance-attribute
alan_pay__provisioning_reminder = (
    "alan_pay__provisioning_reminder"
)

Sent when a profile has no card provisioning after a few days

alan_pay__recovery_session class-attribute instance-attribute
alan_pay__recovery_session = 'alan_pay__recovery_session'

Sent before the members are debited to inform them

alan_pay_recovery_debt_status

AlanPayRecoveryDebtStatus

Bases: AlanBaseEnum

executed class-attribute instance-attribute
executed = 'executed'

Bank transfer has been executed, but we did not store the result in our database yet.

initiated class-attribute instance-attribute
initiated = 'initiated'

Recovery process has been initiated, but no bank transfer has been processed yet.

pending class-attribute instance-attribute
pending = 'pending'

Debt is not recovered yet.

succeeded class-attribute instance-attribute
succeeded = 'succeeded'

Bank transfer has been executed, and we stored the result in our database without any error.

alan_pay_recovery_session_status

AlanPayRecoverySessionStatus

Bases: AlanBaseEnum

failed class-attribute instance-attribute
failed = 'failed'

Recovery session is over but some transactions failed to be recovered.

pending class-attribute instance-attribute
pending = 'pending'

Recovery session is scheduled.

succeeded class-attribute instance-attribute
succeeded = 'succeeded'

Recovery session is over and completed successfully.

components.alan_pay.public.exceptions

invalid_phone_number_verification_code

InvalidPhoneNumberVerificationCodeException

InvalidPhoneNumberVerificationCodeException(message)

Bases: Exception

Source code in components/alan_pay/public/exceptions/invalid_phone_number_verification_code.py
def __init__(self, message: str) -> None:
    super().__init__(message)