Policies

components.payment_gateway.subcomponents.authorizations.business_logic.policies.authorization_request_processing

AuthorizationRequestProcessingPolicy

AuthorizationRequestProcessingPolicy(
    expense_category_resolver, line_of_credit_resolver
)

This class is responsible for processing authorization request events.

It also manages the lifecycle of the authorization requests upon receiving subsequent payment events.

Source code in components/payment_gateway/subcomponents/authorizations/business_logic/policies/authorization_request_processing.py

def __init__(
    self,
    expense_category_resolver: ExpenseCategoryResolver,
    line_of_credit_resolver: LineOfCreditResolver,
):
    self.expense_category_resolver = expense_category_resolver
    self.line_of_credit_resolver = line_of_credit_resolver

authorize_pending_transaction

authorize_pending_transaction(session, pending_transaction)

Authorize a pending transaction.

This method implements the core logic of the Authorization Relay, approving or declining pending transactions based on the high level design described here:

https://www.notion.so/alaninsurance/Authorization-Relay-High-Level-Design-1b21426e8be780e8ba04ee8b2a57deea?pvs=4 ⧉

Source code in components/payment_gateway/subcomponents/authorizations/business_logic/policies/authorization_request_processing.py

@obs.api_call()
def authorize_pending_transaction(
    self,
    session: Session,
    pending_transaction: PendingTransaction,
) -> AuthorizationResult:
    """Authorize a pending transaction.

    This method implements the core logic of the Authorization Relay,
    approving or declining pending transactions based on the high level
    design described here:

    https://www.notion.so/alaninsurance/Authorization-Relay-High-Level-Design-1b21426e8be780e8ba04ee8b2a57deea?pvs=4
    """
    expense_category_codes = (
        self.expense_category_resolver.resolve_expense_categories(
            pending_transaction
        )
    )
    if expense_category_codes is None or len(expense_category_codes) == 0:
        # Decline transactions that don't match any expense category
        return self._decline_pending_transaction(
            session,
            pending_transaction,
        )

    expense_category_ids = {
        ExpenseCategoryId(expense_category_id)
        for expense_category_id in ExpenseCategoryModelBroker.list_expense_category_ids_by_codes(
            session,
            list(expense_category_codes),
        )
    }
    if len(expense_category_ids) != len(expense_category_codes):
        # At least one expense category could not be resolved
        return self._decline_pending_transaction(
            session,
            pending_transaction,
        )

    line_of_credit_ids = self.line_of_credit_resolver.resolve_lines_of_credit(
        expense_category_ids,
        pending_transaction,
    )
    if not line_of_credit_ids or len(line_of_credit_ids) != len(
        expense_category_ids
    ):
        return self._decline_pending_transaction(
            session,
            pending_transaction,
        )

    # This operation is atomic
    result = ExpenseTrackerModelBroker.check_and_update_expensed_credits(
        session,
        [line_of_credit_id for line_of_credit_id in line_of_credit_ids],
        pending_transaction.amount,
    )
    if not result:
        return self._decline_pending_transaction(
            session,
            pending_transaction,
        )

    return self._approve_pending_transaction(
        session,
        pending_transaction,
        expense_category_ids,
        line_of_credit_ids,
    )

expense_category_resolver instance-attribute

expense_category_resolver = expense_category_resolver

line_of_credit_resolver instance-attribute

line_of_credit_resolver = line_of_credit_resolver

on_payment_event

on_payment_event(session, data)

Process an incoming Adyen payment event.

This method is called when a payment event is received from Adyen. It extracts any pending transaction identifier from the payload and releases it if it exists.

Source code in components/payment_gateway/subcomponents/authorizations/business_logic/policies/authorization_request_processing.py

@obs.event_subscriber()
def on_payment_event(
    self,
    session: Session,
    data: "TransferData",
) -> None:
    """Process an incoming Adyen payment event.

    This method is called when a payment event is received from Adyen. It
    extracts any pending transaction identifier from the payload and
    releases it if it exists.
    """
    from components.payment_gateway.subcomponents.authorizations.adapters.adyen.helpers import (
        authorization_request_external_id_from_transfer_data,
    )

    external_id = authorization_request_external_id_from_transfer_data(data)
    if external_id is not None:
        self.update_active_transaction(
            session,
            provider=PaymentServiceProvider.adyen,
            external_id=external_id,
        )

update_active_transaction

update_active_transaction(session, provider, external_id)

Update a previously authorized transaction.

This method manages the lifecycle of the authorization requests upon events received from the PSP.

Source code in components/payment_gateway/subcomponents/authorizations/business_logic/policies/authorization_request_processing.py

@obs.api_call()
def update_active_transaction(
    self,
    session: Session,
    provider: PaymentServiceProvider,
    external_id: str,
) -> None:
    """Update a previously authorized transaction.

    This method manages the lifecycle of the authorization requests upon
    events received from the PSP.
    """
    self._release_pending_transaction(
        session,
        provider,
        external_id,
    )