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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
|
from __future__ import annotations
import typing
import urllib.error
from ..utils import YoutubeDLError, deprecation_warning
if typing.TYPE_CHECKING:
from .common import RequestHandler, Response
class RequestError(YoutubeDLError):
def __init__(
self,
msg: str | None = None,
cause: Exception | str | None = None,
handler: RequestHandler = None
):
self.handler = handler
self.cause = cause
if not msg and cause:
msg = str(cause)
super().__init__(msg)
class UnsupportedRequest(RequestError):
"""raised when a handler cannot handle a request"""
pass
class NoSupportingHandlers(RequestError):
"""raised when no handlers can support a request for various reasons"""
def __init__(self, unsupported_errors: list[UnsupportedRequest], unexpected_errors: list[Exception]):
self.unsupported_errors = unsupported_errors or []
self.unexpected_errors = unexpected_errors or []
# Print a quick summary of the errors
err_handler_map = {}
for err in unsupported_errors:
err_handler_map.setdefault(err.msg, []).append(err.handler.RH_NAME)
reason_str = ', '.join([f'{msg} ({", ".join(handlers)})' for msg, handlers in err_handler_map.items()])
if unexpected_errors:
reason_str = ' + '.join(filter(None, [reason_str, f'{len(unexpected_errors)} unexpected error(s)']))
err_str = 'Unable to handle request'
if reason_str:
err_str += f': {reason_str}'
super().__init__(msg=err_str)
class TransportError(RequestError):
"""Network related errors"""
class HTTPError(RequestError):
def __init__(self, response: Response, redirect_loop=False):
self.response = response
self.status = response.status
self.reason = response.reason
self.redirect_loop = redirect_loop
msg = f'HTTP Error {response.status}: {response.reason}'
if redirect_loop:
msg += ' (redirect loop detected)'
super().__init__(msg=msg)
def close(self):
self.response.close()
def __repr__(self):
return f'<HTTPError {self.status}: {self.reason}>'
class IncompleteRead(TransportError):
def __init__(self, partial, expected=None, **kwargs):
self.partial = partial
self.expected = expected
msg = f'{len(partial)} bytes read'
if expected is not None:
msg += f', {expected} more expected'
super().__init__(msg=msg, **kwargs)
def __repr__(self):
return f'<IncompleteRead: {self.msg}>'
class SSLError(TransportError):
pass
class CertificateVerifyError(SSLError):
"""Raised when certificate validated has failed"""
pass
class ProxyError(TransportError):
pass
class _CompatHTTPError(urllib.error.HTTPError, HTTPError):
"""
Provides backwards compatibility with urllib.error.HTTPError.
Do not use this class directly, use HTTPError instead.
"""
def __init__(self, http_error: HTTPError):
super().__init__(
url=http_error.response.url,
code=http_error.status,
msg=http_error.msg,
hdrs=http_error.response.headers,
fp=http_error.response
)
self._closer.file = None # Disable auto close
self._http_error = http_error
HTTPError.__init__(self, http_error.response, redirect_loop=http_error.redirect_loop)
@property
def status(self):
return self._http_error.status
@status.setter
def status(self, value):
return
@property
def reason(self):
return self._http_error.reason
@reason.setter
def reason(self, value):
return
@property
def headers(self):
deprecation_warning('HTTPError.headers is deprecated, use HTTPError.response.headers instead')
return self._http_error.response.headers
@headers.setter
def headers(self, value):
return
def info(self):
deprecation_warning('HTTPError.info() is deprecated, use HTTPError.response.headers instead')
return self.response.headers
def getcode(self):
deprecation_warning('HTTPError.getcode is deprecated, use HTTPError.status instead')
return self.status
def geturl(self):
deprecation_warning('HTTPError.geturl is deprecated, use HTTPError.response.url instead')
return self.response.url
@property
def code(self):
deprecation_warning('HTTPError.code is deprecated, use HTTPError.status instead')
return self.status
@code.setter
def code(self, value):
return
@property
def url(self):
deprecation_warning('HTTPError.url is deprecated, use HTTPError.response.url instead')
return self.response.url
@url.setter
def url(self, value):
return
@property
def hdrs(self):
deprecation_warning('HTTPError.hdrs is deprecated, use HTTPError.response.headers instead')
return self.response.headers
@hdrs.setter
def hdrs(self, value):
return
@property
def filename(self):
deprecation_warning('HTTPError.filename is deprecated, use HTTPError.response.url instead')
return self.response.url
@filename.setter
def filename(self, value):
return
def __getattr__(self, name):
# File operations are passed through the response.
# Warn for some commonly used ones
passthrough_warnings = {
'read': 'response.read()',
# technically possibly due to passthrough, but we should discourage this
'get_header': 'response.get_header()',
'readable': 'response.readable()',
'closed': 'response.closed',
'tell': 'response.tell()',
}
if name in passthrough_warnings:
deprecation_warning(f'HTTPError.{name} is deprecated, use HTTPError.{passthrough_warnings[name]} instead')
return super().__getattr__(name)
def __str__(self):
return str(self._http_error)
def __repr__(self):
return repr(self._http_error)
network_exceptions = (HTTPError, TransportError)
|
