Skip to content

Reference

shared.services.payment_providers.revolut.entities

CounterPartyInfo dataclass 

CounterPartyInfo(counterparty_id, account_id)

Bases: DataClassJsonMixin

account_id instance-attribute 

account_id

counterparty_id instance-attribute 

counterparty_id

RevolutBusinessAccountConfig dataclass 

RevolutBusinessAccountConfig(
    base_url,
    api_credentials_secret_name,
    private_key,
    private_key_secret_name,
    client_id,
    issuer,
    reimbursement_account_id=None,
)

api_credentials_secret_name instance-attribute 

api_credentials_secret_name

base_url instance-attribute 

base_url

client_id instance-attribute 

client_id

issuer instance-attribute 

issuer

private_key instance-attribute 

private_key

private_key_secret_name instance-attribute 

private_key_secret_name

reimbursement_account_id class-attribute instance-attribute 

reimbursement_account_id = None

RevolutBusinessAccountName

Bases: AlanBaseEnum

alan_insurance class-attribute instance-attribute 

alan_insurance = 'alan_insurance'

alan_sa class-attribute instance-attribute 

alan_sa = 'alan_sa'

alan_services class-attribute instance-attribute 

alan_services = 'alan_services'

marmot_be_belfius_bank class-attribute instance-attribute 

marmot_be_belfius_bank = 'marmot_be_belfius_bank'

marmot_be_belfius_insurance class-attribute instance-attribute 

marmot_be_belfius_insurance = 'marmot_be_belfius_insurance'

RevolutBusinessAccountNamesAvailableByCountry

belgium class-attribute instance-attribute 

belgium = (alan_insurance, alan_sa)

france class-attribute instance-attribute 

france = (alan_insurance, alan_services, alan_sa)

prevoyance_france class-attribute instance-attribute 

prevoyance_france = (alan_insurance, alan_sa)

spain class-attribute instance-attribute 

spain = (alan_insurance, alan_sa)

RevolutPrevoyancePaymentAccount dataclass 

RevolutPrevoyancePaymentAccount(lamie, cnp, alan=None)

This dataclass is storing the name of the env vars used to setup Revolut client. We introduced the model when we had to switch our Revolut business accounts from Alan SA to Alan Insurance.

Alan property is nullable as the bank account for Alan insurer was created immediately on Alan Insurance business account.

alan class-attribute instance-attribute 

alan = None

cnp instance-attribute 

cnp

lamie instance-attribute 

lamie

RevolutTransactionInfo dataclass 

RevolutTransactionInfo(
    id,
    type,
    request_id,
    state,
    created_at,
    updated_at,
    completed_at=None,
    reason_code=None,
    related_transaction_id=None,
    reference=None,
    legs=list(),
)

https://developer.revolut.com/docs/business/get-transaction#response ⧉

Leg dataclass 

Leg(
    leg_id,
    amount,
    currency,
    account_id,
    fee=None,
    bill_amount=None,
    bill_currency=None,
    description=None,
    balance=None,
)
account_id instance-attribute 
account_id
amount instance-attribute 
amount
balance class-attribute instance-attribute 
balance = None
bill_amount class-attribute instance-attribute 
bill_amount = None
bill_currency class-attribute instance-attribute 
bill_currency = None
currency instance-attribute 
currency
description class-attribute instance-attribute 
description = None
fee class-attribute instance-attribute 
fee = None
leg_id instance-attribute 
leg_id

completed_at class-attribute instance-attribute 

completed_at = None

created_at instance-attribute 

created_at

id instance-attribute 

id

legs class-attribute instance-attribute 

legs = field(default_factory=list)

reason_code class-attribute instance-attribute 

reason_code = None

The reason code when the transaction state is declined or failed.

reference class-attribute instance-attribute 

reference = None

related_transaction_id class-attribute instance-attribute 

related_transaction_id = None

request_id instance-attribute 

request_id

state instance-attribute 

state

type instance-attribute 

type

updated_at instance-attribute 

updated_at

RevolutTransactionState

Bases: AlanBaseEnum

Possible states for a payment transaction.

completed class-attribute instance-attribute 

completed = 'completed'

created class-attribute instance-attribute 

