Function fetch_file

src/sentry/http.py:150–243

Pull down a URL, returning a UrlResult object.

(
    url: str,
    domain_lock_enabled: bool = True,
    headers: dict[str, str] | None = None,
    allow_redirects: bool = True,
    verify_ssl: bool = False,
    timeout: float = settings.SENTRY_SOURCE_FETCH_SOCKET_TIMEOUT,
    **kwargs: Any,
)

def fetch_file(
    url: str,
    domain_lock_enabled: bool = True,
    headers: dict[str, str] | None = None,
    allow_redirects: bool = True,
    verify_ssl: bool = False,
    timeout: float = settings.SENTRY_SOURCE_FETCH_SOCKET_TIMEOUT,
    **kwargs: Any,
) -> UrlResult:
    """
    Pull down a URL, returning a UrlResult object.
    """
    # lock down domains that are problematic
    if domain_lock_enabled:
        domain_key = get_domain_key(url)
        domain_result = cache.get(domain_key)
        if domain_result:
            domain_result["url"] = url
            raise CannotFetch(domain_result)

    logger.debug("Fetching %r from the internet", url)

    with contextlib.ExitStack() as ctx:
        http_session = ctx.enter_context(SafeSession())

        try:
            start = time.monotonic()
            response = ctx.enter_context(
                http_session.get(
                    url,
                    allow_redirects=allow_redirects,
                    verify=verify_ssl,
                    headers=headers,
                    timeout=timeout,
                    stream=True,
                    **kwargs,
                )
            )

            try:
                cl = int(response.headers["content-length"])
            except (LookupError, ValueError):
                cl = 0
            if cl > settings.SENTRY_SOURCE_FETCH_MAX_SIZE:
                raise OverflowError()

            outfile = BytesIO()

            cl = 0

            # Only need to even attempt to read the response body if we
            # got a 200 OK
            if response.status_code == 200:
                for chunk in response.iter_content(16 * 1024):
                    if time.monotonic() - start > settings.SENTRY_SOURCE_FETCH_TIMEOUT:
                        raise Timeout()
                    outfile.write(chunk)
                    cl += len(chunk)