deepgram.clients.common.v1.abstract_async_rest

View Source

  1# Copyright 2023-2024 Deepgram SDK contributors. All Rights Reserved.
  2# Use of this source code is governed by a MIT license that can be found in the LICENSE file.
  3# SPDX-License-Identifier: MIT
  4
  5import json
  6import io
  7from typing import Dict, Optional, List, Union
  8
  9import httpx
 10
 11from .helpers import append_query_params
 12from ....options import DeepgramClientOptions
 13from .errors import DeepgramError, DeepgramApiError, DeepgramUnknownApiError
 14
 15
 16class AbstractAsyncRestClient:
 17    """
 18    An abstract base class for a RESTful HTTP client.
 19
 20    This class provides common HTTP methods (GET, POST, PUT, PATCH, DELETE) for making asynchronous HTTP requests.
 21    It handles error responses and provides basic JSON parsing.
 22
 23    Args:
 24        url (Dict): The base URL for the RESTful API, including any path segments.
 25        headers (Optional[Dict[str, Any]]): Optional HTTP headers to include in requests.
 26        params (Optional[Dict[str, Any]]): Optional query parameters to include in requests.
 27        timeout (Optional[httpx.Timeout]): Optional timeout configuration for requests.
 28
 29    Exceptions:
 30        DeepgramApiError: Raised for known API errors.
 31        DeepgramUnknownApiError: Raised for unknown API errors.
 32    """
 33
 34    _config: DeepgramClientOptions
 35
 36    def __init__(self, config: DeepgramClientOptions):
 37        if config is None:
 38            raise DeepgramError("Config are required")
 39        self._config = config
 40
 41    # pylint: disable=too-many-positional-arguments
 42
 43    async def get(
 44        self,
 45        url: str,
 46        options: Optional[Dict] = None,
 47        addons: Optional[Dict] = None,
 48        headers: Optional[Dict] = None,
 49        timeout: Optional[httpx.Timeout] = None,
 50        **kwargs,
 51    ) -> str:
 52        """
 53        Make a GET request to the specified URL.
 54        """
 55        return await self._handle_request(
 56            "GET",
 57            url,
 58            params=options,
 59            addons=addons,
 60            headers=headers,
 61            timeout=timeout,
 62            **kwargs,
 63        )
 64
 65    async def post_raw(
 66        self,
 67        url: str,
 68        options: Optional[Dict] = None,
 69        addons: Optional[Dict] = None,
 70        headers: Optional[Dict] = None,
 71        timeout: Optional[httpx.Timeout] = None,
 72        **kwargs,
 73    ) -> httpx.Response:
 74        """
 75        Make a POST request to the specified URL and return response in raw bytes.
 76        """
 77        return await self._handle_request_raw(
 78            "POST",
 79            url,
 80            params=options,
 81            addons=addons,
 82            headers=headers,
 83            timeout=timeout,
 84            **kwargs,
 85        )
 86
 87    async def post_memory(
 88        self,
 89        url: str,
 90        file_result: List,
 91        options: Optional[Dict] = None,
 92        addons: Optional[Dict] = None,
 93        headers: Optional[Dict] = None,
 94        timeout: Optional[httpx.Timeout] = None,
 95        **kwargs,
 96    ) -> Dict[str, Union[str, io.BytesIO]]:
 97        """
 98        Make a POST request to the specified URL and return response in memory.
 99        """
