Api

T module-attribute

T = TypeVar('T', bound=Transfer)

TransferLogic

audit_accounting_report

audit_accounting_report(
    session,
    account_id,
    accounting_report_document,
    start_date,
)

Audits an accounting report against the Payment Gateway database.

:param account_id: The account id to audit :param accounting_report_document: The accounting report file in CSV format extracted from Payment provider :param start_date: The start date to filter the transfers

Note: This currently works for Adyen accounting reports only

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def audit_accounting_report(
    self,
    session: Session,
    account_id: AccountId,
    accounting_report_document: BytesIO,
    start_date: datetime | None,
) -> ReportAuditResult:
    """
    Audits an accounting report against the Payment Gateway database.

    :param account_id: The account id to audit
    :param accounting_report_document: The accounting report file in CSV format extracted from Payment provider
    :param start_date: The start date to filter the transfers

    Note: This currently works for Adyen accounting reports only

    """
    from components.payment_gateway.subcomponents.transfers.business_logic.accounting_report.audit_accounting_report import (
        audit_accounting_report as _audit_accounting_report,
    )

    return _audit_accounting_report(
        session,
        account_id=account_id,
        accounting_report_document=accounting_report_document,
        start_date=sanitize_tz_or_none(start_date),
    )

create_internal_transfer

create_internal_transfer(
    session,
    /,
    *,
    source,
    destination,
    amount,
    effective_date,
    description,
    reference,
)

Create a double-entry internal transfer.

This will create a pair of internal transfers with opposite amounts, one negative for the source and one postitive for the destination.

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def create_internal_transfer(
    self,
    session: Session,
    /,
    *,
    source: TransferHistoryId,
    destination: TransferHistoryId,
    amount: int,
    effective_date: datetime,
    description: str,
    reference: str,
) -> None:
    """
    Create a double-entry internal transfer.

    This will create a pair of internal transfers with opposite amounts, one
    negative for the source and one postitive for the destination.
    """
    if amount <= 0:
        raise ValueError(
            "Amount must be positive. Tip: invert source and destination"
        )

    InternalTransferModelBroker.create_internal_transfer(
        session,
        transfer_history_id=source,
        effective_date=effective_date,
        amount=-amount,
        description=description,
        reference=reference,
    )
    InternalTransferModelBroker.create_internal_transfer(
        session,
        transfer_history_id=destination,
        effective_date=effective_date,
        amount=amount,
        description=description,
        reference=reference,
    )
    session.flush()

create_internal_transfer_entry

create_internal_transfer_entry(
    session,
    /,
    *,
    transfer_history_id,
    amount,
    effective_date,
    description,
    reference,
)

Create a single-entry internal transfer.

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def create_internal_transfer_entry(
    self,
    session: Session,
    /,
    *,
    transfer_history_id: TransferHistoryId,
    amount: int,
    effective_date: datetime,
    description: str,
    reference: str,
) -> InternalTransferId:
    """
    Create a single-entry internal transfer.
    """
    transfer = InternalTransferModelBroker.create_internal_transfer(
        session,
        transfer_history_id=transfer_history_id,
        effective_date=effective_date,
        amount=amount,
        description=description,
        reference=reference,
    )
    session.flush()
    return InternalTransferId(transfer.id)

create_transfer_history

create_transfer_history(
    session, /, private_type, private_ref
)

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def create_transfer_history(  # noqa: D102
    self,
    session: Session,
    /,
    private_type: str,
    private_ref: str,
) -> TransferHistoryId:
    transfer_history = TransferHistoryModelBroker.create_transfer_history(
        session,
        private_type=private_type,
        private_ref=private_ref,
    )
    session.flush()
    return TransferHistoryId(transfer_history.id)

get_account_transfer_balance_within_period

get_account_transfer_balance_within_period(
    session,
    /,
    transfer_history_id,
    *,
    period_start_date,
    period_end_date,
)

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def get_account_transfer_balance_within_period(  # noqa: D102
    self,
    session: Session,
    /,
    transfer_history_id: TransferHistoryId,
    *,
    period_start_date: datetime,
    period_end_date: datetime,
) -> int:
    # Sanitize datetimes
    period_start_date = sanitize_tz(period_start_date)
    period_end_date = sanitize_tz(period_end_date)

    account_transfer_balance_at_period_start_date = (
        AccountTransferModelBroker.aggregate_account_transfer_balance(
            session,
            transfer_history_id,
            effective_date=period_start_date,
            creation_date=period_start_date,
        )
    )
    account_transfer_balance_at_period_end_date = (
        AccountTransferModelBroker.aggregate_account_transfer_balance(
            session,
            transfer_history_id,
            effective_date=period_end_date,
            creation_date=period_end_date,
        )
    )

    balance_at_start_date = (
        account_transfer_balance_at_period_start_date.received
        + account_transfer_balance_at_period_start_date.reserved
        + account_transfer_balance_at_period_start_date.balance
    )
    balance_at_end_date = (
        account_transfer_balance_at_period_end_date.received
        + account_transfer_balance_at_period_end_date.reserved
        + account_transfer_balance_at_period_end_date.balance
    )

    return balance_at_end_date - balance_at_start_date