created = 'created'

declined class-attribute instance-attribute 

declined = 'declined'

failed class-attribute instance-attribute 

failed = 'failed'

pending class-attribute instance-attribute 

pending = 'pending'

reverted class-attribute instance-attribute 

reverted = 'reverted'

RevolutTransactionType

Bases: AlanBaseEnum

Possible types for a payment transaction.

atm class-attribute instance-attribute 

atm = 'atm'

card_chargeback class-attribute instance-attribute 

card_chargeback = 'card_chargeback'

card_credit class-attribute instance-attribute 

card_credit = 'card_credit'

card_payment class-attribute instance-attribute 

card_payment = 'card_payment'

card_refund class-attribute instance-attribute 

card_refund = 'card_refund'

exchange class-attribute instance-attribute 

exchange = 'exchange'

fee class-attribute instance-attribute 

fee = 'fee'

loan class-attribute instance-attribute 

loan = 'loan'

refund class-attribute instance-attribute 

refund = 'refund'

tax class-attribute instance-attribute 

tax = 'tax'

tax_refund class-attribute instance-attribute 

tax_refund = 'tax_refund'

topup class-attribute instance-attribute 

topup = 'topup'

topup_return class-attribute instance-attribute 

topup_return = 'topup_return'

transfer class-attribute instance-attribute 

transfer = 'transfer'

shared.services.payment_providers.revolut.errors

PotentialSwiftError 

PotentialSwiftError(code=None, message=None)

Bases: RevolutError

A special case of RevolutError. Raised when we think it may be fixable by resetting the SWIFT. Context: many calls to /counterparty fail because we supply the wrong SWIFT number. We don't get details from the 400 response from Revolut, but we want to communicate that it's possibly caused by a SWIFT problem.

Source code in shared/services/payment_providers/revolut/errors.py 
4
5
6
def __init__(self, code=None, message=None) -> None:  # type: ignore[no-untyped-def]
    self.code = code
    self.message = message

RevolutError 

RevolutError(code=None, message=None)

Bases: Exception

A generic revolut exception we should handle

Source code in shared/services/payment_providers/revolut/errors.py 
4
5
6
def __init__(self, code=None, message=None) -> None:  # type: ignore[no-untyped-def]
    self.code = code
    self.message = message

code instance-attribute 

code = code

message instance-attribute 

message = message

RevolutInternalError 

RevolutInternalError(code=None, message=None)

Bases: RevolutError

Revolut issue, we can't act on it

Source code in shared/services/payment_providers/revolut/errors.py 
4
5
6
def __init__(self, code=None, message=None) -> None:  # type: ignore[no-untyped-def]
    self.code = code
    self.message = message

RevolutNotAuthorizedError 

RevolutNotAuthorizedError(code=None, message=None)

Bases: RevolutError

The access token is not valid.

We have a scheduled job running every 15 minutes to rotate the access token as it is valid for less than 1h. If the command failed to run for some reason, the token won't have been rotated, and will thus be expired / invalid.

See also https://www.notion.so/alaninsurance/Revolut-API-keys-Management-d536aabbf7ba4c278dc86a2da05b6b21#1cca1b9404314144bdc58f2fbd06b791 ⧉

Source code in shared/services/payment_providers/revolut/errors.py 
4
5
6
def __init__(self, code=None, message=None) -> None:  # type: ignore[no-untyped-def]
    self.code = code
    self.message = message

RevolutNotFoundError 

RevolutNotFoundError(code=None, message=None)

Bases: RevolutError

Resource does not exist on Revolut, it is up to us to decide whether we take this error into account or not

Source code in shared/services/payment_providers/revolut/errors.py 
4
5
6
def __init__(self, code=None, message=None) -> None:  # type: ignore[no-untyped-def]
    self.code = code
    self.message = message

RevolutParallelRequestsError 

RevolutParallelRequestsError(code=None, message=None)

Bases: RevolutError

See https://www.notion.so/alaninsurance/Revolut-payments-and-its-failures-68efe555a6a14610835b928ce09de883#ee51064d48de444f9763fbf66831b34d ⧉

Source code in shared/services/payment_providers/revolut/errors.py 
4
5
6
def __init__(self, code=None, message=None) -> None:  # type: ignore[no-untyped-def]
    self.code = code
    self.message = message

