()
| 94 | |
| 95 | @contextlib.contextmanager |
| 96 | def map_httpcore_exceptions() -> typing.Iterator[None]: |
| 97 | global HTTPCORE_EXC_MAP |
| 98 | if len(HTTPCORE_EXC_MAP) == 0: |
| 99 | HTTPCORE_EXC_MAP = _load_httpcore_exceptions() |
| 100 | try: |
| 101 | yield |
| 102 | except Exception as exc: |
| 103 | mapped_exc = None |
| 104 | |
| 105 | for from_exc, to_exc in HTTPCORE_EXC_MAP.items(): |
| 106 | if not isinstance(exc, from_exc): |
| 107 | continue |
| 108 | # We want to map to the most specific exception we can find. |
| 109 | # Eg if `exc` is an `httpcore.ReadTimeout`, we want to map to |
| 110 | # `httpx.ReadTimeout`, not just `httpx.TimeoutException`. |
| 111 | if mapped_exc is None or issubclass(to_exc, mapped_exc): |
| 112 | mapped_exc = to_exc |
| 113 | |
| 114 | if mapped_exc is None: # pragma: no cover |
| 115 | raise |
| 116 | |
| 117 | message = str(exc) |
| 118 | raise mapped_exc(message) from exc |
| 119 | |
| 120 | |
| 121 | class ResponseStream(SyncByteStream): |
no test coverage detected