100        return await self._handle_request_memory(
101            "POST",
102            url,
103            file_result=file_result,
104            params=options,
105            addons=addons,
106            headers=headers,
107            timeout=timeout,
108            **kwargs,
109        )
110
111    async def post(
112        self,
113        url: str,
114        options: Optional[Dict] = None,
115        addons: Optional[Dict] = None,
116        headers: Optional[Dict] = None,
117        timeout: Optional[httpx.Timeout] = None,
118        **kwargs,
119    ) -> str:
120        """
121        Make a POST request to the specified URL.
122        """
123        return await self._handle_request(
124            "POST",
125            url,
126            params=options,
127            addons=addons,
128            headers=headers,
129            timeout=timeout,
130            **kwargs,
131        )
132
133    async def put(
134        self,
135        url: str,
136        options: Optional[Dict] = None,
137        addons: Optional[Dict] = None,
138        headers: Optional[Dict] = None,
139        timeout: Optional[httpx.Timeout] = None,
140        **kwargs,
141    ) -> str:
142        """
143        Make a PUT request to the specified URL.
144        """
145        return await self._handle_request(
146            "PUT",
147            url,
148            params=options,
149            addons=addons,
150            headers=headers,
151            timeout=timeout,
152            **kwargs,
153        )
154
155    async def patch(
156        self,
157        url: str,
158        options: Optional[Dict] = None,
159        addons: Optional[Dict] = None,
160        headers: Optional[Dict] = None,
161        timeout: Optional[httpx.Timeout] = None,
162        **kwargs,
163    ) -> str:
164        """
165        Make a PATCH request to the specified URL.
166        """
167        return await self._handle_request(
168            "PATCH",
169            url,
170            params=options,
171            addons=addons,
172            headers=headers,
173            timeout=timeout,
174            **kwargs,
175        )
176
177    async def delete(
178        self,
179        url: str,
180        options: Optional[Dict] = None,
181        addons: Optional[Dict] = None,
182        headers: Optional[Dict] = None,
183        timeout: Optional[httpx.Timeout] = None,
184        **kwargs,
185    ) -> str:
186        """
187        Make a DELETE request to the specified URL.
188        """
189        return await self._handle_request(
190            "DELETE",
191            url,
192            params=options,
193            addons=addons,
194            headers=headers,
195            timeout=timeout,
196            **kwargs,
197        )
198
199    # pylint: disable-msg=too-many-locals,too-many-branches,too-many-locals
200    async def _handle_request(
201        self,
202        method: str,
203        url: str,
204        params: Optional[Dict] = None,
205        addons: Optional[Dict] = None,
206        headers: Optional[Dict] = None,
207        timeout: Optional[httpx.Timeout] = None,
208        **kwargs,
209    ) -> str:
210        _url = url
211        if params is not None:
212            _url = append_query_params(_url, params)
213        if addons is not None:
214            _url = append_query_params(_url, addons)
215        _headers = self._config.headers
216        if headers is not None:
217            _headers.update(headers)
218        if timeout is None:
219            timeout = httpx.Timeout(30.0, connect=10.0)
220
221        try:
222            transport = kwargs.get("transport")
223            async with httpx.AsyncClient(
224                timeout=timeout, transport=transport
225            ) as client:
226                if transport:
227                    kwargs.pop("transport")
228                response = await client.request(
229                    method, _url, headers=_headers, **kwargs
230                )
231                response.raise_for_status()
232
233                # throw exception if response is None or response.text is None
234                if response is None or response.text is None:
235                    raise DeepgramError(
236                        "Response is not available yet. Please try again later."
237                    )
238
239                return response.text
240
241        except httpx.HTTPError as e1:
242            if isinstance(e1, httpx.HTTPStatusError):
243                status_code = e1.response.status_code or 500
244                try:
245                    json_object = json.loads(e1.response.text)
246                    raise DeepgramApiError(
247                        json_object.get("err_msg"),
248                        str(status_code),
249                        json.dumps(json_object),
250                    ) from e1
251                except json.decoder.JSONDecodeError as e2:
252                    raise DeepgramUnknownApiError(e2.msg, str(status_code)) from e2
253                except ValueError as e2:
254                    raise DeepgramUnknownApiError(str(e2), str(status_code)) from e2
255            else:
256                raise  # pylint: disable-msg=try-except-raise
257        except Exception:  # pylint: disable-msg=try-except-raise
258            raise
259
260    # pylint: enable-msg=too-many-locals,too-many-branches,too-many-locals
261
262    # pylint: disable-msg=too-many-locals,too-many-branches
263    async def _handle_request_memory(
264        self,
265        method: str,
266        url: str,
267        file_result: List,
268        params: Optional[Dict] = None,
269        addons: Optional[Dict] = None,
270        headers: Optional[Dict] = None,
271        timeout: Optional[httpx.Timeout] = None,
272        **kwargs,
273    ) -> Dict[str, Union[str, io.BytesIO]]:
274        _url = url
275        if params is not None:
276            _url = append_query_params(_url, params)
277        if addons is not None:
278            _url = append_query_params(_url, addons)
279        _headers = self._config.headers
280        if headers is not None:
281            _headers.update(headers)
282        if timeout is None:
283            timeout = httpx.Timeout(30.0, connect=10.0)
284
285        try:
286            transport = kwargs.get("transport")
287            async with httpx.AsyncClient(
288                timeout=timeout, transport=transport
289            ) as client:
290                if transport:
291                    kwargs.pop("transport")
292                response = await client.request(
293                    method, _url, headers=_headers, **kwargs
294                )
295                response.raise_for_status()
296
297                ret: Dict[str, Union[str, io.BytesIO]] = {}
298                for item in file_result:
299                    if item in response.headers:
300                        ret[item] = response.headers[item]
301                        continue
302                    tmp_item = f"dg-{item}"
303                    if tmp_item in response.headers:
304                        ret[item] = response.headers[tmp_item]
305                        continue
306                    tmp_item = f"x-dg-{item}"
307                    if tmp_item in response.headers:
308                        ret[item] = response.headers[tmp_item]
309                ret["stream"] = io.BytesIO(response.content)
310                return ret
311
312        except httpx.HTTPError as e1:
313            if isinstance(e1, httpx.HTTPStatusError):
314                status_code = e1.response.status_code or 500
315                try:
316                    json_object = json.loads(e1.response.text)
317                    raise DeepgramApiError(
318                        json_object.get("err_msg"),
319                        str(status_code),
320                        json.dumps(json_object),
321                    ) from e1
322                except json.decoder.JSONDecodeError as e2:
323                    raise DeepgramUnknownApiError(e2.msg, str(status_code)) from e2
324                except ValueError as e2:
325                    raise DeepgramUnknownApiError(str(e2), str(status_code)) from e2
326            else:
327                raise  # pylint: disable-msg=try-except-raise
328        except Exception:  # pylint: disable-msg=try-except-raise
329            raise
330
331    # pylint: enable-msg=too-many-locals,too-many-branches
332
333    # pylint: disable-msg=too-many-locals,too-many-branches
334    async def _handle_request_raw(
335        self,
336        method: str,
337        url: str,
338        params: Optional[Dict] = None,
339        addons: Optional[Dict] = None,
340        headers: Optional[Dict] = None,
341        timeout: Optional[httpx.Timeout] = None,
342        **kwargs,
343    ) -> httpx.Response:
344        _url = url
345        if params is not None:
346            _url = append_query_params(_url, params)
347        if addons is not None:
348            _url = append_query_params(_url, addons)
349        _headers = self._config.headers
350        if headers is not None:
351            _headers.update(headers)
352        if timeout is None:
353            timeout = httpx.Timeout(30.0, connect=10.0)
354
355        try:
356            transport = kwargs.get("transport")
357            client = httpx.AsyncClient(timeout=timeout, transport=transport)
358            if transport:
359                kwargs.pop("transport")
360            req = client.build_request(method, _url, headers=_headers, **kwargs)
361            return await client.send(req, stream=True)
362
363        except httpx.HTTPError as e1:
364            if isinstance(e1, httpx.HTTPStatusError):
365                status_code = e1.response.status_code or 500
366                try:
367                    json_object = json.loads(e1.response.text)
368                    raise DeepgramApiError(
369                        json_object.get("err_msg"),
370                        str(status_code),
371                        json.dumps(json_object),
372                    ) from e1
373                except json.decoder.JSONDecodeError as e2:
374                    raise DeepgramUnknownApiError(e2.msg, str(status_code)) from e2
375                except ValueError as e2:
376                    raise DeepgramUnknownApiError(str(e2), str(status_code)) from e2
377            else:
378                raise  # pylint: disable-msg=try-except-raise
379        except Exception:  # pylint: disable-msg=try-except-raise
380            raise
381
382    # pylint: enable-msg=too-many-locals,too-many-branches
383    # pylint: enable=too-many-positional-arguments