shared.services.payment_providers.revolut.helpers

EEA_COUNTRIES module-attribute 

EEA_COUNTRIES = [
    "AT",
    "AX",
    "BE",
    "BG",
    "CY",
    "CZ",
    "DE",
    "DK",
    "EA",
    "EE",
    "ES",
    "FI",
    "FR",
    "GB",
    "GF",
    "GI",
    "GP",
    "GR",
    "HR",
    "HU",
    "IC",
    "IE",
    "IT",
    "LT",
    "LU",
    "LV",
    "MF",
    "MQ",
    "MT",
    "NL",
    "PL",
    "PT",
    "RE",
    "RO",
    "SE",
    "SI",
    "SK",
    "YT",
    "NO",
    "IS",
    "LI",
    "JE",
    "GG",
    "IM",
]

PREVOYANCE_BANK_ACCOUNT_CONFIG_MAPPING module-attribute 

PREVOYANCE_BANK_ACCOUNT_CONFIG_MAPPING = {
    alan_sa: RevolutPrevoyancePaymentAccount(
        cnp="REVOLUT_PREVOYANCE_CNP_PAYMENT_ACCOUNT_ID",
        lamie="REVOLUT_PREVOYANCE_LAMIE_PAYMENT_ACCOUNT_ID",
    ),
    alan_insurance: RevolutPrevoyancePaymentAccount(
        cnp="REVOLUT_ALAN_INSURANCE_PREVOYANCE_CNP_PAYMENT_ACCOUNT_ID",
        lamie="REVOLUT_ALAN_INSURANCE_PREVOYANCE_LAMIE_PAYMENT_ACCOUNT_ID",
        alan="REVOLUT_ALAN_INSURANCE_PREVOYANCE_ALAN_PAYMENT_ACCOUNT_ID",
    ),
}

get_revolut_config_from_business_account_name 

get_revolut_config_from_business_account_name(account_name)
Source code in shared/services/payment_providers/revolut/helpers.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def get_revolut_config_from_business_account_name(
    account_name: RevolutBusinessAccountName,
) -> RevolutBusinessAccountConfig:
    if RevolutBusinessAccountName[account_name] not in RevolutBusinessAccountName:
        raise ValueError(
            f"Invalid Revolut business account name {account_name}. "
            f"Valid values are: {', '.join(RevolutBusinessAccountName.get_values())}"
        )

    revolut_config = _business_account_name_mapping[account_name]

    if not revolut_config:
        raise ValueError(
            f"Could not find Revolut config for business account name {account_name}"
        )

    return revolut_config

is_reimbursement_transaction 

is_reimbursement_transaction(transaction)

Check if a Revolut transaction is linked to Alan's Revolut reimbursement account

Source code in shared/services/payment_providers/revolut/helpers.py 
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
def is_reimbursement_transaction(transaction: RevolutTransactionInfo) -> bool:
    """
    Check if a Revolut transaction is linked to Alan's Revolut reimbursement account
    """

    reimbursement_account_ids = {
        # Revolut account for Health
        current_config.get("REVOLUT_REIMBURSEMENT_ACCOUNT_ID"),
        # Revolut account for Health insurance
        current_config.get("REVOLUT_ALAN_INSURANCE_REIMBURSEMENT_ACCOUNT_ID"),
        # Prevoyance bank accounts declared on Alan SA business account.
        # Still used, but supposed to be migrated during 2024-Q1
        current_config.get("REVOLUT_PREVOYANCE_LAMIE_PAYMENT_ACCOUNT_ID"),
        current_config.get("REVOLUT_PREVOYANCE_CNP_PAYMENT_ACCOUNT_ID"),
        # Prevoyance bank accounts declared on Alan Insurance business account.
        # CNP & Lamie are not used yet (see previous comment), Alan will be used starting 2024-01-01
        current_config.get("REVOLUT_ALAN_INSURANCE_PREVOYANCE_CNP_PAYMENT_ACCOUNT_ID"),
        current_config.get(
            "REVOLUT_ALAN_INSURANCE_PREVOYANCE_LAMIE_PAYMENT_ACCOUNT_ID"
        ),
        current_config.get("REVOLUT_ALAN_INSURANCE_PREVOYANCE_ALAN_PAYMENT_ACCOUNT_ID"),
    }

    account_ids = [leg.account_id for leg in transaction.legs]

    return any(
        reimbursement_account_id in account_ids
        for reimbursement_account_id in reimbursement_account_ids
    )

