Skip to content

PyRevolut Asynchronous Client

PyRevolut provides an asynchronous client that can be used to interact with the Revolut Business API.

Example initialization of the client object:

import asyncio
from pyrevolut.client import AsyncClient

CREDS_JSON_LOC = "path/to/creds.json"

client = AsyncClient(
    creds_loc=CREDS_JSON_LOC,
    sandbox=True,
)

async def run():
  async with client:
      accounts = await client.Accounts.get_all_accounts()
      print(accounts)

asyncio.run(run())

AsyncClient

Bases: BaseClient

The asynchronous client for the Revolut API

Source code in pyrevolut/client/asynchronous.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
 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
112
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
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
class AsyncClient(BaseClient):
    """The asynchronous client for the Revolut API"""

    Accounts: EndpointAccountsAsync
    Cards: EndpointCardsAsync
    Counterparties: EndpointCounterpartiesAsync
    ForeignExchange: EndpointForeignExchangeAsync
    PaymentDrafts: EndpointPaymentDraftsAsync
    PayoutLinks: EndpointPayoutLinksAsync
    Simulations: EndpointSimulationsAsync
    TeamMembers: EndpointTeamMembersAsync
    Transactions: EndpointTransactionsAsync
    Transfers: EndpointTransfersAsync
    Webhooks: EndpointWebhooksAsync

    async def open(self):
        """Opens the client connection"""
        if self.client is not None:
            return

        self.client = HTTPClient()

    async def close(self):
        """Closes the client connection"""
        if self.client is None:
            return

        await self.client.aclose()
        self.client = None

    async def get(
        self,
        path: str,
        response_model: Type[BaseModel],
        params: Type[BaseModel] | None = None,
        **kwargs,
    ):
        """Send an async GET request to the Revolut API

        Parameters
        ----------
        path : str
            The path to send the request to
        response_model : Type[BaseModel]
            The model to use for the response
        params : Type[BaseModel] | None
            The parameters to send in the request

        Returns
        -------
        Response
            The response from the request
        """
        resp = await self.client.get(
            **self._prep_get(
                path=path,
                params=params,
                **kwargs,
            )
        )
        return self.process_response(
            response=resp,
            response_model=response_model,
            return_type=None,
            error_response=None,
        )

    async def post(
        self,
        path: str,
        response_model: Type[BaseModel],
        body: Type[BaseModel] | None = None,
        **kwargs,
    ):
        """Send an async POST request to the Revolut API

        Parameters
        ----------
        path : str
            The path to send the request to
        response_model : Type[BaseModel]
            The model to use for the response
        body : Type[BaseModel] | None
            The body to send in the request

        Returns
        -------
        Response
            The response from the request
        """
        resp = await self.client.post(
            **self._prep_post(
                path=path,
                body=body,
                **kwargs,
            )
        )
        return self.process_response(
            response=resp,
            response_model=response_model,
            return_type=None,
            error_response=None,
        )

    async def patch(
        self,
        path: str,
        response_model: Type[BaseModel],
        body: Type[BaseModel] | None = None,
        **kwargs,
    ):
        """Send an async PATCH request to the Revolut API

        Parameters
        ----------
        path : str
            The path to send the request to
        response_model : Type[BaseModel]
            The model to use for the response
        body : Type[BaseModel] | None
            The body to send in the request

        Returns
        -------
        Response
            The response from the request
        """
        resp = await self.client.patch(
            **self._prep_patch(
                path=path,
                body=body,
                **kwargs,
            )
        )
        return self.process_response(
            response=resp,
            response_model=response_model,
            return_type=None,
            error_response=None,
        )

    async def delete(
        self,
        path: str,
        response_model: Type[BaseModel],
        params: Type[BaseModel] | None = None,
        **kwargs,
    ):
        """Send an async DELETE request to the Revolut API

        Parameters
        ----------
        path : str
            The path to send the request to
        response_model : Type[BaseModel]
            The model to use for the response
        params : Type[BaseModel] | None
            The parameters to add to the request route

        Returns
        -------
        Response
            The response from the request
        """
        resp = await self.client.delete(
            **self._prep_delete(
                path=path,
                params=params,
                **kwargs,
            )
        )
        return self.process_response(
            response=resp,
            response_model=response_model,
            return_type=None,
            error_response=None,
        )

    async def put(
        self,
        path: str,
        response_model: Type[BaseModel],
        body: Type[BaseModel] | None = None,
        **kwargs,
    ):
        """Send an async PUT request to the Revolut API

        Parameters
        ----------
        path : str
            The path to send the request to
        response_model : Type[BaseModel]
            The model to use for the response
        body : Type[BaseModel] | None
            The body to send in the request

        Returns
        -------
        Response
            The response from the request
        """
        resp = await self.client.put(
            **self._prep_put(
                path=path,
                body=body,
                **kwargs,
            )
        )
        return self.process_response(
            response=resp,
            response_model=response_model,
            return_type=None,
            error_response=None,
        )

    def load_endpoints(self):
        """Loads all the endpoints from the api directory"""
        self.Accounts = EndpointAccountsAsync(client=self)
        self.Cards = EndpointCardsAsync(client=self)
        self.Counterparties = EndpointCounterpartiesAsync(client=self)
        self.ForeignExchange = EndpointForeignExchangeAsync(client=self)
        self.PaymentDrafts = EndpointPaymentDraftsAsync(client=self)
        self.PayoutLinks = EndpointPayoutLinksAsync(client=self)
        self.Simulations = EndpointSimulationsAsync(client=self)
        self.TeamMembers = EndpointTeamMembersAsync(client=self)
        self.Transactions = EndpointTransactionsAsync(client=self)
        self.Transfers = EndpointTransfersAsync(client=self)
        self.Webhooks = EndpointWebhooksAsync(client=self)

    async def __aenter__(self):
        """Open the async client connection"""
        await self.open()
        return self

    async def __aexit__(self, *args, **kwargs):
        """Close the async client connection"""
        await self.close()