class AbstractAsyncRestClient: View Source

 17class AbstractAsyncRestClient:
 18    """
 19    An abstract base class for a RESTful HTTP client.
 20
 21    This class provides common HTTP methods (GET, POST, PUT, PATCH, DELETE) for making asynchronous HTTP requests.
 22    It handles error responses and provides basic JSON parsing.
 23
 24    Args:
 25        url (Dict): The base URL for the RESTful API, including any path segments.
 26        headers (Optional[Dict[str, Any]]): Optional HTTP headers to include in requests.
 27        params (Optional[Dict[str, Any]]): Optional query parameters to include in requests.
 28        timeout (Optional[httpx.Timeout]): Optional timeout configuration for requests.
 29
 30    Exceptions:
 31        DeepgramApiError: Raised for known API errors.
 32        DeepgramUnknownApiError: Raised for unknown API errors.
 33    """
 34
 35    _config: DeepgramClientOptions
 36
 37    def __init__(self, config: DeepgramClientOptions):
 38        if config is None:
 39            raise DeepgramError("Config are required")
 40        self._config = config
 41
 42    # pylint: disable=too-many-positional-arguments
 43
 44    async def get(
 45        self,
 46        url: str,
 47        options: Optional[Dict] = None,
 48        addons: Optional[Dict] = None,
 49        headers: Optional[Dict] = None,
 50        timeout: Optional[httpx.Timeout] = None,
 51        **kwargs,
 52    ) -> str:
 53        """
 54        Make a GET request to the specified URL.
 55        """
 56        return await self._handle_request(
 57            "GET",
 58            url,
 59            params=options,
 60            addons=addons,
 61            headers=headers,
 62            timeout=timeout,
 63            **kwargs,
 64        )
 65
 66    async def post_raw(
 67        self,
 68        url: str,
 69        options: Optional[Dict] = None,
 70        addons: Optional[Dict] = None,
 71        headers: Optional[Dict] = None,
 72        timeout: Optional[httpx.Timeout] = None,
 73        **kwargs,
 74    ) -> httpx.Response:
 75        """
 76        Make a POST request to the specified URL and return response in raw bytes.
 77        """
 78        return await self._handle_request_raw(
 79            "POST",
 80            url,
 81            params=options,
 82            addons=addons,
 83            headers=headers,
 84            timeout=timeout,
 85            **kwargs,
 86        )
 87
 88    async def post_memory(
 89        self,
 90        url: str,
 91        file_result: List,
 92        options: Optional[Dict] = None,
 93        addons: Optional[Dict] = None,
 94        headers: Optional[Dict] = None,
 95        timeout: Optional[httpx.Timeout] = None,
 96        **kwargs,
 97    ) -> Dict[str, Union[str, io.BytesIO]]:
 98        """
 99        Make a POST request to the specified URL and return response in memory.
100        """
101        return await self._handle_request_memory(
102            "POST",
103            url,
104            file_result=file_result,
105            params=options,
106            addons=addons,
107            headers=headers,
108            timeout=timeout,
109            **kwargs,
110        )
111
112    async def post(
113        self,
114        url: str,
115        options: Optional[Dict] = None,
116        addons: Optional[Dict] = None,
117        headers: Optional[Dict] = None,
118        timeout: Optional[httpx.Timeout] = None,
119        **kwargs,
120    ) -> str:
121        """
122        Make a POST request to the specified URL.
123        """
124        return await self._handle_request(
125            "POST",
126            url,
127            params=options,
128            addons=addons,
129            headers=headers,
130            timeout=timeout,
131            **kwargs,
132        )
133
134    async def put(
135        self,
136        url: str,
137        options: Optional[Dict] = None,
138        addons: Optional[Dict] = None,
139        headers: Optional[Dict] = None,
140        timeout: Optional[httpx.Timeout] = None,
141        **kwargs,
142    ) -> str:
143        """
144        Make a PUT request to the specified URL.
145        """
146        return await self._handle_request(
147            "PUT",
148            url,
149            params=options,
150            addons=addons,
151            headers=headers,
152            timeout=timeout,
153            **kwargs,
154        )
155
156    async def patch(
157        self,
158        url: str,
159        options: Optional[Dict] = None,
160        addons: Optional[Dict] = None,
161        headers: Optional[Dict] = None,
162        timeout: Optional[httpx.Timeout] = None,
163        **kwargs,
164    ) -> str:
165        """
166        Make a PATCH request to the specified URL.
167        """
168        return await self._handle_request(
169            "PATCH",
170            url,
171            params=options,
172            addons=addons,
173            headers=headers,
174            timeout=timeout,
175            **kwargs,
176        )
177
178    async def delete(
179        self,
180        url: str,
181        options: Optional[Dict] = None,
182        addons: Optional[Dict] = None,
183        headers: Optional[Dict] = None,
184        timeout: Optional[httpx.Timeout] = None,
185        **kwargs,
186    ) -> str:
187        """
188        Make a DELETE request to the specified URL.
189        """
190        return await self._handle_request(
191            "DELETE",
192            url,
193            params=options,
194            addons=addons,
195            headers=headers,
196            timeout=timeout,
197            **kwargs,
198        )
199
200    # pylint: disable-msg=too-many-locals,too-many-branches,too-many-locals
201    async def _handle_request(
202        self,
203        method: str,
204        url: str,
205        params: Optional[Dict] = None,
206        addons: Optional[Dict] = None,
207        headers: Optional[Dict] = None,
208        timeout: Optional[httpx.Timeout] = None,
209        **kwargs,
210    ) -> str:
211        _url = url
212        if params is not None:
213            _url = append_query_params(_url, params)
214        if addons is not None:
215            _url = append_query_params(_url, addons)
216        _headers = self._config.headers
217        if headers is not None:
218            _headers.update(headers)
219        if timeout is None:
220            timeout = httpx.Timeout(30.0, connect=10.0)
221
222        try:
223            transport = kwargs.get("transport")
224            async with httpx.AsyncClient(
225                timeout=timeout, transport=transport
226            ) as client:
227                if transport:
228                    kwargs.pop("transport")
229                response = await client.request(
230                    method, _url, headers=_headers, **kwargs
231                )
232                response.raise_for_status()
233
234                # throw exception if response is None or response.text is None
235                if response is None or response.text is None:
236                    raise DeepgramError(
237                        "Response is not available yet. Please try again later."
238                    )
239
240                return response.text
241
242        except httpx.HTTPError as e1:
243            if isinstance(e1, httpx.HTTPStatusError):
244                status_code = e1.response.status_code or 500
245                try:
246                    json_object = json.loads(e1.response.text)
247                    raise DeepgramApiError(
248                        json_object.get("err_msg"),
249                        str(status_code),
250                        json.dumps(json_object),
251                    ) from e1
252                except json.decoder.JSONDecodeError as e2:
253                    raise DeepgramUnknownApiError(e2.msg, str(status_code)) from e2
254                except ValueError as e2:
255                    raise DeepgramUnknownApiError(str(e2), str(status_code)) from e2
256            else:
257                raise  # pylint: disable-msg=try-except-raise
258        except Exception:  # pylint: disable-msg=try-except-raise
259            raise
260
261    # pylint: enable-msg=too-many-locals,too-many-branches,too-many-locals
262
263    # pylint: disable-msg=too-many-locals,too-many-branches
264    async def _handle_request_memory(
265        self,
266        method: str,
267        url: str,
268        file_result: List,
269        params: Optional[Dict] = None,
270        addons: Optional[Dict] = None,
271        headers: Optional[Dict] = None,
272        timeout: Optional[httpx.Timeout] = None,
273        **kwargs,
274    ) -> Dict[str, Union[str, io.BytesIO]]:
275        _url = url
276        if params is not None:
277            _url = append_query_params(_url, params)
278        if addons is not None:
279            _url = append_query_params(_url, addons)
280        _headers = self._config.headers
281        if headers is not None:
282            _headers.update(headers)
283        if timeout is None:
284            timeout = httpx.Timeout(30.0, connect=10.0)
285
286        try:
287            transport = kwargs.get("transport")
288            async with httpx.AsyncClient(
289                timeout=timeout, transport=transport
290            ) as client:
291                if transport:
292                    kwargs.pop("transport")
293                response = await client.request(
294                    method, _url, headers=_headers, **kwargs
295                )
296                response.raise_for_status()
297
298                ret: Dict[str, Union[str, io.BytesIO]] = {}
299                for item in file_result:
300                    if item in response.headers:
301                        ret[item] = response.headers[item]
302                        continue
303                    tmp_item = f"dg-{item}"
304                    if tmp_item in response.headers:
305                        ret[item] = response.headers[tmp_item]
306                        continue
307                    tmp_item = f"x-dg-{item}"
308                    if tmp_item in response.headers:
309                        ret[item] = response.headers[tmp_item]
310                ret["stream"] = io.BytesIO(response.content)
311                return ret
312
313        except httpx.HTTPError as e1:
314            if isinstance(e1, httpx.HTTPStatusError):
315                status_code = e1.response.status_code or 500
316                try:
317                    json_object = json.loads(e1.response.text)
318                    raise DeepgramApiError(
319                        json_object.get("err_msg"),
320                        str(status_code),
321                        json.dumps(json_object),
322                    ) from e1
323                except json.decoder.JSONDecodeError as e2:
324                    raise DeepgramUnknownApiError(e2.msg, str(status_code)) from e2
325                except ValueError as e2:
326                    raise DeepgramUnknownApiError(str(e2), str(status_code)) from e2
327            else:
328                raise  # pylint: disable-msg=try-except-raise
329        except Exception:  # pylint: disable-msg=try-except-raise
330            raise
331
332    # pylint: enable-msg=too-many-locals,too-many-branches
333
334    # pylint: disable-msg=too-many-locals,too-many-branches
335    async def _handle_request_raw(
336        self,
337        method: str,
338        url: str,
339        params: Optional[Dict] = None,
340        addons: Optional[Dict] = None,
341        headers: Optional[Dict] = None,
342        timeout: Optional[httpx.Timeout] = None,
343        **kwargs,
344    ) -> httpx.Response:
345        _url = url
346        if params is not None:
347            _url = append_query_params(_url, params)
348        if addons is not None:
349            _url = append_query_params(_url, addons)
350        _headers = self._config.headers
351        if headers is not None:
352            _headers.update(headers)
353        if timeout is None:
354            timeout = httpx.Timeout(30.0, connect=10.0)
355
356        try:
357            transport = kwargs.get("transport")
358            client = httpx.AsyncClient(timeout=timeout, transport=transport)
359            if transport:
360                kwargs.pop("transport")
361            req = client.build_request(method, _url, headers=_headers, **kwargs)
362            return await client.send(req, stream=True)
363
364        except httpx.HTTPError as e1:
365            if isinstance(e1, httpx.HTTPStatusError):
366                status_code = e1.response.status_code or 500
367                try:
368                    json_object = json.loads(e1.response.text)
369                    raise DeepgramApiError(
370                        json_object.get("err_msg"),
371                        str(status_code),
372                        json.dumps(json_object),
373                    ) from e1
374                except json.decoder.JSONDecodeError as e2:
375                    raise DeepgramUnknownApiError(e2.msg, str(status_code)) from e2
376                except ValueError as e2:
377                    raise DeepgramUnknownApiError(str(e2), str(status_code)) from e2
378            else:
379                raise  # pylint: disable-msg=try-except-raise
380        except Exception:  # pylint: disable-msg=try-except-raise
381            raise
382
383    # pylint: enable-msg=too-many-locals,too-many-branches
384    # pylint: enable=too-many-positional-arguments

