24
24
from __future__ import annotations
25
25
26
26
import json
27
- from typing import TYPE_CHECKING , Awaitable , Optional , Union
27
+ from typing import TYPE_CHECKING , Coroutine , Optional , Union
28
28
29
29
import aiohttp
30
30
@@ -45,41 +45,37 @@ class HTTPClient:
45
45
46
46
Attributes
47
47
----------
48
- api_key: Optional[:class:`str`]
49
- Your private API token to access the Mystb.in API.
50
- Currently unobtainable.
51
48
session: Optional[Union[:class:`aiohttp.ClientSession`, :class:`requests.Session`]]
52
49
Optional session to be passed to the creation of the client.
53
50
"""
54
51
55
- __slots__ = ("api_key" , "session" , "_are_we_async" )
52
+ __slots__ = (
53
+ "session" ,
54
+ "_are_we_async" ,
55
+ )
56
56
57
57
def __init__ (
58
58
self ,
59
59
* ,
60
- api_key : str = "" ,
61
60
session : Optional [Union [aiohttp .ClientSession , requests .Session ]] = None ,
62
61
) -> None :
63
- self .api_key = api_key
64
62
self ._are_we_async = session is None or isinstance (session , aiohttp .ClientSession )
65
- self .session = session
63
+ self .session = session # type: ignore
66
64
67
- async def _generate_async_session (self ) -> aiohttp . ClientSession :
65
+ async def _generate_async_session (self ) -> None :
68
66
"""
69
67
This method will create a new and blank `aiohttp.ClientSession` instance for use.
70
68
This method should not be called if a session is passed to the constructor.
71
69
"""
72
- self .session = aiohttp .ClientSession ()
70
+ self .session : aiohttp . ClientSession = aiohttp .ClientSession ()
73
71
74
- return self .session
75
-
76
- def post (self , content : str , syntax : str = None ) -> Union [Paste , Awaitable ]:
72
+ def post (self , content : str , syntax : str = None ) -> Union [Paste , Coroutine [None , None , Paste ]]:
77
73
"""
78
74
This will post to the Mystb.in API and return the url.
79
75
Can pass an optional suffix for the syntax highlighting.
80
76
81
77
Parameters
82
- ----------
78
+ -----------
83
79
content: :class:`str`
84
80
The content you are posting to the Mystb.in API.
85
81
syntax: :class:`str`
@@ -90,16 +86,19 @@ def post(self, content: str, syntax: str = None) -> Union[Paste, Awaitable]:
90
86
return self ._perform_sync_post (content , syntax )
91
87
92
88
def _perform_sync_post (self , content : str , syntax : str = None ) -> Paste :
93
- """ Sync post request. """
89
+ assert not isinstance (self .session , aiohttp .ClientSession )
90
+
94
91
payload = {"meta" : [{"index" : 0 , "syntax" : syntax }]}
92
+
93
+ assert self .session is not None and not isinstance (self .session , aiohttp .ClientSession )
94
+
95
95
response : requests .Response = self .session .post (
96
96
API_BASE_URL ,
97
97
files = {
98
98
"data" : content ,
99
99
"meta" : (None , json .dumps (payload ), "application/json" ),
100
100
},
101
101
timeout = CLIENT_TIMEOUT ,
102
- headers = {"Authorization" : self .api_key },
103
102
)
104
103
105
104
if response .status_code not in [200 , 201 ]:
@@ -108,9 +107,10 @@ def _perform_sync_post(self, content: str, syntax: str = None) -> Paste:
108
107
return Paste (response .json (), syntax )
109
108
110
109
async def _perform_async_post (self , content : str , syntax : str = None ) -> Paste :
111
- """ Async post request. """
112
110
if not self .session and self ._are_we_async :
113
- self .session = await self ._generate_async_session ()
111
+ await self ._generate_async_session ()
112
+
113
+ assert self .session is not None and isinstance (self .session , aiohttp .ClientSession )
114
114
115
115
multi_part_write = aiohttp .MultipartWriter ()
116
116
paste_content = multi_part_write .append (content )
@@ -122,17 +122,18 @@ async def _perform_async_post(self, content: str, syntax: str = None) -> Paste:
122
122
API_BASE_URL ,
123
123
data = multi_part_write ,
124
124
timeout = aiohttp .ClientTimeout (CLIENT_TIMEOUT ),
125
- headers = {"Authorization" : self .api_key },
126
125
) as response :
127
126
status_code = response .status
128
127
response_text = await response .text ()
129
- if status_code not in (200 ,):
128
+
129
+ if status_code != 200 :
130
130
raise APIError (status_code , response_text )
131
+
131
132
response_data = await response .json ()
132
133
133
134
return Paste (response_data , syntax )
134
135
135
- def get (self , paste_id : str ) -> Union [PasteData , Awaitable ]:
136
+ def get (self , paste_id : str ) -> Union [PasteData , Coroutine [ None , None , PasteData ] ]:
136
137
"""
137
138
This will perform a GET request against the Mystb.in API and return the url.
138
139
Must be passed a valid paste ID or URL.
@@ -155,19 +156,20 @@ def get(self, paste_id: str) -> Union[PasteData, Awaitable]:
155
156
return self ._perform_async_get (paste_id )
156
157
157
158
def _perform_sync_get (self , paste_id : str ) -> PasteData :
158
- """ Sync get request. """
159
- response : requests .Response = self .session .get (f"{ API_BASE_URL } /{ paste_id } " , timeout = CLIENT_TIMEOUT )
159
+ assert self .session is not None and not isinstance (self .session , aiohttp .ClientSession )
160
160
161
- if response .status_code not in (200 ,):
162
- raise BadPasteID ("This is an invalid Mystb.in paste ID." )
161
+ with self .session .get (f"{ API_BASE_URL } /{ paste_id } " , timeout = CLIENT_TIMEOUT ) as response :
162
+ if response .status_code not in (200 ,):
163
+ raise BadPasteID ("This is an invalid Mystb.in paste ID." )
163
164
164
165
paste_data = response .json ()
165
166
return PasteData (paste_id , paste_data )
166
167
167
168
async def _perform_async_get (self , paste_id : str ) -> PasteData :
168
- """ Async get request. """
169
169
if not self .session :
170
- self .session : aiohttp .ClientSession = await self ._generate_async_session ()
170
+ await self ._generate_async_session ()
171
+
172
+ assert self .session is not None and isinstance (self .session , aiohttp .ClientSession )
171
173
172
174
async with self .session .get (f"{ API_BASE_URL } /{ paste_id } " , timeout = aiohttp .ClientTimeout (CLIENT_TIMEOUT )) as response :
173
175
if response .status not in (200 ,):
@@ -176,7 +178,7 @@ async def _perform_async_get(self, paste_id: str) -> PasteData:
176
178
177
179
return PasteData (paste_id , paste_data )
178
180
179
- async def close (self ):
180
- """ Async only - close the session. """
181
+ async def close (self ) -> None :
182
+ """Async only - close the session."""
181
183
if self .session and isinstance (self .session , aiohttp .ClientSession ):
182
184
await self .session .close ()
0 commit comments