]> jfr.im git - yt-dlp.git/blob - yt_dlp/networking/exceptions.py
[cleanup] Add more ruff rules (#10149)
[yt-dlp.git] / yt_dlp / networking / exceptions.py
1 from __future__ import annotations
2
3 import typing
4
5 from ..utils import YoutubeDLError
6
7 if typing.TYPE_CHECKING:
8 from .common import RequestHandler, Response
9
10
11 class RequestError(YoutubeDLError):
12 def __init__(
13 self,
14 msg: str | None = None,
15 cause: Exception | str | None = None,
16 handler: RequestHandler = None,
17 ):
18 self.handler = handler
19 self.cause = cause
20 if not msg and cause:
21 msg = str(cause)
22 super().__init__(msg)
23
24
25 class UnsupportedRequest(RequestError):
26 """raised when a handler cannot handle a request"""
27 pass
28
29
30 class NoSupportingHandlers(RequestError):
31 """raised when no handlers can support a request for various reasons"""
32
33 def __init__(self, unsupported_errors: list[UnsupportedRequest], unexpected_errors: list[Exception]):
34 self.unsupported_errors = unsupported_errors or []
35 self.unexpected_errors = unexpected_errors or []
36
37 # Print a quick summary of the errors
38 err_handler_map = {}
39 for err in unsupported_errors:
40 err_handler_map.setdefault(err.msg, []).append(err.handler.RH_NAME)
41
42 reason_str = ', '.join([f'{msg} ({", ".join(handlers)})' for msg, handlers in err_handler_map.items()])
43 if unexpected_errors:
44 reason_str = ' + '.join(filter(None, [reason_str, f'{len(unexpected_errors)} unexpected error(s)']))
45
46 err_str = 'Unable to handle request'
47 if reason_str:
48 err_str += f': {reason_str}'
49
50 super().__init__(msg=err_str)
51
52
53 class TransportError(RequestError):
54 """Network related errors"""
55
56
57 class HTTPError(RequestError):
58 def __init__(self, response: Response, redirect_loop=False):
59 self.response = response
60 self.status = response.status
61 self.reason = response.reason
62 self.redirect_loop = redirect_loop
63 msg = f'HTTP Error {response.status}: {response.reason}'
64 if redirect_loop:
65 msg += ' (redirect loop detected)'
66
67 super().__init__(msg=msg)
68
69 def close(self):
70 self.response.close()
71
72 def __repr__(self):
73 return f'<HTTPError {self.status}: {self.reason}>'
74
75
76 class IncompleteRead(TransportError):
77 def __init__(self, partial: int, expected: int | None = None, **kwargs):
78 self.partial = partial
79 self.expected = expected
80 msg = f'{partial} bytes read'
81 if expected is not None:
82 msg += f', {expected} more expected'
83
84 super().__init__(msg=msg, **kwargs)
85
86 def __repr__(self):
87 return f'<IncompleteRead: {self.msg}>'
88
89
90 class SSLError(TransportError):
91 pass
92
93
94 class CertificateVerifyError(SSLError):
95 """Raised when certificate validated has failed"""
96 pass
97
98
99 class ProxyError(TransportError):
100 pass
101
102
103 network_exceptions = (HTTPError, TransportError)