An abstract base class for a RESTful HTTP client.

This class provides common HTTP methods (GET, POST, PUT, PATCH, DELETE) for making asynchronous HTTP requests. It handles error responses and provides basic JSON parsing.

Args: url (Dict): The base URL for the RESTful API, including any path segments. headers (Optional[Dict[str, Any]]): Optional HTTP headers to include in requests. params (Optional[Dict[str, Any]]): Optional query parameters to include in requests. timeout (Optional[httpx.Timeout]): Optional timeout configuration for requests.

Exceptions: DeepgramApiError: Raised for known API errors. DeepgramUnknownApiError: Raised for unknown API errors.

AbstractAsyncRestClient(config: deepgram.options.DeepgramClientOptions) View Source

37    def __init__(self, config: DeepgramClientOptions):
38        if config is None:
39            raise DeepgramError("Config are required")
40        self._config = config

async def get( self, url: str, options: Optional[Dict] = None, addons: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: Optional[httpx.Timeout] = None, **kwargs) -> str: View Source

44    async def get(
45        self,
46        url: str,
47        options: Optional[Dict] = None,
48        addons: Optional[Dict] = None,
49        headers: Optional[Dict] = None,
50        timeout: Optional[httpx.Timeout] = None,
51        **kwargs,
52    ) -> str:
53        """
54        Make a GET request to the specified URL.
55        """
56        return await self._handle_request(
57            "GET",
58            url,
59            params=options,
60            addons=addons,
61            headers=headers,
62            timeout=timeout,
63            **kwargs,
64        )