shared.services.payment_providers.revolut.mixins

Mixins for Revolut payment provider integration.

RevolutPaymentMixin

Mixin for models that need to track Revolut payment information.

Provides fields for Revolut payment ID, refund ID, and business account.

revolut_business_account_name class-attribute instance-attribute 

revolut_business_account_name = mapped_column(
    AlanBaseEnumTypeDecorator(RevolutBusinessAccountName),
    nullable=False,
    index=True,
    server_default=alan_insurance,
)

revolut_payment_id class-attribute instance-attribute 

revolut_payment_id = mapped_column(String(255), unique=True)

revolut_refund_id class-attribute instance-attribute 

revolut_refund_id = mapped_column(String(255), unique=True)

validates_revolut_business_account_name class-attribute instance-attribute 

validates_revolut_business_account_name = create_validator(
    "revolut_business_account_name"
)

shared.services.payment_providers.revolut.noop_revolut_client

NoopRevolutClient

Revolut client that does nothing

get_supported_currency 

get_supported_currency()
Source code in shared/services/payment_providers/revolut/noop_revolut_client.py 
22
23
def get_supported_currency(self) -> str:
    return "EUR"

get_transaction_info 

get_transaction_info(
    transaction_id=None,
    request_id=None,
    ignore_not_found_error=False,
)
Source code in shared/services/payment_providers/revolut/noop_revolut_client.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def get_transaction_info(
    self,
    transaction_id: str | None = None,
    request_id: str | None = None,
    ignore_not_found_error: bool = False,  # noqa: ARG002
) -> RevolutTransactionInfo:
    return RevolutTransactionInfo(
        id=transaction_id or str(uuid.uuid4()),
        request_id=request_id or str(uuid.uuid4()),
        state=RevolutTransactionState.failed,
        created_at=datetime.fromisoformat("2023-03-12T12:08:07.833414"),
        updated_at=datetime.fromisoformat("2023-03-13T12:08:07.833705"),
        completed_at=datetime.fromisoformat("2023-03-13T12:08:07.833705"),
        reference=None,
        type=RevolutTransactionType.transfer,
        legs=[],
    )

pay 

pay(
    request_id,
    amount_in_cents,
    recipient_first_name=None,
    recipient_last_name=None,
    recipient_company_name=None,
    revolut_counterparty_id=None,
    revolut_account_id=None,
    ibancode=None,
    swift=None,
    country_iso=None,
    address=None,
    payment_description=None,
    reimbursement_payment_id=None,
)
Source code in shared/services/payment_providers/revolut/noop_revolut_client.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def pay(  # type: ignore[no-untyped-def]
    self,
    # The request_id is controlled by us, and is a string of 40 chars max
    request_id: str,
    amount_in_cents: int,
    # These fields must be specified if the recipient is an individual
    recipient_first_name: str | None = None,  # noqa: ARG002
    recipient_last_name: str | None = None,  # noqa: ARG002
    # This field must be specified if the recipient is a company or a hospital
    recipient_company_name: str | None = None,  # noqa: ARG002
    revolut_counterparty_id=None,
    revolut_account_id=None,
    ibancode=None,  # noqa: ARG002
    swift=None,  # noqa: ARG002
    country_iso=None,  # noqa: ARG002
    address: BillingAddress | None = None,  # noqa: ARG002
    payment_description: str | None = None,  # noqa: ARG002
    reimbursement_payment_id: str | None = None,  # noqa: ARG002
) -> PaymentInfo:
    if not isinstance(amount_in_cents, int):
        raise ValueError("amount is not int")

    if not request_id:
        raise ValueError("request_id is not set")

    if revolut_counterparty_id and revolut_account_id:
        counterparty_info = CounterPartyInfo(
            revolut_counterparty_id, revolut_account_id
        )
    elif not revolut_counterparty_id and not revolut_account_id:
        # We generate fake `counterparty_id` and `account_id`
        counterparty_info = CounterPartyInfo(str(uuid.uuid4()), str(uuid.uuid4()))
    else:
        raise ValueError(
            "revolut_counterparty_id and revolut_account_id must be set or omitted together"
        )

    # We generate a fake `payment_id`
    return PaymentInfo(
        counterparty_id=counterparty_info.counterparty_id,
        account_id=counterparty_info.account_id,
        transaction_id=str(uuid.uuid4()),
        request_id=request_id,
        created_at=datetime.now(),
        transfer_type=PaymentTransferType.instant,
        currency="EUR",
    )