get_all_transfers

get_all_transfers(session, /, transfer_history_ids)

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def get_all_transfers(  # noqa: D102
    self,
    session: Session,
    /,
    transfer_history_ids: list[TransferHistoryId],
) -> list[Transfer]:
    card_transfers: list[Transfer]
    internal_transfers: list[Transfer]
    bank_transfers: list[Transfer]
    account_transfers: list[Transfer]

    card_transfers = [
        card_transfer_model_to_dataclass(card_transfer)
        for card_transfer in CardTransferModelBroker.list_card_transfers_for_transfer_histories(
            session,
            transfer_history_ids,
        )
    ]

    internal_transfers = [
        internal_transfer_model_to_dataclass(internal_transfer)
        for internal_transfer in InternalTransferModelBroker.list_internal_transfers_for_transfer_histories(
            session,
            transfer_history_ids,
        )
    ]

    bank_transfers = [
        bank_transfer_model_to_dataclass(bank_transfer)
        for bank_transfer in BankTransferModelBroker.list_bank_transfers_for_transfer_histories(
            session,
            transfer_history_ids,
        )
    ]

    account_transfers = [
        account_transfer_model_to_dataclass(account_transfer)
        for account_transfer in AccountTransferModelBroker.list_account_transfers_for_transfer_histories(
            session,
            transfer_history_ids,
        )
    ]

    return sorted(
        card_transfers + internal_transfers + bank_transfers + account_transfers,
        key=lambda t: t.effective_date,
    )

get_balance

get_balance(
    session,
    /,
    transfer_history_id,
    *,
    effective_date,
    creation_date,
)

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def get_balance(  # noqa: D102
    self,
    session: Session,
    /,
    transfer_history_id: TransferHistoryId,
    *,
    effective_date: datetime,
    creation_date: datetime,
) -> BalanceInformation:
    # Sanitize datetimes
    effective_date = sanitize_tz(effective_date)
    creation_date = sanitize_tz(creation_date)

    card_transfer_balance = CardTransferModelBroker.aggregate_card_transfer_balance(
        session,
        transfer_history_id,
        effective_date=effective_date,
        creation_date=creation_date,
    )
    internal_transfer_balance = (
        InternalTransferModelBroker.aggregate_internal_transfer_balance(
            session,
            transfer_history_id,
            effective_date=effective_date,
            creation_date=creation_date,
        )
    )
    bank_transfer_balance = BankTransferModelBroker.aggregate_bank_transfer_balance(
        session,
        transfer_history_id,
        effective_date=effective_date,
        creation_date=creation_date,
    )
    account_transfer_balance = (
        AccountTransferModelBroker.aggregate_account_transfer_balance(
            session,
            transfer_history_id,
            effective_date=effective_date,
            creation_date=creation_date,
        )
    )

    return BalanceInformation(
        received=card_transfer_balance.received
        + bank_transfer_balance.received
        + account_transfer_balance.received,
        reserved=card_transfer_balance.reserved
        + bank_transfer_balance.reserved
        + account_transfer_balance.reserved,
        balance=card_transfer_balance.balance
        + bank_transfer_balance.balance
        + account_transfer_balance.balance
        + internal_transfer_balance,
    )

get_bank_transfer_balance_within_period

