]> jfr.im git - yt-dlp.git/blob - yt_dlp/networking/exceptions.py
[cleanup] Misc (#8182)
[yt-dlp.git] / yt_dlp / networking / exceptions.py
1 from __future__ import annotations
2
3 import typing
4 import urllib.error
5
6 from ..utils import YoutubeDLError, deprecation_warning
7
8 if typing.TYPE_CHECKING:
9 from .common import RequestHandler, Response
10
11
12 class RequestError(YoutubeDLError):
13 def __init__(
14 self,
15 msg: str | None = None,
16 cause: Exception | str | None = None,
17 handler: RequestHandler = None
18 ):
19 self.handler = handler
20 self.cause = cause
21 if not msg and cause:
22 msg = str(cause)
23 super().__init__(msg)
24
25
26 class UnsupportedRequest(RequestError):
27 """raised when a handler cannot handle a request"""
28 pass
29
30
31 class NoSupportingHandlers(RequestError):
32 """raised when no handlers can support a request for various reasons"""
33
34 def __init__(self, unsupported_errors: list[UnsupportedRequest], unexpected_errors: list[Exception]):
35 self.unsupported_errors = unsupported_errors or []
36 self.unexpected_errors = unexpected_errors or []
37
38 # Print a quick summary of the errors
39 err_handler_map = {}
40 for err in unsupported_errors:
41 err_handler_map.setdefault(err.msg, []).append(err.handler.RH_NAME)
42
43 reason_str = ', '.join([f'{msg} ({", ".join(handlers)})' for msg, handlers in err_handler_map.items()])
44 if unexpected_errors:
45 reason_str = ' + '.join(filter(None, [reason_str, f'{len(unexpected_errors)} unexpected error(s)']))
46
47 err_str = 'Unable to handle request'
48 if reason_str:
49 err_str += f': {reason_str}'
50
51 super().__init__(msg=err_str)
52
53
54 class TransportError(RequestError):
55 """Network related errors"""
56
57
58 class HTTPError(RequestError):
59 def __init__(self, response: Response, redirect_loop=False):
60 self.response = response
61 self.status = response.status
62 self.reason = response.reason
63 self.redirect_loop = redirect_loop
64 msg = f'HTTP Error {response.status}: {response.reason}'
65 if redirect_loop:
66 msg += ' (redirect loop detected)'
67
68 super().__init__(msg=msg)
69
70 def close(self):
71 self.response.close()
72
73 def __repr__(self):
74 return f'<HTTPError {self.status}: {self.reason}>'
75
76
77 class IncompleteRead(TransportError):
78 def __init__(self, partial: int, expected: int = None, **kwargs):
79 self.partial = partial
80 self.expected = expected
81 msg = f'{partial} bytes read'
82 if expected is not None:
83 msg += f', {expected} more expected'
84
85 super().__init__(msg=msg, **kwargs)
86
87 def __repr__(self):
88 return f'<IncompleteRead: {self.msg}>'
89
90
91 class SSLError(TransportError):
92 pass
93
94
95 class CertificateVerifyError(SSLError):
96 """Raised when certificate validated has failed"""
97 pass
98
99
100 class ProxyError(TransportError):
101 pass
102
103
104 class _CompatHTTPError(urllib.error.HTTPError, HTTPError):
105 """
106 Provides backwards compatibility with urllib.error.HTTPError.
107 Do not use this class directly, use HTTPError instead.
108 """
109
110 def __init__(self, http_error: HTTPError):
111 super().__init__(
112 url=http_error.response.url,
113 code=http_error.status,
114 msg=http_error.msg,
115 hdrs=http_error.response.headers,
116 fp=http_error.response
117 )
118 self._closer.close_called = True # Disable auto close
119 self._http_error = http_error
120 HTTPError.__init__(self, http_error.response, redirect_loop=http_error.redirect_loop)
121
122 @property
123 def status(self):
124 return self._http_error.status
125
126 @status.setter
127 def status(self, value):
128 return
129
130 @property
131 def reason(self):
132 return self._http_error.reason
133
134 @reason.setter
135 def reason(self, value):
136 return
137
138 @property
139 def headers(self):
140 deprecation_warning('HTTPError.headers is deprecated, use HTTPError.response.headers instead')
141 return self._http_error.response.headers
142
143 @headers.setter
144 def headers(self, value):
145 return
146
147 def info(self):
148 deprecation_warning('HTTPError.info() is deprecated, use HTTPError.response.headers instead')
149 return self.response.headers
150
151 def getcode(self):
152 deprecation_warning('HTTPError.getcode is deprecated, use HTTPError.status instead')
153 return self.status
154
155 def geturl(self):
156 deprecation_warning('HTTPError.geturl is deprecated, use HTTPError.response.url instead')
157 return self.response.url
158
159 @property
160 def code(self):
161 deprecation_warning('HTTPError.code is deprecated, use HTTPError.status instead')
162 return self.status
163
164 @code.setter
165 def code(self, value):
166 return
167
168 @property
169 def url(self):
170 deprecation_warning('HTTPError.url is deprecated, use HTTPError.response.url instead')
171 return self.response.url
172
173 @url.setter
174 def url(self, value):
175 return
176
177 @property
178 def hdrs(self):
179 deprecation_warning('HTTPError.hdrs is deprecated, use HTTPError.response.headers instead')
180 return self.response.headers
181
182 @hdrs.setter
183 def hdrs(self, value):
184 return
185
186 @property
187 def filename(self):
188 deprecation_warning('HTTPError.filename is deprecated, use HTTPError.response.url instead')
189 return self.response.url
190
191 @filename.setter
192 def filename(self, value):
193 return
194
195 def __getattr__(self, name):
196 # File operations are passed through the response.
197 # Warn for some commonly used ones
198 passthrough_warnings = {
199 'read': 'response.read()',
200 # technically possibly due to passthrough, but we should discourage this
201 'get_header': 'response.get_header()',
202 'readable': 'response.readable()',
203 'closed': 'response.closed',
204 'tell': 'response.tell()',
205 }
206 if name in passthrough_warnings:
207 deprecation_warning(f'HTTPError.{name} is deprecated, use HTTPError.{passthrough_warnings[name]} instead')
208 return super().__getattr__(name)
209
210 def __str__(self):
211 return str(self._http_error)
212
213 def __repr__(self):
214 return repr(self._http_error)
215
216
217 network_exceptions = (HTTPError, TransportError)