shared.services.payment_providers.revolut.revolut_client

RevolutAuth 

RevolutAuth(token)

Bases: AuthBase

Source code in shared/services/payment_providers/revolut/revolut_client.py 
51
52
def __init__(self, token: str) -> None:
    self.token = token

__call__ 

__call__(r)
Source code in shared/services/payment_providers/revolut/revolut_client.py 
54
55
56
def __call__(self, r: requests.PreparedRequest) -> requests.PreparedRequest:
    r.headers["Authorization"] = f"Bearer {self.token}"
    return r

token instance-attribute 

token = token

RevolutClient 

RevolutClient(
    business_account_name, reimbursement_account_id=None
)
Source code in shared/services/payment_providers/revolut/revolut_client.py 
 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
101
102
103
104
105
106
107
108
109
110
111
def __init__(  # type: ignore[no-untyped-def]
    self,
    business_account_name: RevolutBusinessAccountName,
    reimbursement_account_id=None,
) -> None:
    if business_account_name is None:
        business_account_name = current_config.get("REVOLUT_MAIN_ACCOUNT_NAME")  # type: ignore[unreachable]

    if business_account_name is None:
        raise ValueError("Revolut Business account name is not set configured")

    revolut_config = get_revolut_config_from_business_account_name(
        business_account_name
    )
    private_key = raw_secret_from_config(
        config_key=revolut_config.private_key_secret_name,
        default_secret_value=current_config.get(revolut_config.private_key),
    )
    if private_key is None:
        raise ValueError("Revolut private key is not set/configured")
    self._client_id = get_config_or_fail(revolut_config.client_id)
    self._private_key = private_key
    self._base_url = get_config_or_fail(revolut_config.base_url)

    # Alan Services Revolut config doesn't have default reimbursement account reimbursement_account_id parameter should be passed
    if not revolut_config.reimbursement_account_id:
        if not reimbursement_account_id:
            raise ValueError(
                "Reimbursement account ID is missing to initialize Revolut client"
            )
        self._reimbursement_account_id = reimbursement_account_id
    else:
        self._reimbursement_account_id = (
            reimbursement_account_id
            or get_config_or_fail(revolut_config.reimbursement_account_id)
        )

    self._revolut_issuer = get_config_or_fail(revolut_config.issuer)

    secret_name = get_config_or_fail(revolut_config.api_credentials_secret_name)
    if not secret_name:
        raise RuntimeError(
            f"{revolut_config.api_credentials_secret_name} is not set"
        )

    self._secrets_manager_client = SecretsManager(secret_name)

AUTH_ENDPOINT class-attribute instance-attribute 

AUTH_ENDPOINT = 'auth/token'

COUNTERPARTY_ENDPOINT class-attribute instance-attribute 

COUNTERPARTY_ENDPOINT = 'counterparty'

PAY_ENDPOINT class-attribute instance-attribute 

PAY_ENDPOINT = 'pay'

REFUND_TRANSACTION_ENDPOINT class-attribute instance-attribute 

REFUND_TRANSACTION_ENDPOINT = 'transactions'

TRANSACTION_ENDPOINT class-attribute instance-attribute 

TRANSACTION_ENDPOINT = 'transaction/{}'

create_counterparty 

create_counterparty(
    *,
    recipient_company_name=None,
    recipient_first_name=None,
    recipient_last_name=None,
    iban,
    swift,
    country_iso=None,
    address,
    reimbursement_payment_id=None
)

Create a new counterparty in Revolut https://developer.revolut.com/docs/business/add-counterparty ⧉