get_bank_transfer_balance_within_period(
    session,
    /,
    transfer_history_id,
    *,
    period_start_date,
    period_end_date,
)

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def get_bank_transfer_balance_within_period(  # noqa: D102
    self,
    session: Session,
    /,
    transfer_history_id: TransferHistoryId,
    *,
    period_start_date: datetime,
    period_end_date: datetime,
) -> int:
    # Sanitize datetimes
    period_start_date = sanitize_tz(period_start_date)
    period_end_date = sanitize_tz(period_end_date)

    bank_transfer_balance_at_period_start_date = (
        BankTransferModelBroker.aggregate_bank_transfer_balance(
            session,
            transfer_history_id,
            effective_date=period_start_date,
            creation_date=period_start_date,
        )
    )
    bank_transfer_balance_at_period_end_date = (
        BankTransferModelBroker.aggregate_bank_transfer_balance(
            session,
            transfer_history_id,
            effective_date=period_end_date,
            creation_date=period_end_date,
        )
    )

    balance_at_start_date = (
        bank_transfer_balance_at_period_start_date.received
        + bank_transfer_balance_at_period_start_date.reserved
        + bank_transfer_balance_at_period_start_date.balance
    )
    balance_at_end_date = (
        bank_transfer_balance_at_period_end_date.received
        + bank_transfer_balance_at_period_end_date.reserved
        + bank_transfer_balance_at_period_end_date.balance
    )

    return balance_at_end_date - balance_at_start_date

get_card_transfers_by_card_ids

get_card_transfers_by_card_ids(
    session,
    /,
    card_ids,
    *,
    without_transfer_history_only=False,
)

Retrieves a list of CardTransfer objects associated with a specific card ID, optionally filtering for those without a transfer history.

Parameters:

Name	Type	Description	Default
session	Session	The session to use for the database operations.	required
card_ids	Iterable[CardId]	The ID of the card for which to retrieve associated CardTransfer objects.	required
without_transfer_history_only	bool	If True, the method will return only those CardTransfer objects that have no associated transfer history. Defaults to False.	False

• list[CardTransfer]: A list of CardTransfer objects matching the provided criteria. This list may be empty if no matching objects are found.

Note: - The returned CardTransfer objects are ordered by their effective date in descending order, meaning the most recent payments are returned first.

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def get_card_transfers_by_card_ids(
    self,
    session: Session,
    /,
    card_ids: Iterable[CardId],
    *,
    without_transfer_history_only: bool = False,
) -> list[CardTransfer]:
    """
    Retrieves a list of CardTransfer objects associated with a specific card ID,
    optionally filtering for those without a transfer history.

    Args:
        session: The session to use for the database operations.
        card_ids: The ID of the card for which to retrieve associated CardTransfer objects.
        without_transfer_history_only (bool, optional): If True, the method will return only those CardTransfer objects
            that have no associated transfer history. Defaults to False.

    Returns:
    - list[CardTransfer]: A list of CardTransfer objects matching the provided criteria.
        This list may be empty if no matching objects are found.

    Note:
    - The returned CardTransfer objects are ordered by their effective date in descending order,
    meaning the most recent payments are returned first.
    """
    card_transfers = CardTransferModelBroker.list_card_transfers_for_cards(
        session,
        card_ids,
        without_transfer_history_only=without_transfer_history_only,
    )
    return [
        card_transfer_model_to_dataclass(card_transfer)
        for card_transfer in card_transfers
    ]

get_internal_transfer_balance_within_period

get_internal_transfer_balance_within_period(
    session,
    /,
    transfer_history_id,
    *,
    period_start_date,
    period_end_date,
)

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def get_internal_transfer_balance_within_period(  # noqa: D102
    self,
    session: Session,
    /,
    transfer_history_id: TransferHistoryId,
    *,
    period_start_date: datetime,
    period_end_date: datetime,
) -> int:
    # Sanitize datetimes
    period_start_date = sanitize_tz(period_start_date)
    period_end_date = sanitize_tz(period_end_date)

    internal_transfer_balance_at_period_start_date = (
        InternalTransferModelBroker.aggregate_internal_transfer_balance(
            session,
            transfer_history_id,
            effective_date=period_start_date,
            creation_date=period_start_date,
        )
    )
    internal_transfer_balance_at_period_end_date = (
        InternalTransferModelBroker.aggregate_internal_transfer_balance(
            session,
            transfer_history_id,
            effective_date=period_end_date,
            creation_date=period_end_date,
        )
    )
    return (
        internal_transfer_balance_at_period_end_date
        - internal_transfer_balance_at_period_start_date
    )

get_transfer_by_ref

