def _get_response_from_http()

in torchdata/datapipes/iter/load/online.py [0:0]


def _get_response_from_http(url: str, *, timeout: Optional[float]) -> Tuple[str, StreamWrapper]:
    try:
        with requests.Session() as session:
            if timeout is None:
                r = session.get(url, stream=True)
            else:
                r = session.get(url, timeout=timeout, stream=True)
        return url, StreamWrapper(r.raw)
    except HTTPError as e:
        raise Exception(f"Could not get the file. [HTTP Error] {e.response}.")
    except RequestException as e:
        raise Exception(f"Could not get the file at {url}. [RequestException] {e.response}.")
    except Exception:
        raise