-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathhttp.py
More file actions
122 lines (87 loc) · 3.41 KB
/
http.py
File metadata and controls
122 lines (87 loc) · 3.41 KB
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
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
from abc import ABC, abstractmethod
class HttpResponse(ABC):
@property
@abstractmethod
def status(self) -> int:
pass
@abstractmethod
def raise_for_status(self):
pass
@abstractmethod
async def text(self) -> str:
pass
class HttpClient(ABC):
@abstractmethod
async def get(self, url) -> HttpResponse:
pass
try:
import aiohttp
class AioHttpResponse(HttpResponse):
response: aiohttp.ClientResponse
def __init__(self, response: aiohttp.ClientResponse):
self.response = response
async def __aenter__(self):
return await self.response.__aenter__()
async def __aexit__(self, exc_type, exc_val, exc_tb):
return await self.response.__aexit__(exc_type, exc_val, exc_tb)
@property
def status(self) -> int:
return self.response.status
def raise_for_status(self):
return self.response.raise_for_status()
async def text(self) -> str:
return await self.response.text()
class AioHttpClient(HttpClient):
session: aiohttp.ClientSession
def __init__(self, session: aiohttp.ClientSession):
assert isinstance(session, aiohttp.ClientSession)
self.session = session
async def __aenter__(self):
return await self.session.__aenter__()
async def __aexit__(self, exc_type, exc_val, exc_tb):
return await self.session.__aexit__(exc_type, exc_val, exc_tb)
async def get(self, url) -> HttpResponse:
response = await self.session.get(url)
return AioHttpResponse(response)
except ImportError:
AioHttpClient = None
try:
import httpx
class HttpxResponse(HttpResponse):
response: httpx.Response
def __init__(self, response: httpx.Response):
self.response = response
async def __aenter__(self):
return self.response
async def __aexit__(self, exc_type, exc_val, exc_tb):
pass
@property
def status(self) -> int:
return self.response.status_code
def raise_for_status(self):
return self.response.raise_for_status()
async def text(self) -> str:
return self.response.text
class HttpxClient(HttpClient):
client: httpx.AsyncClient
def __init__(self, client: httpx.AsyncClient, auth: httpx.DigestAuth | httpx.UseClientDefault):
assert isinstance(client, httpx.AsyncClient)
self.client = client
self.auth = auth
async def __aenter__(self):
return await self.client.__aenter__()
async def __aexit__(self, exc_type, exc_val, exc_tb):
return await self.client.__aexit__(exc_type, exc_val, exc_tb)
async def get(self, url) -> HttpResponse:
response = await self.client.get(url, auth=self.auth)
return HttpxResponse(response)
except ImportError:
HttpxClient = None
def create_aiohttp_client(client) -> HttpClient:
assert AioHttpClient is not None and isinstance(client, aiohttp.ClientSession)
return AioHttpClient(client)
def create_httpx_client(client, auth=None) -> HttpClient:
assert HttpxClient is not None and isinstance(client, httpx.AsyncClient)
if auth is None:
auth = httpx.USE_CLIENT_DEFAULT
return HttpxClient(client, auth)