22
22
DEALINGS IN THE SOFTWARE.
23
23
"""
24
24
25
- import asyncio
26
25
import json
27
- from typing import TYPE_CHECKING , Awaitable , Callable , Optional , Type , Union
26
+ from typing import TYPE_CHECKING , Awaitable , Optional , Type , Union
28
27
29
28
import aiohttp
30
29
31
30
if TYPE_CHECKING :
32
31
import requests
33
32
34
- from .constants import *
35
- from .errors import *
36
- from .objects import *
33
+ from .constants import API_BASE_URL , CLIENT_TIMEOUT , MB_URL_RE
34
+ from .errors import APIError , BadPasteID
35
+ from .objects import Paste , PasteData
37
36
38
- __all__ = ("Client" , )
37
+ __all__ = ("HTTPClient" , )
39
38
40
39
41
- class Client :
40
+ class HTTPClient :
42
41
"""
43
42
Client for interacting with the Mystb.in API.
44
43
45
44
Attributes
46
45
----------
47
46
api_key: Optional[:class:`str`]
48
47
Your private API token to access the Mystb.in API.
49
- Can be obtained via: #TODO
48
+ Currently unobtainable.
50
49
session: Optional[Union[:class:`aiohttp.ClientSession`, :class:`requests.Session`]]
51
50
Optional session to be passed to the creation of the client.
52
51
"""
52
+
53
53
__slots__ = ("api_key" , "session" , "_are_we_async" )
54
54
55
55
def __init__ (
56
- self , * ,
56
+ self ,
57
+ * ,
57
58
api_key : str = None ,
58
- session : Optional [Union [aiohttp .ClientSession ,
59
- Type ["requests.Session" ]]] = None
59
+ session : Optional [Union [aiohttp .ClientSession , Type ["requests.Session" ]]] = None ,
60
60
) -> None :
61
61
self .api_key = api_key
62
62
self ._are_we_async = session is None or isinstance (
63
- session , aiohttp .ClientSession )
64
- self .session = self ._generate_sync_session (
65
- session ) if not self ._are_we_async else None
63
+ session , aiohttp .ClientSession
64
+ )
65
+ self .session = (
66
+ self ._generate_sync_session (session ) if not self ._are_we_async else None
67
+ )
66
68
67
- def _generate_sync_session (self , session : Type ["requests.Session" ]) -> Type ["requests.Session" ]:
69
+ def _generate_sync_session (
70
+ self , session : Type ["requests.Session" ]
71
+ ) -> Type ["requests.Session" ]:
68
72
""" We will update a :class:`requests.Session` instance with the auth we require. """
69
73
# the passed session was found to be 'sync'.
70
74
if self .api_key :
71
75
session .headers .update (
72
- {"Authorization" : self .api_key , "User-Agent" : f"Mystbin.py" })
76
+ {"Authorization" : self .api_key , "User-Agent" : "Mystbin.py" }
77
+ )
73
78
74
79
return session
75
80
76
- async def _generate_async_session (self , session : Optional [aiohttp .ClientSession ] = None ) -> aiohttp .ClientSession :
81
+ async def _generate_async_session (
82
+ self , session : Optional [aiohttp .ClientSession ] = None
83
+ ) -> aiohttp .ClientSession :
77
84
""" We will update (or create) a :class:`aiohttp.ClientSession` instance with the auth we require. """
78
85
if not session :
79
86
session = aiohttp .ClientSession (raise_for_status = False )
80
87
81
88
if self .api_key :
82
89
session ._default_headers .update (
83
- {"Authorization" : self .api_key , "User-Agent" : f"Mystbin.py" })
90
+ {"Authorization" : self .api_key , "User-Agent" : "Mystbin.py" }
91
+ )
84
92
85
93
session ._timeout = aiohttp .ClientTimeout (CLIENT_TIMEOUT )
86
94
return session
@@ -103,9 +111,15 @@ def post(self, content: str, syntax: str = None) -> Union[Paste, Awaitable]:
103
111
104
112
def _perform_sync_post (self , content : str , syntax : str = None ) -> Paste :
105
113
""" Sync post request. """
106
- payload = {'meta' : [{'index' : 0 , 'syntax' : syntax }]}
107
- response : Type ["requests.Response" ] = self .session .post (API_BASE_URL , files = {
108
- 'data' : content , 'meta' : (None , json .dumps (payload ), 'application/json' )}, timeout = CLIENT_TIMEOUT )
114
+ payload = {"meta" : [{"index" : 0 , "syntax" : syntax }]}
115
+ response : Type ["requests.Response" ] = self .session .post (
116
+ API_BASE_URL ,
117
+ files = {
118
+ "data" : content ,
119
+ "meta" : (None , json .dumps (payload ), "application/json" ),
120
+ },
121
+ timeout = CLIENT_TIMEOUT ,
122
+ )
109
123
110
124
if response .status_code not in [200 , 201 ]:
111
125
raise APIError (response .status_code , response .text )
@@ -121,14 +135,14 @@ async def _perform_async_post(self, content: str, syntax: str = None) -> Paste:
121
135
paste_content = multi_part_write .append (content )
122
136
paste_content .set_content_disposition ("form-data" , name = "data" )
123
137
paste_content = multi_part_write .append_json (
124
- {' meta' : [{' index' : 0 , ' syntax' : syntax }]}
138
+ {" meta" : [{" index" : 0 , " syntax" : syntax }]}
125
139
)
126
140
paste_content .set_content_disposition ("form-data" , name = "meta" )
127
141
128
142
async with self .session .post (API_BASE_URL , data = multi_part_write ) as response :
129
143
status_code = response .status
130
144
response_text = await response .text ()
131
- if status_code not in (200 , ):
145
+ if status_code not in (200 ,):
132
146
raise APIError (status_code , response_text )
133
147
response_data = await response .json ()
134
148
@@ -149,8 +163,8 @@ def get(self, paste_id: str) -> Union[PasteData, Awaitable]:
149
163
if not paste_id_match :
150
164
raise BadPasteID ("This is an invalid Mystb.in paste ID." )
151
165
152
- paste_id = paste_id_match .group ('ID' )
153
- syntax = paste_id_match .group (' syntax' )
166
+ paste_id = paste_id_match .group ("ID" )
167
+ syntax = paste_id_match .group (" syntax" )
154
168
155
169
if not self ._are_we_async :
156
170
return self ._perform_sync_get (paste_id , syntax )
@@ -160,9 +174,10 @@ def get(self, paste_id: str) -> Union[PasteData, Awaitable]:
160
174
def _perform_sync_get (self , paste_id : str , syntax : str = None ) -> PasteData :
161
175
""" Sync get request. """
162
176
response : Type ["requests.Response" ] = self .session .get (
163
- f"{ API_BASE_URL } /{ paste_id } " , timeout = CLIENT_TIMEOUT )
177
+ f"{ API_BASE_URL } /{ paste_id } " , timeout = CLIENT_TIMEOUT
178
+ )
164
179
165
- if response .status_code not in (200 , ):
180
+ if response .status_code not in (200 ,):
166
181
raise BadPasteID ("This is an invalid Mystb.in paste ID." )
167
182
168
183
paste_data = response .json ()
@@ -173,8 +188,12 @@ async def _perform_async_get(self, paste_id: str, syntax: str = None) -> PasteDa
173
188
if not self .session :
174
189
self .session : aiohttp .ClientSession = await self ._generate_async_session ()
175
190
176
- async with self .session .get (f"{ API_BASE_URL } /{ paste_id } " , timeout = aiohttp .ClientTimeout (CLIENT_TIMEOUT )) as response :
177
- if response .status not in (200 , ):
191
+ self .session : aiohttp .ClientSession
192
+
193
+ async with self .session .get (
194
+ f"{ API_BASE_URL } /{ paste_id } " , timeout = aiohttp .ClientTimeout (CLIENT_TIMEOUT )
195
+ ) as response :
196
+ if response .status not in (200 ,):
178
197
raise BadPasteID ("This is an invalid Mystb.in paste ID." )
179
198
paste_data = await response .json ()
180
199
0 commit comments