Make a GET request to the specified URL.

async def post_raw( self, url: str, options: Optional[Dict] = None, addons: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: Optional[httpx.Timeout] = None, **kwargs) -> httpx.Response: View Source

66    async def post_raw(
67        self,
68        url: str,
69        options: Optional[Dict] = None,
70        addons: Optional[Dict] = None,
71        headers: Optional[Dict] = None,
72        timeout: Optional[httpx.Timeout] = None,
73        **kwargs,
74    ) -> httpx.Response:
75        """
76        Make a POST request to the specified URL and return response in raw bytes.
77        """
78        return await self._handle_request_raw(
79            "POST",
80            url,
81            params=options,
82            addons=addons,
83            headers=headers,
84            timeout=timeout,
85            **kwargs,
86        )

Make a POST request to the specified URL and return response in raw bytes.

async def post_memory( self, url: str, file_result: List, options: Optional[Dict] = None, addons: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: Optional[httpx.Timeout] = None, **kwargs) -> Dict[str, Union[str, _io.BytesIO]]: View Source

 88    async def post_memory(
 89        self,
 90        url: str,
 91        file_result: List,
 92        options: Optional[Dict] = None,
 93        addons: Optional[Dict] = None,
 94        headers: Optional[Dict] = None,
 95        timeout: Optional[httpx.Timeout] = None,
 96        **kwargs,
 97    ) -> Dict[str, Union[str, io.BytesIO]]:
 98        """
 99        Make a POST request to the specified URL and return response in memory.
100        """
101        return await self._handle_request_memory(
102            "POST",
103            url,
104            file_result=file_result,
105            params=options,
106            addons=addons,
107            headers=headers,
108            timeout=timeout,
109            **kwargs,
110        )