:param country_iso: Set by default to the IBAN first two characters. Must be overridden for countries like French Polynesia (=> PF). You can use IBANClient().get_iban_info(iban) to fetch it

Source code in shared/services/payment_providers/revolut/revolut_client.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def create_counterparty(
    self,
    *,
    # This field must be specified if the recipient is a company or a hospital
    recipient_company_name: str | None = None,
    # These two fields must be specified if the recipient is an individual
    recipient_first_name: str | None = None,
    recipient_last_name: str | None = None,
    iban: str,
    swift: str | None,
    country_iso: str | None = None,
    address: BillingAddress | None,
    reimbursement_payment_id: str | None = None,
) -> CounterPartyInfo:
    """
    Create a new counterparty in Revolut https://developer.revolut.com/docs/business/add-counterparty

    :param country_iso: Set by default to the IBAN first two characters. Must be overridden for countries like French Polynesia (=> PF).
        You can use IBANClient().get_iban_info(iban) to fetch it
    """
    current_logger.info("Creating Revolut Counterparty...")

    bank_country = country_iso if country_iso else iban[:2]
    # Revolut counterparty API will fail if company name contains more than 80 characters
    normalized_company_name = self._normalize_name_for_revolut(
        recipient_company_name, max_length=80
    )
    # Revolut counterparty API will fail if user names contains more than 40 characters
    normalized_first_name = self._normalize_name_for_revolut(
        recipient_first_name, max_length=40
    )
    normalized_last_name = self._normalize_name_for_revolut(
        recipient_last_name, max_length=40
    )

    payload = {
        "bank_country": bank_country,
        "currency": self.get_supported_currency(),
    }

    # Revolut does not handle properly new IBAN format from Senegal (at least), so they shared a
    # workaround to avoid rejections: https://alanhealth.slack.com/archives/CPN8N22LU/p1723818151846399
    # For French polynesia we need to send the account_no and bic without the IBAN
    # https://alanhealth.slack.com/archives/CPN8N22LU/p1709893605379069?thread_ts=1709813063.688019&cid=CPN8N22LU
    if bank_country in {"SN", "PF", "NC"}:
        payload["account_no"] = iban
        payload["bic"] = mandatory(
            swift, f"Swift is mandatory for IBAN from {bank_country}"
        )
    else:
        payload["iban"] = iban
        if swift:
            # swift/bic is optional for some countries like SEPA countries
            payload["bic"] = swift

    if normalized_company_name:
        payload["company_name"] = normalized_company_name
    elif normalized_first_name or normalized_last_name:
        payload["individual_name"] = {  # type: ignore[assignment]
            "first_name": normalized_first_name,
            "last_name": normalized_last_name,
        }
    else:
        current_logger.warning(
            "Trying to create a counterparty with no name. This should not happen and might get flagged by Revolut's compliance team.",
            stack_info=True,
        )

    if address:
        # We don't have validation on address.country. UK is not a valid country code, it should be GB.
        # This breaks payments to Revolut, so correcting here.
        country = "GB" if address.country == "UK" else address.country

        payload["address"] = {  # type: ignore[assignment]
            "street_line1": address.street,
            "postcode": address.postal_code,
            "city": address.city,
            "country": country,
        }

    try:
        counterparty_info: CounterPartyInfo | None = self._do_create_counterparty(
            payload
        )
    except PotentialSwiftError as e:
        counterparty_info = self._retry_create_counterparty_without_bic(
            payload=payload,
            reimbursement_payment_id=reimbursement_payment_id,
        )
        if not counterparty_info:
            raise e

    current_logger.info(f"Created Revolut Counterparty {counterparty_info}.")
    return mandatory(counterparty_info)

get_refund_transactions 

