1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
|
import abc
import sys
from typing import Any, Callable, Dict, Iterator, Optional, Union
import requests
from requests_toolbelt.multipart.encoder import MultipartEncoder # type: ignore
if sys.version_info >= (3, 8):
from typing import Protocol
else:
from typing_extensions import Protocol
class BackendResponse(Protocol):
@abc.abstractmethod
def __init__(self, response: requests.Response) -> None:
...
class Backend(Protocol):
@staticmethod
@abc.abstractmethod
def response_content(
response: requests.Response,
streamed: bool,
action: Optional[Callable[[bytes], None]],
chunk_size: int,
*,
iterator: bool,
) -> Optional[Union[bytes, Iterator[Any]]]:
...
@abc.abstractmethod
def http_request(
self,
method: str,
url: str,
json: Optional[Union[Dict[str, Any], bytes]],
data: Optional[Union[Dict[str, Any], MultipartEncoder]],
params: Optional[Any],
timeout: Optional[float],
verify: Optional[Union[bool, str]],
stream: Optional[bool],
**kwargs: Any,
) -> BackendResponse:
...
|