Make a POST request to the specified URL and return response in memory.

async def post( self, url: str, options: Optional[Dict] = None, addons: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: Optional[httpx.Timeout] = None, **kwargs) -> str: View Source

112    async def post(
113        self,
114        url: str,
115        options: Optional[Dict] = None,
116        addons: Optional[Dict] = None,
117        headers: Optional[Dict] = None,
118        timeout: Optional[httpx.Timeout] = None,
119        **kwargs,
120    ) -> str:
121        """
122        Make a POST request to the specified URL.
123        """
124        return await self._handle_request(
125            "POST",
126            url,
127            params=options,
128            addons=addons,
129            headers=headers,
130            timeout=timeout,
131            **kwargs,
132        )

Make a POST request to the specified URL.

async def put( self, url: str, options: Optional[Dict] = None, addons: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: Optional[httpx.Timeout] = None, **kwargs) -> str: View Source

134    async def put(
135        self,
136        url: str,
137        options: Optional[Dict] = None,
138        addons: Optional[Dict] = None,
139        headers: Optional[Dict] = None,
140        timeout: Optional[httpx.Timeout] = None,
141        **kwargs,
142    ) -> str:
143        """
144        Make a PUT request to the specified URL.
145        """
146        return await self._handle_request(
147            "PUT",
148            url,
149            params=options,
150            addons=addons,
151            headers=headers,
152            timeout=timeout,
153            **kwargs,
154        )