get_transfer_by_ref(session, /, transfer_ref)

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def get_transfer_by_ref(  # noqa: D102
    self,
    session: Session,
    /,
    transfer_ref: str,
) -> CardTransfer | InternalTransfer | BankTransfer | AccountTransfer | None:
    _transfer_id = UUID(
        transfer_ref
    )  # the caller potentially does not know when transfer type is the id referring to
    card_transfer = CardTransferModelBroker.find_card_transfer_by_id(
        session,
        _transfer_id,
    )
    if card_transfer is not None:
        return card_transfer_model_to_dataclass(card_transfer)

    internal_transfer = InternalTransferModelBroker.find_internal_transfer_by_id(
        session,
        _transfer_id,
    )
    if internal_transfer is not None:
        return internal_transfer_model_to_dataclass(internal_transfer)

    account_transfer = AccountTransferModelBroker.find_account_transfer_by_id(
        session,
        _transfer_id,
    )
    if account_transfer is not None:
        return account_transfer_model_to_dataclass(account_transfer)

    bank_transfer = BankTransferModelBroker.find_bank_transfer_by_id(
        session,
        _transfer_id,
    )
    if bank_transfer is not None:
        return bank_transfer_model_to_dataclass(bank_transfer)

    return None

get_transfer_histories

get_transfer_histories(session, /, transfer_history_ids)

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def get_transfer_histories(  # noqa: D102
    self, session: Session, /, transfer_history_ids: list[TransferHistoryId]
) -> list[TransferHistory]:
    transfer_histories = TransferHistoryModelBroker.list_transfer_histories(
        session,
        list(transfer_history_ids),
    )

    return [
        TransferHistory(
            id=TransferHistoryId(transfer_history.id),
            private_type=transfer_history.private_type,
            private_ref=transfer_history.private_ref,
            created_at=transfer_history.created_at,
        )
        for transfer_history in transfer_histories
    ]

get_transfer_history_id_by_private_ref

get_transfer_history_id_by_private_ref(
    session, /, private_type, private_ref
)

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def get_transfer_history_id_by_private_ref(  # noqa: D102
    self,
    session: Session,
    /,
    private_type: str,
    private_ref: str,
) -> TransferHistoryId:
    transfer_history = (
        TransferHistoryModelBroker.get_transfer_history_by_private_ref(
            session,
            private_type=private_type,
            private_ref=private_ref,
        )
    )

    return TransferHistoryId(transfer_history.id)

get_transfers_by_effective_date

get_transfers_by_effective_date(
    session,
    /,
    transfer_history_ids,
    *,
    start_effective_date,
    end_effective_date=None,
)

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def get_transfers_by_effective_date(  # noqa: D102
    self,
    session: Session,
    /,
    transfer_history_ids: list[TransferHistoryId],
    *,
    start_effective_date: datetime,
    end_effective_date: datetime | None = None,
) -> list[Transfer]:
    card_transfers: list[Transfer]
    internal_transfers: list[Transfer]
    bank_transfers: list[Transfer]
    account_transfers: list[Transfer]

    # Sanitize datetimes
    start_effective_date = sanitize_tz(start_effective_date)
    end_effective_date = sanitize_tz_or_none(end_effective_date)

    card_transfers = [
        card_transfer_model_to_dataclass(card_transfer)
        for card_transfer in CardTransferModelBroker.list_card_transfers_for_transfer_histories(
            session,
            transfer_history_ids,
            start_effective_date=start_effective_date,
            end_effective_date=end_effective_date,
        )
    ]

    internal_transfers = [
        internal_transfer_model_to_dataclass(internal_transfer)
        for internal_transfer in InternalTransferModelBroker.list_internal_transfers_for_transfer_histories(
            session,
            transfer_history_ids,
            start_effective_date=start_effective_date,
            end_effective_date=end_effective_date,
        )
    ]

    bank_transfers = [
        bank_transfer_model_to_dataclass(bank_transfer)
        for bank_transfer in BankTransferModelBroker.list_bank_transfers_for_transfer_histories(
            session,
            transfer_history_ids,
            start_effective_date=start_effective_date,
            end_effective_date=end_effective_date,
        )
    ]

    account_transfers = [
        account_transfer_model_to_dataclass(account_transfer)
        for account_transfer in AccountTransferModelBroker.list_account_transfers_for_transfer_histories(
            session,
            transfer_history_ids,
            start_effective_date=start_effective_date,
            end_effective_date=end_effective_date,
        )
    ]

    return sorted(
        card_transfers + internal_transfers + bank_transfers + account_transfers,
        key=lambda t: t.effective_date,
    )

get_transfers_by_event_date

get_transfers_by_event_date(
    session: Session,
    /,
    transfer_history_ids: list[TransferHistoryId],
    *,
    start_effective_date: datetime,
    end_effective_date: datetime | None,
    transfer_types: None = None,
) -> list[Transfer]

get_transfers_by_event_date(
    session: Session,
    /,
    transfer_history_ids: list[TransferHistoryId],
    *,
    start_effective_date: datetime,
    end_effective_date: datetime | None,
    transfer_types: Sequence[type[T]],
) -> list[T]