__aenter__() async

Open the async client connection

Source code in pyrevolut/client/asynchronous.py 
254
255
256
257
async def __aenter__(self):
    """Open the async client connection"""
    await self.open()
    return self

__aexit__(*args, **kwargs) async

Close the async client connection

Source code in pyrevolut/client/asynchronous.py 
259
260
261
async def __aexit__(self, *args, **kwargs):
    """Close the async client connection"""
    await self.close()

close() async

Closes the client connection

Source code in pyrevolut/client/asynchronous.py 
47
48
49
50
51
52
53
async def close(self):
    """Closes the client connection"""
    if self.client is None:
        return

    await self.client.aclose()
    self.client = None

delete(path, response_model, params=None, **kwargs) async

Send an async DELETE request to the Revolut API

Parameters:

Name Type Description Default
path str

The path to send the request to

 required
response_model Type[BaseModel]

The model to use for the response

 required
params Type[BaseModel] | None

The parameters to add to the request route

 None

Returns:

Type Description
Response

The response from the request
Source code in pyrevolut/client/asynchronous.py 
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
async def delete(
    self,
    path: str,
    response_model: Type[BaseModel],
    params: Type[BaseModel] | None = None,
    **kwargs,
):
    """Send an async DELETE request to the Revolut API

    Parameters
    ----------
    path : str
        The path to send the request to
    response_model : Type[BaseModel]
        The model to use for the response
    params : Type[BaseModel] | None
        The parameters to add to the request route

    Returns
    -------
    Response
        The response from the request
    """
    resp = await self.client.delete(
        **self._prep_delete(
            path=path,
            params=params,
            **kwargs,
        )
    )
    return self.process_response(
        response=resp,
        response_model=response_model,
        return_type=None,
        error_response=None,
    )

get(path, response_model, params=None, **kwargs) async

Send an async GET request to the Revolut API

Parameters:

Name Type Description Default
path str

The path to send the request to

 required
response_model Type[BaseModel]

The model to use for the response

 required
params Type[BaseModel] | None

The parameters to send in the request

 None

Returns:

Type Description
Response

The response from the request
Source code in pyrevolut/client/asynchronous.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
async def get(
    self,
    path: str,
    response_model: Type[BaseModel],
    params: Type[BaseModel] | None = None,
    **kwargs,
):
    """Send an async GET request to the Revolut API

    Parameters
    ----------
    path : str
        The path to send the request to
    response_model : Type[BaseModel]
        The model to use for the response
    params : Type[BaseModel] | None
        The parameters to send in the request

    Returns
    -------
    Response
        The response from the request
    """
    resp = await self.client.get(
        **self._prep_get(
            path=path,
            params=params,
            **kwargs,
        )
    )
    return self.process_response(
        response=resp,
        response_model=response_model,
        return_type=None,
        error_response=None,
    )