Make a PUT request to the specified URL.

async def patch( self, url: str, options: Optional[Dict] = None, addons: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: Optional[httpx.Timeout] = None, **kwargs) -> str: View Source

156    async def patch(
157        self,
158        url: str,
159        options: Optional[Dict] = None,
160        addons: Optional[Dict] = None,
161        headers: Optional[Dict] = None,
162        timeout: Optional[httpx.Timeout] = None,
163        **kwargs,
164    ) -> str:
165        """
166        Make a PATCH request to the specified URL.
167        """
168        return await self._handle_request(
169            "PATCH",
170            url,
171            params=options,
172            addons=addons,
173            headers=headers,
174            timeout=timeout,
175            **kwargs,
176        )

Make a PATCH request to the specified URL.

async def delete( self, url: str, options: Optional[Dict] = None, addons: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: Optional[httpx.Timeout] = None, **kwargs) -> str: View Source

178    async def delete(
179        self,
180        url: str,
181        options: Optional[Dict] = None,
182        addons: Optional[Dict] = None,
183        headers: Optional[Dict] = None,
184        timeout: Optional[httpx.Timeout] = None,
185        **kwargs,
186    ) -> str:
187        """
188        Make a DELETE request to the specified URL.
189        """
190        return await self._handle_request(
191            "DELETE",
192            url,
193            params=options,
194            addons=addons,
195            headers=headers,
196            timeout=timeout,
197            **kwargs,
198        )

Make a DELETE request to the specified URL.