get_refund_transactions(since, until=None, count=1000)
Source code in shared/services/payment_providers/revolut/revolut_client.py 
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
def get_refund_transactions(
    self,
    since: date,
    until: date | None = None,  # included
    count: int = 1000,
) -> Iterable[RevolutTransactionInfo]:
    if count > 1000:
        # https://developer.revolut.com/docs/api-reference/business/#operation/getTransactions
        raise ValueError(
            "Revolut doesn't support returning more than 1000 transactions"
        )

    if until is None:
        until = date.today()

    for start_date, end_date in months_in_interval(
        start_date=since,
        end_date=until,
    ):
        params = {
            "type": "refund",
            "count": count,
            "from": start_date.isoformat(),
            "to": end_date.isoformat(),
            "account": self._reimbursement_account_id,
        }

        request = requests.Request(
            method="GET",
            url=f"{self._base_url}{self.REFUND_TRANSACTION_ENDPOINT}",
            params=params,
            headers={"Cache-Control": "no-cache"},
        ).prepare()

        transactions = self._authenticate_and_execute_request(
            request=request,
            endpoint=self.REFUND_TRANSACTION_ENDPOINT,
        )

        current_logger.debug(
            "Call to Revolut API returned %d refund transactions", len(transactions)
        )
        for info in transactions:
            yield RevolutTransactionInfo(
                id=info["id"],
                type=RevolutTransactionType(info["type"]),
                request_id=info["request_id"],
                state=RevolutTransactionState(info["state"]),
                created_at=isoparse(info["created_at"]),
                updated_at=isoparse(info["updated_at"]),
                completed_at=(
                    isoparse(info["completed_at"])
                    if info.get("completed_at")
                    else None
                ),
                reason_code=info.get("reason_code"),
                related_transaction_id=info.get("related_transaction_id"),
                reference=info.get("reference"),
                legs=[
                    RevolutTransactionInfo.Leg(
                        leg_id=leg["leg_id"],
                        amount=leg["amount"],
                        currency=leg["currency"],
                        account_id=leg["account_id"],
                    )
                    for leg in info["legs"]
                ],
            )

get_supported_currency 

get_supported_currency()
Source code in shared/services/payment_providers/revolut/revolut_client.py 
328
329
def get_supported_currency(self) -> str:
    return current_config.get("REVOLUT_CURRENCY", "EUR")  # type: ignore[no-any-return]

get_transaction_info 

get_transaction_info(
    *,
    transaction_id=None,
    request_id=None,
    ignore_not_found_error=False
)

https://developer.revolut.com/docs/business/get-transaction ⧉

Source code in shared/services/payment_providers/revolut/revolut_client.py 
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
@retry(
    stop=stop_after_attempt(4),
    retry=retry_if_exception_type(RevolutInternalError),
    wait=wait_fixed(timedelta(milliseconds=5000)),
    reraise=True,
    before_sleep=before_sleep_log(current_logger, logging.INFO),  # type: ignore[arg-type]
)
def get_transaction_info(
    self,
    *,
    transaction_id: str | None = None,
    request_id: str | None = None,
    ignore_not_found_error: bool = False,
) -> RevolutTransactionInfo | None:
    """
    https://developer.revolut.com/docs/business/get-transaction
    """

    if transaction_id is None and request_id is None:
        raise ValueError("Either transaction_id or request_id must be provided")
    if transaction_id and request_id:
        raise ValueError(
            "Only one of transaction_id or request_id must be provided"
        )

    request = requests.Request(
        method="GET",
        url=f"{self._base_url}{self.TRANSACTION_ENDPOINT.format(transaction_id or request_id)}",
        params={"id_type": "request_id"} if request_id else None,
        headers={"Cache-Control": "no-cache"},
    ).prepare()

    try:
        info = self._authenticate_and_execute_request(
            request=request,
            endpoint=self.TRANSACTION_ENDPOINT,
            log_params={
                "transaction_id": transaction_id,
                "request_id": request_id,
            },
        )

        return RevolutTransactionInfo(
            id=info["id"],
            type=RevolutTransactionType(info["type"]),
            request_id=info["request_id"],
            state=RevolutTransactionState(info["state"]),
            created_at=isoparse(info["created_at"]),
            updated_at=isoparse(info["updated_at"]),
            completed_at=(
                isoparse(info["completed_at"]) if info.get("completed_at") else None
            ),
            reason_code=info.get("reason_code"),
            related_transaction_id=info.get("related_transaction_id"),
            reference=info.get("reference"),
            legs=[
                RevolutTransactionInfo.Leg(
                    leg_id=leg["leg_id"],
                    amount=leg["amount"],
                    currency=leg["currency"],
                    account_id=leg["account_id"],
                )
                for leg in info["legs"]
            ],
        )
    except RevolutError as e:
        if ignore_not_found_error and isinstance(e, RevolutNotFoundError):
            return None
        raise e