load_endpoints()

Loads all the endpoints from the api directory

Source code in pyrevolut/client/asynchronous.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
def load_endpoints(self):
    """Loads all the endpoints from the api directory"""
    self.Accounts = EndpointAccountsAsync(client=self)
    self.Cards = EndpointCardsAsync(client=self)
    self.Counterparties = EndpointCounterpartiesAsync(client=self)
    self.ForeignExchange = EndpointForeignExchangeAsync(client=self)
    self.PaymentDrafts = EndpointPaymentDraftsAsync(client=self)
    self.PayoutLinks = EndpointPayoutLinksAsync(client=self)
    self.Simulations = EndpointSimulationsAsync(client=self)
    self.TeamMembers = EndpointTeamMembersAsync(client=self)
    self.Transactions = EndpointTransactionsAsync(client=self)
    self.Transfers = EndpointTransfersAsync(client=self)
    self.Webhooks = EndpointWebhooksAsync(client=self)

open() async

Opens the client connection

Source code in pyrevolut/client/asynchronous.py 
40
41
42
43
44
45
async def open(self):
    """Opens the client connection"""
    if self.client is not None:
        return

    self.client = HTTPClient()

patch(path, response_model, body=None, **kwargs) async

Send an async PATCH request to the Revolut API

Parameters:

Name Type Description Default
path str

The path to send the request to

 required
response_model Type[BaseModel]

The model to use for the response

 required
body Type[BaseModel] | None

The body to send in the request

 None

Returns:

Type Description
Response

The response from the request
Source code in pyrevolut/client/asynchronous.py 
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
async def patch(
    self,
    path: str,
    response_model: Type[BaseModel],
    body: Type[BaseModel] | None = None,
    **kwargs,
):
    """Send an async PATCH request to the Revolut API

    Parameters
    ----------
    path : str
        The path to send the request to
    response_model : Type[BaseModel]
        The model to use for the response
    body : Type[BaseModel] | None
        The body to send in the request

    Returns
    -------
    Response
        The response from the request
    """
    resp = await self.client.patch(
        **self._prep_patch(
            path=path,
            body=body,
            **kwargs,
        )
    )
    return self.process_response(
        response=resp,
        response_model=response_model,
        return_type=None,
        error_response=None,
    )

post(path, response_model, body=None, **kwargs) async

Send an async POST request to the Revolut API

Parameters:

Name Type Description Default
path str

The path to send the request to

 required
response_model Type[BaseModel]

The model to use for the response

 required
body Type[BaseModel] | None

The body to send in the request

 None

Returns:

Type Description
Response

The response from the request
Source code in pyrevolut/client/asynchronous.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def post(
    self,
    path: str,
    response_model: Type[BaseModel],
    body: Type[BaseModel] | None = None,
    **kwargs,
):
    """Send an async POST request to the Revolut API

    Parameters
    ----------
    path : str
        The path to send the request to
    response_model : Type[BaseModel]
        The model to use for the response
    body : Type[BaseModel] | None
        The body to send in the request

    Returns
    -------
    Response
        The response from the request
    """
    resp = await self.client.post(
        **self._prep_post(
            path=path,
            body=body,
            **kwargs,
        )
    )
    return self.process_response(
        response=resp,
        response_model=response_model,
        return_type=None,
        error_response=None,
    )

put(path, response_model, body=None, **kwargs) async

Send an async PUT request to the Revolut API

Parameters:

Name Type Description Default
path str

The path to send the request to

 required
response_model Type[BaseModel]

The model to use for the response

 required
body Type[BaseModel] | None

The body to send in the request

 None

Returns:

Type Description
Response

The response from the request
Source code in pyrevolut/client/asynchronous.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
async def put(
    self,
    path: str,
    response_model: Type[BaseModel],
    body: Type[BaseModel] | None = None,
    **kwargs,
):
    """Send an async PUT request to the Revolut API

    Parameters
    ----------
    path : str
        The path to send the request to
    response_model : Type[BaseModel]
        The model to use for the response
    body : Type[BaseModel] | None
        The body to send in the request

    Returns
    -------
    Response
        The response from the request
    """
    resp = await self.client.put(
        **self._prep_put(
            path=path,
            body=body,
            **kwargs,
        )
    )
    return self.process_response(
        response=resp,
        response_model=response_model,
        return_type=None,
        error_response=None,
    )