get_transfers_by_event_date(
    session,
    /,
    transfer_history_ids,
    *,
    start_effective_date,
    end_effective_date,
    transfer_types=None,
)

Use when dealing with card transfers and

you are not only interested in the date at which the card transfer was originated, but want to fetch the individual events associated to it. A Card payment can be theoretically opened for 30 days on mastercard (on Adyen is even more, we have cases of years, a bug on their side). During this period, each event can performa a balance change, therefore the date at which you made a balance change can be different of that of when the card transfer was originated. In Spain, for healthy benefits we have a real time approach for card transfers, therefore the flag filter_by_event_dates allows to retrieve events that happened during a given period, not just card transfers for which we only have the date at which it was started. NON card transfers are more simple and don't need to be treated differently for real time cases vs non-real time.

This method returns each type of transfer ordered by created_at, then it orders by effective_at

Source code in components/payment_gateway/subcomponents/transfers/protected/api.py

@obs.api_call()
def get_transfers_by_event_date(
    self,
    session: Session,
    /,
    transfer_history_ids: list[TransferHistoryId],
    *,
    start_effective_date: datetime,
    end_effective_date: datetime | None,
    transfer_types: Sequence[type[T]] | None = None,
) -> list[Transfer] | list[T]:
    """
        Use when dealing with card transfers and
    you are not only interested in the date at which the card transfer was
    originated, but want to fetch the individual events associated to it. A Card
    payment can be theoretically opened for 30 days on mastercard (on Adyen is even
    more, we have cases of years, a bug on their side). During this period, each
    event can performa a balance change, therefore the date at which you made a
    balance change can be different of that of when the card transfer was
    originated. In Spain, for healthy benefits we have a real time approach for
    card transfers, therefore the flag filter_by_event_dates allows to retrieve
    events that happened during a given period, not just card transfers for which we
    only have the date at which it was started. NON card transfers are more simple
    and don't need to be treated differently for real time cases vs non-real time.

    This method returns each type of transfer ordered by created_at, then it orders by
    effective_at
    """
    card_transfers: list[Transfer]
    internal_transfers: list[Transfer]
    bank_transfers: list[Transfer]
    account_transfers: list[Transfer]

    if transfer_types is None:
        transfer_types = [
            CardTransfer,  # type: ignore[list-item]
            BankTransfer,  # type: ignore[list-item]
            AccountTransfer,  # type: ignore[list-item]
            InternalTransfer,  # type: ignore[list-item]
        ]

    # Sanitize datetimes
    start_effective_date = sanitize_tz(start_effective_date)
    end_effective_date = sanitize_tz_or_none(end_effective_date)

    card_transfers = []
    if CardTransfer in transfer_types:
        card_transfers = [
            card_transfer_model_to_dataclass(
                card_transfer,
                start_event_date=start_effective_date,
                end_event_date=end_effective_date,
            )
            for card_transfer in CardTransferModelBroker.list_card_transfers_for_transfer_histories_by_event_date(
                session,
                transfer_history_ids,
                start_effective_date=start_effective_date,
                end_effective_date=end_effective_date,
                order_by_created_at=True,
            )
        ]

    internal_transfers = []
    if InternalTransfer in transfer_types:
        internal_transfers = [
            internal_transfer_model_to_dataclass(internal_transfer)
            for internal_transfer in InternalTransferModelBroker.list_internal_transfers_for_transfer_histories(
                session,
                transfer_history_ids,
                start_effective_date=start_effective_date,
                end_effective_date=end_effective_date,
                order_by_created_at=True,
            )
        ]

    bank_transfers = []
    if BankTransfer in transfer_types:
        bank_transfers = [
            bank_transfer_model_to_dataclass(bank_transfer)
            for bank_transfer in BankTransferModelBroker.list_bank_transfers_for_transfer_histories(
                session,
                transfer_history_ids,
                start_effective_date=start_effective_date,
                end_effective_date=end_effective_date,
                order_by_created_at=True,
            )
        ]

    account_transfers = []
    if AccountTransfer in transfer_types:
        account_transfers = [
            account_transfer_model_to_dataclass(account_transfer)
            for account_transfer in AccountTransferModelBroker.list_account_transfers_for_transfer_histories(
                session,
                transfer_history_ids,
                start_effective_date=start_effective_date,
                end_effective_date=end_effective_date,
                order_by_created_at=True,
            )
        ]

    return sorted(
        card_transfers + internal_transfers + bank_transfers + account_transfers,
        key=lambda t: t.effective_date,
    )