pay 

pay(
    *,
    request_id,
    amount_in_cents,
    payment_description=None,
    reimbursement_payment_id=None,
    revolut_counterparty_id=None,
    revolut_account_id=None,
    recipient_first_name=None,
    recipient_last_name=None,
    recipient_company_name=None,
    ibancode=None,
    swift=None,
    country_iso=None,
    address=None
)

https://developer.revolut.com/docs/business/create-transfer ⧉

Trigger a payment using Revolut.

The payment will be sent to the recipient described by revolut_counterparty_id. If this is not specified, a new counterpary will be created from the recipient_* fields.

Source code in shared/services/payment_providers/revolut/revolut_client.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
def pay(
    self,
    *,
    # The request_id is controlled by us, and is a string of 40 chars max
    request_id: str,
    amount_in_cents: int,
    payment_description: str | None = None,
    reimbursement_payment_id: str | None = None,
    revolut_counterparty_id: str | None = None,
    revolut_account_id: str | None = None,
    # -- The following fields are used to create a new counterparty if no revolut_counterparty_id is provided
    # These two fields must be specified if the recipient is an individual
    recipient_first_name: str | None = None,
    recipient_last_name: str | None = None,
    # This field must be specified if the recipient is a company or a hospital
    recipient_company_name: str | None = None,
    ibancode: str | None = None,
    swift: str | None = None,
    country_iso: str | None = None,
    address: BillingAddress | None = None,
) -> PaymentInfo:
    """
    https://developer.revolut.com/docs/business/create-transfer

    Trigger a payment using Revolut.

    The payment will be sent to the recipient described by revolut_counterparty_id. If this is not specified,
    a new counterpary will be created from the recipient_* fields.
    """
    if not isinstance(amount_in_cents, int):
        raise ValueError("amount is not int")

    if not request_id:
        raise ValueError("request_id is not set")

    if revolut_counterparty_id and revolut_account_id:
        counterparty_info = CounterPartyInfo(
            revolut_counterparty_id, revolut_account_id
        )
    elif not revolut_counterparty_id and not revolut_account_id:
        counterparty_info = self.create_counterparty(
            recipient_company_name=recipient_company_name,
            recipient_first_name=recipient_first_name,
            recipient_last_name=recipient_last_name,
            iban=mandatory(
                ibancode, "We must provide an IBAN to create a counterparty"
            ),
            swift=swift,
            country_iso=country_iso,
            address=address,
            reimbursement_payment_id=reimbursement_payment_id,
        )
    else:
        raise ValueError(
            "revolut_counterparty_id and revolut_account_id must be set or omitted together"
        )

    return self._create_payment(
        request_id,
        counterparty_info.counterparty_id,
        counterparty_info.account_id,
        amount_in_cents,
        payment_description,
    )

rotate_access_token 

rotate_access_token()
Source code in shared/services/payment_providers/revolut/revolut_client.py 
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
def rotate_access_token(self) -> None:
    data = {
        "grant_type": "refresh_token",
        "refresh_token": self._secrets_manager_client.get("refresh_token"),
        "client_id": self._client_id,
        "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
        "client_assertion": jwt.encode(
            {
                "iss": self._revolut_issuer,
                "sub": self._client_id,
                "aud": "https://revolut.com",
                "exp": datetime.utcnow() + timedelta(seconds=30),
            },
            self._private_key,
            algorithm="RS256",
            headers={"alg": "RS256", "typ": "JWT"},
        ),
    }

    request = requests.Request(
        method="POST",
        url=self._base_url + self.AUTH_ENDPOINT,
        data=data,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    ).prepare()

    current_logger.debug("Calling Revolut API: POST %s", request.url)

    # Note: request does not need to be authenticated
    response = requests.Session().send(request)
    self._raise_for_status(response, endpoint=self.AUTH_ENDPOINT)

    self._secrets_manager_client.set(
        "access_token", response.json()["access_token"]
    )